taskqueue

Nov 10, 2023

Changes in this version

+ type Description struct

+ Change []string

+ Entity string

+ Filter string

+ type FunctionProcessor struct

+ F func(task Task[schemaless.Record[T]])

+ func (proc *FunctionProcessor[T]) Process(task Task[schemaless.Record[schema.Schema]]) error

+ type Processor interface

+ Process func(task Task[T]) error

+ type Queue struct

+ func NewInMemoryQueue[T any]() *Queue[T]

+ func (*Queue[T]) Delete(ctx context.Context, tasks []Task[schemaless.Record[schema.Schema]]) error

+ func (q *Queue[T]) Close() error

+ func (q *Queue[T]) Pop(ctx context.Context) ([]Task[T], error)

+ func (q *Queue[T]) Push(ctx context.Context, task Task[T]) error

+ type Queuer interface

+ Delete func(ctx context.Context, tasks []Task[schemaless.Record[schema.Schema]]) error

+ Pop func(ctx context.Context) ([]Task[T], error)

+ Push func(ctx context.Context, task Task[T]) error

+ type Repository interface

+ FindingRecords func(query schemaless.FindingRecords[schemaless.Record[schema.Schema]]) (schemaless.PageResult[schemaless.Record[schema.Schema]], error)

+ type SQSQueue struct

+ func NewSQSQueue(c *sqs.Client, queueURL string) *SQSQueue[schemaless.Record[schema.Schema]]

+ func (queue *SQSQueue[T]) Delete(ctx context.Context, tasks []Task[schemaless.Record[schema.Schema]]) error

+ func (queue *SQSQueue[T]) Pop(ctx context.Context) ([]Task[T], error)

+ func (queue *SQSQueue[T]) Push(ctx context.Context, task Task[T]) error

+ type Task struct

+ Data T

+ ID string

+ Meta map[string]string

+ type TaskQueue struct

+ func NewTaskQueue(desc *Description, queue Queuer[schemaless.Record[schema.Schema]], ...) *TaskQueue

+ func (q *TaskQueue) RunCDC(ctx context.Context) error

+ func (q *TaskQueue) RunProcessor(ctx context.Context) error

+ func (q *TaskQueue) RunSelector(ctx context.Context) error