redis

Documentation

Index

• Constants
• func NewBackend(logger zerolog.Logger, cfg config.Config) (iface.Backend, error)
• type Backend
  • func (b *Backend) AddTask(ctx context.Context, queueName string, spec task.TaskSpec) (*task.Task, error)
  • func (b *Backend) CreateQueue(ctx context.Context, queueSpec taskqueue.TaskQueueSpec) (*taskqueue.TaskQueue, error)
  • func (b *Backend) DeleteQueue(ctx context.Context, queueName string) error
  • func (b *Backend) GetAllQueues(ctx context.Context) ([]*taskqueue.TaskQueue, error)
  • func (b *Backend) GetAllTasks(ctx context.Context, queueName string) ([]*task.Task, error)
  • func (b *Backend) GetAllWorkers(ctx context.Context, queueUID uuid.UUID) ([]*apiworker.Worker, error)
  • func (b *Backend) GetCompletedTasks(ctx context.Context, queueName string) ([]*task.Task, error)
  • func (b *Backend) GetDeadLetter(ctx context.Context, queueName string) ([]taskqueue.TaskToDeadletterError, error)
  • func (b *Backend) GetFailedTasks(ctx context.Context, queueName string) ([]*task.Task, error)
  • func (b *Backend) GetFailedWorkers(ctx context.Context, queueUID uuid.UUID) ([]*apiworker.Worker, error)
  • func (b *Backend) GetLostWorker(ctx context.Context, queueUID uuid.UUID) ([]*apiworker.Worker, error)
  • func (b *Backend) GetPendingTasks(ctx context.Context, queueName string) ([]*task.Task, error)
  • func (b *Backend) GetProcessingTasks(ctx context.Context, queueName string) ([]*task.Task, error)
  • func (b *Backend) GetQueueByName(ctx context.Context, queueName string) (*taskqueue.TaskQueue, error)
  • func (b *Backend) GetReceivedTasks(ctx context.Context, queueName string) ([]*task.Task, error)
  • func (b *Backend) GetRunningWorkers(ctx context.Context, queueUID uuid.UUID) ([]*apiworker.Worker, error)
  • func (b *Backend) GetSucceededTasks(ctx context.Context, queueName string) ([]*task.Task, error)
  • func (b *Backend) GetSucceededWorkers(ctx context.Context, queueUID uuid.UUID) ([]*apiworker.Worker, error)
  • func (b *Backend) GetWorker(ctx context.Context, queueUID, workerUID uuid.UUID) (*apiworker.Worker, error)
  • func (b *Backend) GetWorkersToSalvage(ctx context.Context, queueUID uuid.UUID) ([]*apiworker.Worker, error)
  • func (b *Backend) NextTask(ctx context.Context, queueUID, workerUID uuid.UUID) (*task.Task, error)
  • func (b *Backend) RecordFailure(ctx context.Context, queueUID, workerUID uuid.UUID, t *task.Task, ...) error
  • func (b *Backend) RegisterWorker(ctx context.Context, queueUID uuid.UUID, workerSpec apiworker.WorkerSpec) (*apiworker.Worker, error)
  • func (b *Backend) SalvageWorker(ctx context.Context, ...) (*apiworker.Worker, []*task.Task, error)
  • func (b *Backend) SendWorkerHeartBeat(ctx context.Context, queueUID, workerUID uuid.UUID) (*apiworker.Worker, error)
  • func (b *Backend) SetProcessing(ctx context.Context, queueUID, workerUID uuid.UUID, t *task.Task) error
  • func (b *Backend) SetSucceeded(ctx context.Context, queueUID, workerUID uuid.UUID, t *task.Task, ...) error
  • func (b *Backend) SetWorkerFailed(ctx context.Context, queueUID, workerUID uuid.UUID) (*apiworker.Worker, error)
  • func (b *Backend) SetWorkerSucceeded(ctx context.Context, queueUID, workerUID uuid.UUID) (*apiworker.Worker, error)
  • func (b *Backend) UpdateQueue(ctx context.Context, queueSpec taskqueue.TaskQueueSpec) (*taskqueue.TaskQueue, error)

Constants

const (
	KB                 = 1 << 10
	PayloadMaxSizeInKB = 1
	MessageMaxSizeInKB = 1
	HistoryLengthMax   = 10
	MaxNameLength      = 1024
)

const (
	GlobalKeyPrefix = "_pftaskqueue:"
)

Functions

func NewBackend

func NewBackend(logger zerolog.Logger, cfg config.Config) (iface.Backend, error)

Types

type Backend

type Backend struct {
	*common.Backend
	*config.RedisConfig
}

func (*Backend) AddTask

func (b *Backend) AddTask(ctx context.Context, queueName string, spec task.TaskSpec) (*task.Task, error)

func (*Backend) CreateQueue

func (b *Backend) CreateQueue(ctx context.Context, queueSpec taskqueue.TaskQueueSpec) (*taskqueue.TaskQueue, error)

func (*Backend) DeleteQueue

func (b *Backend) DeleteQueue(ctx context.Context, queueName string) error

func (*Backend) GetAllQueues

func (b *Backend) GetAllQueues(ctx context.Context) ([]*taskqueue.TaskQueue, error)

func (*Backend) GetAllTasks

func (b *Backend) GetAllTasks(ctx context.Context, queueName string) ([]*task.Task, error)

func (*Backend) GetAllWorkers

func (b *Backend) GetAllWorkers(ctx context.Context, queueUID uuid.UUID) ([]*apiworker.Worker, error)

func (*Backend) GetCompletedTasks

func (b *Backend) GetCompletedTasks(ctx context.Context, queueName string) ([]*task.Task, error)

func (*Backend) GetDeadLetter

func (b *Backend) GetDeadLetter(ctx context.Context, queueName string) ([]taskqueue.TaskToDeadletterError, error)

func (*Backend) GetFailedTasks

func (b *Backend) GetFailedTasks(ctx context.Context, queueName string) ([]*task.Task, error)

func (*Backend) GetFailedWorkers

func (b *Backend) GetFailedWorkers(ctx context.Context, queueUID uuid.UUID) ([]*apiworker.Worker, error)

func (*Backend) GetLostWorker

func (b *Backend) GetLostWorker(ctx context.Context, queueUID uuid.UUID) ([]*apiworker.Worker, error)

func (*Backend) GetPendingTasks

func (b *Backend) GetPendingTasks(ctx context.Context, queueName string) ([]*task.Task, error)

func (*Backend) GetProcessingTasks

func (b *Backend) GetProcessingTasks(ctx context.Context, queueName string) ([]*task.Task, error)

func (*Backend) GetQueueByName

func (b *Backend) GetQueueByName(ctx context.Context, queueName string) (*taskqueue.TaskQueue, error)

func (*Backend) GetReceivedTasks

func (b *Backend) GetReceivedTasks(ctx context.Context, queueName string) ([]*task.Task, error)

func (*Backend) GetRunningWorkers

func (b *Backend) GetRunningWorkers(ctx context.Context, queueUID uuid.UUID) ([]*apiworker.Worker, error)

func (*Backend) GetSucceededTasks

func (b *Backend) GetSucceededTasks(ctx context.Context, queueName string) ([]*task.Task, error)

func (*Backend) GetSucceededWorkers

func (b *Backend) GetSucceededWorkers(ctx context.Context, queueUID uuid.UUID) ([]*apiworker.Worker, error)

func (*Backend) GetWorker

func (b *Backend) GetWorker(ctx context.Context, queueUID, workerUID uuid.UUID) (*apiworker.Worker, error)

func (*Backend) GetWorkersToSalvage

func (b *Backend) GetWorkersToSalvage(ctx context.Context, queueUID uuid.UUID) ([]*apiworker.Worker, error)

func (*Backend) NextTask

func (b *Backend) NextTask(ctx context.Context, queueUID, workerUID uuid.UUID) (*task.Task, error)

func (*Backend) RecordFailure

func (b *Backend) RecordFailure(ctx context.Context, queueUID, workerUID uuid.UUID, t *task.Task, resultPayload *string, message *string, reason task.TaskResultReason, onFailureSpecs []task.TaskSpec) error

func (*Backend) RegisterWorker

func (b *Backend) RegisterWorker(ctx context.Context, queueUID uuid.UUID, workerSpec apiworker.WorkerSpec) (*apiworker.Worker, error)

func (*Backend) SalvageWorker

func (b *Backend) SalvageWorker(ctx context.Context, queueUID, salvagingWorkerUID, salvageTargetWorkerUID uuid.UUID) (*apiworker.Worker, []*task.Task, error)

func (*Backend) SendWorkerHeartBeat

func (b *Backend) SendWorkerHeartBeat(ctx context.Context, queueUID, workerUID uuid.UUID) (*apiworker.Worker, error)

func (*Backend) SetProcessing

func (b *Backend) SetProcessing(ctx context.Context, queueUID, workerUID uuid.UUID, t *task.Task) error

func (*Backend) SetSucceeded

func (b *Backend) SetSucceeded(ctx context.Context, queueUID, workerUID uuid.UUID, t *task.Task, resultPayload *string, message *string, onSuccessSpecs []task.TaskSpec) error

func (*Backend) SetWorkerFailed

func (b *Backend) SetWorkerFailed(ctx context.Context, queueUID, workerUID uuid.UUID) (*apiworker.Worker, error)

func (*Backend) SetWorkerSucceeded

func (b *Backend) SetWorkerSucceeded(ctx context.Context, queueUID, workerUID uuid.UUID) (*apiworker.Worker, error)

func (*Backend) UpdateQueue

func (b *Backend) UpdateQueue(ctx context.Context, queueSpec taskqueue.TaskQueueSpec) (*taskqueue.TaskQueue, error)