worker

Documentation

Index

• type Engine
  • func NewEngine(config *support.Config, l *support.Logger) *Engine
  • func (w *Engine) Drain()
  • func (w *Engine) Enqueue(task *Task, opts *TaskOptions) (*TaskResult, error)
  • func (w *Engine) Info() []string
  • func (w *Engine) ProcessTask(ctx context.Context, task *Task) error
  • func (w *Engine) Run() error
  • func (w *Engine) Tasks() []*Task
  • func (w *Engine) Use(handlers ...MiddlewareFunc)
• type Handler
• type HandlerFunc
• type MiddlewareFunc
• type Task
  • func NewJob(id string, data map[string]interface{}) *Task
• type TaskOptions
• type TaskPayload
• type TaskResult

Types

type Engine

type Engine struct {
	*asynq.Server
	*asynq.Client
	*asynq.ServeMux
	asynq.RedisConnOpt
	// contains filtered or unexported fields
}

Engine processes the background tasks.

func NewEngine

func NewEngine(config *support.Config, l *support.Logger) *Engine

NewEngine initializes a worker to process background tasks.

func (*Engine) Drain

func (w *Engine) Drain()

Drain simulates task processing by removing them, only available for unit test with APPY_ENV=test.

func (*Engine) Enqueue

func (w *Engine) Enqueue(task *Task, opts *TaskOptions) (*TaskResult, error)

Enqueue enqueues task to be processed asynchronously.

Enqueue returns nil if the task is enqueued successfully, otherwise returns a non-nil error.

The argument opts specifies the behavior of task processing. If there are conflicting TaskOption values the last one overrides others. By default, max retry is set to 25 and timeout is set to 30 minutes. If no ProcessAt or ProcessIn options are passed, the task will be processed immediately.

func (*Engine) Info

func (w *Engine) Info() []string

Info returns the worker info.

func (*Engine) ProcessTask

func (w *Engine) ProcessTask(ctx context.Context, task *Task) error

ProcessTask dispatches the task to the handler whose pattern most closely matches the task type.

func (*Engine) Run

func (w *Engine) Run() error

Run starts running the worker to process background tasks.

func (*Engine) Tasks

func (w *Engine) Tasks() []*Task

Tasks returns the enqueued tasks, only available for unit test with APPY_ENV=test.

func (*Engine) Use

func (w *Engine) Use(handlers ...MiddlewareFunc)

Use appends a MiddlewareFunc to the chain. Middlewares are executed in the order that they are applied to the ServeMux.

type Handler

type Handler = asynq.Handler

Handler processes background tasks.

ProcessTask should return nil if the processing of a background task is successful.

If ProcessTask return a non-nil error or panics, the background task will be retried after delay.

type HandlerFunc

type HandlerFunc = asynq.HandlerFunc

HandlerFunc is an adapter to allow the use of ordinary functions as a Handler. If f is a function with the appropriate signature, HandlerFunc(f) is a Handler that calls f.

type MiddlewareFunc

type MiddlewareFunc = asynq.MiddlewareFunc

MiddlewareFunc is a function which receives an Handler and returns another Handler. Typically, the returned handler is a closure which does something with the context and task passed to it, and then calls the handler passed as parameter to the MiddlewareFunc.

type Task

type Task = asynq.Task

Task represents a unit of work to be performed.

func NewJob

func NewJob(id string, data map[string]interface{}) *Task

NewJob initializes a job with a unique identifier and its data for background job processing.

type TaskOptions

type TaskOptions struct {
	Deadline  time.Time
	MaxRetry  int
	ProcessAt time.Time
	ProcessIn time.Duration
	Queue     string
	Timeout   time.Duration
	UniqueTTL time.Duration
}

TaskOptions specifies how a task should be processed.

type TaskPayload

type TaskPayload = asynq.Payload

TaskPayload holds arbitrary data needed for task processing.

type TaskResult

type TaskResult = asynq.Result

TaskResult holds enqueued task's metadata.