tommyqueue

README

tommyqueue

Documentation

Index

• Variables
• type ErrorHandler
• type ErrorHandlerFunc
  • func (fn ErrorHandlerFunc) HandleError(ctx context.Context, task *Task, err error)
• type HandlerFunc
  • func (fn HandlerFunc) ProcessTask(ctx context.Context, task *Task) error
• type Logger
• type Payload
• type ProcessHandler
• type RetryDelayFunc
• type Server
  • func NewServer(ops ...ServerOption) *Server
• type ServerOption
  • func WithBroker(b broker.CtxBroker) ServerOption
  • func WithLogger(l Logger) ServerOption
• type Task
  • func NewTask(typename string, payload map[string]interface{}) *Task
• type Worker

Variables

var (
	// ProcessHandler 的返回值，表示当前 task 需要 retry or archive
	ErrSkipRetry = errors.New("shoule retry later")
)

Types

type ErrorHandler

type ErrorHandler interface {
	HandleError(ctx context.Context, task *Task, err error)
}

An ErrorHandler handles an error occured during task processing.

type ErrorHandlerFunc

type ErrorHandlerFunc func(ctx context.Context, task *Task, err error)

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

func (ErrorHandlerFunc) HandleError

func (fn ErrorHandlerFunc) HandleError(ctx context.Context, task *Task, err error)

HandleError calls fn(ctx, task, err)

type HandlerFunc

type HandlerFunc func(context.Context, *Task) error

The HandlerFunc type 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.

func (HandlerFunc) ProcessTask

func (fn HandlerFunc) ProcessTask(ctx context.Context, task *Task) error

ProcessTask calls fn(ctx, task)

type Logger

type Logger interface {
	// Debug logs a message at Debug level.
	Debug(args ...interface{})

	// Info logs a message at Info level.
	Info(args ...interface{})

	Infof(format string, args ...interface{})

	// Warn logs a message at Warning level.
	Warn(args ...interface{})

	Warnf(format string, args ...interface{})

	// Error logs a message at Error level.
	Error(args ...interface{})

	Errorf(format string, args ...interface{})
}

type Payload

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

Payload holds arbitrary data needed for task execution.

type ProcessHandler

type ProcessHandler interface {
	ProcessTask(context.Context, *Task) error
}

A ProcessHandler processes tasks.

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

If ProcessTask return a non-nil error or panics, the task will be retried after delay. One exception to this rule is when ProcessTask returns SkipRetry error. If the returned error is SkipRetry or the error wraps SkipRetry, retry is skipped and task will be archived instead.

type RetryDelayFunc

type RetryDelayFunc func(n int, e error, t *Task) time.Duration

RetryDelayFunc calculates the retry delay duration for a failed task given the retry count, error, and the task.

n is the number of times the task canhas been retry. e is the error returned by the task handler. t is the task in question.

type Server

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

func NewServer

func NewServer(ops ...ServerOption) *Server

type ServerOption

type ServerOption func(s *Server)

func WithBroker

func WithBroker(b broker.CtxBroker) ServerOption

func WithLogger

func WithLogger(l Logger) ServerOption

type Task

type Task struct {
	// Type indicates the type of task to be performed.
	Type string

	// Payload holds data needed to perform the task.
	Payload Payload
}

Task represents a unit of work to be performed.

func NewTask

func NewTask(typename string, payload map[string]interface{}) *Task

NewTask returns a new Task given a type name and payload data.

The payload values must be serializable.

type Worker

type Worker interface {
	Start(ctx context.Context, wg *sync.WaitGroup)
	Stop(ctx context.Context)
}