memdb

Documentation

Overview

Package memdb implements the storage engine interface for MemDB.

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) 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) 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)
• type StateFieldIndex
  • func (i *StateFieldIndex) FromArgs(args ...any) ([]byte, error)
  • func (i *StateFieldIndex) FromObject(obj any) (bool, []byte, error)
• type TimeFieldIndex
  • func (i *TimeFieldIndex) FromArgs(args ...any) ([]byte, error)
  • func (i *TimeFieldIndex) FromObject(obj any) (bool, []byte, error)

Types

type Config

type Config struct {
	SnapshotPath     string        `arg:"--memdb-snapshot-path,env:MEMDB_SNAPSHOT_PATH" placeholder:"PATH" help:"path to the snapshot file" default:""`
	SnapshotInterval time.Duration `` /* 141-byte string literal not displayed */

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

Config contains configurations for the MemDB storage engine.

type Engine

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

Engine implements the storage engine interface for MemDB.

func New

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

New creates a new MemDB 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) 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) 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.

type StateFieldIndex

type StateFieldIndex struct {
	Field string

	// Filter down to a particular state for building partial indexes.
	Filter ratus.TaskState
}

StateFieldIndex encodes task state fields for index building.

func (*StateFieldIndex) FromArgs

func (i *StateFieldIndex) FromArgs(args ...any) ([]byte, error)

FromArgs implements the memdb.Indexer interface.

func (*StateFieldIndex) FromObject

func (i *StateFieldIndex) FromObject(obj any) (bool, []byte, error)

FromObject implements the memdb.SingleIndexer interface.

type TimeFieldIndex

type TimeFieldIndex struct {
	Field string
}

TimeFieldIndex encodes time fields for index building.

func (*TimeFieldIndex) FromArgs

func (i *TimeFieldIndex) FromArgs(args ...any) ([]byte, error)

FromArgs implements the memdb.Indexer interface.

func (*TimeFieldIndex) FromObject

func (i *TimeFieldIndex) FromObject(obj any) (bool, []byte, error)

FromObject implements the memdb.SingleIndexer interface.