taskqueue

README

x/taskqueue - Golang task queue with transactional support

Documentation

Index

• type Description
• type FunctionProcessor
  • func (proc *FunctionProcessor[T]) Process(task Task[schemaless.Record[schema.Schema]]) error
• type Processor
• type Queue
  • func NewInMemoryQueue[T any]() *Queue[T]
  • func (q *Queue[T]) Close() error
  • func (*Queue[T]) Delete(ctx context.Context, tasks []Task[schemaless.Record[schema.Schema]]) 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
• type Repository
• type SQSQueue
  • 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
• type TaskQueue
  • 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

Types

type Description

type Description struct {
	Change []string
	Entity string
	Filter string
}

type FunctionProcessor

type FunctionProcessor[T any] struct {
	F func(task Task[schemaless.Record[T]])
}

func (*FunctionProcessor[T]) Process

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

type Processor

type Processor[T any] interface {
	Process(task Task[T]) error
}

type Queue

type Queue[T any] struct {
	// contains filtered or unexported fields
}

func NewInMemoryQueue

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

func (*Queue[T]) Close

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

func (*Queue[T]) Delete

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

func (*Queue[T]) Pop

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

func (*Queue[T]) Push

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

type Queuer

type Queuer[T any] interface {
	Push(ctx context.Context, task Task[T]) error
	Pop(ctx context.Context) ([]Task[T], error)
	Delete(ctx context.Context, tasks []Task[schemaless.Record[schema.Schema]]) error
}

type Repository

type Repository interface {
	FindingRecords(query schemaless.FindingRecords[schemaless.Record[schema.Schema]]) (schemaless.PageResult[schemaless.Record[schema.Schema]], error)
}

type SQSQueue

type SQSQueue[T any] struct {
	// contains filtered or unexported fields
}

SQSQueue is a queue that uses AWS SQS as a backend.

func NewSQSQueue

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

func (*SQSQueue[T]) Delete

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

func (*SQSQueue[T]) Pop

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

func (*SQSQueue[T]) Push

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

type Task

type Task[T any] struct {
	ID   string
	Data T
	Meta map[string]string
}

type TaskQueue

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

func NewTaskQueue

func NewTaskQueue(
	desc *Description,
	queue Queuer[schemaless.Record[schema.Schema]],
	find Repository,
	stream *schemaless.AppendLog[schema.Schema],
	proc Processor[schemaless.Record[schema.Schema]],
) *TaskQueue

func (*TaskQueue) RunCDC

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

func (*TaskQueue) RunProcessor

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

func (*TaskQueue) RunSelector

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