mongodb

Documentation

Overview

Package mongodb implements the storage engine interface for MongoDB.

Index

• type Config
• type Engine
  • func New(c *Config) (*Engine, error)
  • func (g *Engine) Chore(ctx context.Context) error
  • func (g *Engine) Close(ctx context.Context) error
  • func (g *Engine) Collection() *mongo.Collection
  • func (g *Engine) Commit(ctx context.Context, id string, m *ratus.Commit) (*ratus.Task, error)
  • func (g *Engine) DeletePromise(ctx context.Context, id string) (*ratus.Deleted, error)
  • func (g *Engine) DeletePromises(ctx context.Context, topic string) (*ratus.Deleted, error)
  • func (g *Engine) DeleteTask(ctx context.Context, id string) (*ratus.Deleted, error)
  • func (g *Engine) DeleteTasks(ctx context.Context, topic string) (*ratus.Deleted, error)
  • func (g *Engine) DeleteTopic(ctx context.Context, topic string) (*ratus.Deleted, error)
  • func (g *Engine) DeleteTopics(ctx context.Context) (*ratus.Deleted, error)
  • func (g *Engine) Destroy(ctx context.Context) error
  • func (g *Engine) Fallback(v int32) *Engine
  • func (g *Engine) GetPromise(ctx context.Context, id string) (*ratus.Promise, error)
  • func (g *Engine) GetTask(ctx context.Context, id string) (*ratus.Task, error)
  • func (g *Engine) GetTopic(ctx context.Context, topic string) (*ratus.Topic, error)
  • func (g *Engine) InsertPromise(ctx context.Context, p *ratus.Promise) (*ratus.Task, error)
  • func (g *Engine) InsertTask(ctx context.Context, t *ratus.Task) (*ratus.Updated, error)
  • func (g *Engine) InsertTasks(ctx context.Context, ts []*ratus.Task) (*ratus.Updated, error)
  • func (g *Engine) ListPromises(ctx context.Context, topic string, limit, offset int) ([]*ratus.Promise, error)
  • func (g *Engine) ListTasks(ctx context.Context, topic string, limit, offset int) ([]*ratus.Task, error)
  • func (g *Engine) ListTopics(ctx context.Context, limit, offset int) ([]*ratus.Topic, error)
  • func (g *Engine) Open(ctx context.Context) error
  • func (g *Engine) Poll(ctx context.Context, topic string, p *ratus.Promise) (*ratus.Task, error)
  • func (g *Engine) Ready(ctx context.Context) error
  • func (g *Engine) UpsertPromise(ctx context.Context, p *ratus.Promise) (*ratus.Task, error)
  • func (g *Engine) UpsertTask(ctx context.Context, t *ratus.Task) (*ratus.Updated, error)
  • func (g *Engine) UpsertTasks(ctx context.Context, ts []*ratus.Task) (*ratus.Updated, error)

Types

type Config

type Config struct {
	URI        string `` /* 151-byte string literal not displayed */
	Database   string `arg:"--mongodb-database,env:MONGODB_DATABASE" placeholder:"NAME" help:"name of the MongoDB database to use" default:"ratus"`
	Collection string `` /* 137-byte string literal not displayed */

	RetentionPeriod time.Duration `` /* 142-byte string literal not displayed */

	DisableIndexCreation bool `arg:"--mongodb-disable-index-creation,env:MONGODB_DISABLE_INDEX_CREATION" help:"disable automatic index creation on startup"`
	DisableAutoFallback  bool `` /* 135-byte string literal not displayed */
	DisableAtomicPoll    bool `` /* 132-byte string literal not displayed */
}

Config contains configurations for the MongoDB storage engine.

type Engine

type Engine struct {
	// contains filtered or unexported fields
}

Engine implements the storage engine interface for MongoDB.

func New

func New(c *Config) (*Engine, error)

New creates a new MongoDB storage engine instance.

func (*Engine) Chore

func (g *Engine) Chore(ctx context.Context) error

Chore recovers timed out tasks and deletes expired tasks.

func (*Engine) Close

func (g *Engine) Close(ctx context.Context) error

Close or disconnect from the storage engine.

func (*Engine) Collection

func (g *Engine) Collection() *mongo.Collection

Collection returns the handle for the task collection.

func (*Engine) Commit

func (g *Engine) Commit(ctx context.Context, id string, m *ratus.Commit) (*ratus.Task, error)

Commit applies a set of updates to a task and returns the updated task.

func (*Engine) DeletePromise

func (g *Engine) DeletePromise(ctx context.Context, id string) (*ratus.Deleted, error)

DeletePromise deletes a promise by the unique ID of its target task.

func (*Engine) DeletePromises

func (g *Engine) DeletePromises(ctx context.Context, topic string) (*ratus.Deleted, error)

DeletePromises deletes all promises in a topic.

func (*Engine) DeleteTask

func (g *Engine) DeleteTask(ctx context.Context, id string) (*ratus.Deleted, error)

DeleteTask deletes a task by its unique ID.

func (*Engine) DeleteTasks

func (g *Engine) DeleteTasks(ctx context.Context, topic string) (*ratus.Deleted, error)

DeleteTasks deletes all tasks in a topic.

func (*Engine) DeleteTopic

func (g *Engine) DeleteTopic(ctx context.Context, topic string) (*ratus.Deleted, error)

DeleteTopic deletes a topic and its tasks.

func (*Engine) DeleteTopics

func (g *Engine) DeleteTopics(ctx context.Context) (*ratus.Deleted, error)

DeleteTopics deletes all topics and tasks.

func (*Engine) Destroy

func (g *Engine) Destroy(ctx context.Context) error

Destroy clears all data and closes the storage engine.

func (*Engine) Fallback

func (g *Engine) Fallback(v int32) *Engine

Fallback sets all fallback flags to the given value.

func (*Engine) GetPromise

func (g *Engine) GetPromise(ctx context.Context, id string) (*ratus.Promise, error)

GetPromise gets a promise by the unique ID of its target task.

func (*Engine) GetTask

func (g *Engine) GetTask(ctx context.Context, id string) (*ratus.Task, error)

GetTask gets a task by its unique ID.

func (*Engine) GetTopic

func (g *Engine) GetTopic(ctx context.Context, topic string) (*ratus.Topic, error)

GetTopic gets information about a topic.

func (*Engine) InsertPromise

func (g *Engine) InsertPromise(ctx context.Context, p *ratus.Promise) (*ratus.Task, error)

InsertPromise makes a promise to claim and execute a task if it is in pending state.

func (*Engine) InsertTask

func (g *Engine) InsertTask(ctx context.Context, t *ratus.Task) (*ratus.Updated, error)

InsertTask inserts a new task.

func (*Engine) InsertTasks

func (g *Engine) InsertTasks(ctx context.Context, ts []*ratus.Task) (*ratus.Updated, error)

InsertTasks inserts a batch of tasks while ignoring existing ones.

func (*Engine) ListPromises

func (g *Engine) ListPromises(ctx context.Context, topic string, limit, offset int) ([]*ratus.Promise, error)

ListPromises lists all promises in a topic.

func (*Engine) ListTasks

func (g *Engine) ListTasks(ctx context.Context, topic string, limit, offset int) ([]*ratus.Task, error)

ListTasks lists all tasks in a topic.

func (*Engine) ListTopics

func (g *Engine) ListTopics(ctx context.Context, limit, offset int) ([]*ratus.Topic, error)

ListTopics lists all topics.

func (*Engine) Open

func (g *Engine) Open(ctx context.Context) error

Open or connect to the storage engine.

func (*Engine) Poll

func (g *Engine) Poll(ctx context.Context, topic string, p *ratus.Promise) (*ratus.Task, error)

Poll makes a promise to claim and execute the next available task in a topic.

func (*Engine) Ready

func (g *Engine) Ready(ctx context.Context) error

Ready probes the storage engine and returns an error if it is not ready.

func (*Engine) UpsertPromise

func (g *Engine) UpsertPromise(ctx context.Context, p *ratus.Promise) (*ratus.Task, error)

UpsertPromise makes a promise to claim and execute a task regardless of its current state.

func (*Engine) UpsertTask

func (g *Engine) UpsertTask(ctx context.Context, t *ratus.Task) (*ratus.Updated, error)

UpsertTask inserts or updates a task.

func (*Engine) UpsertTasks

func (g *Engine) UpsertTasks(ctx context.Context, ts []*ratus.Task) (*ratus.Updated, error)

UpsertTasks inserts or updates a batch of tasks.