Documentation
¶
Overview ¶
Package queue runs background jobs: producers add jobs to a named queue, and processors handle them in workers with retries. A Driver stores the jobs; NewMemory keeps them in the process (lost on restart, for development and tests), and the separate module github.com/dygoframework/dygo/queue/asynq keeps them in Redis.
A provider with a Process method is a processor:
type WelcomeEmail struct{ UserID string }
func (m *Mailer) Process(r *queue.Registry) {
queue.Handle(r, "welcome-email", m.sendWelcome) // func(ctx, WelcomeEmail) error
}
and producers inject *queue.Queue:
_, err := s.queue.Add(ctx, "welcome-email", WelcomeEmail{UserID: id}, queue.Delay(time.Minute))
Payloads are JSON. A handler error retries the job, up to its attempts (3 by default), with the driver's backoff. dygo generate job writes a processor.
Index ¶
- Constants
- Variables
- func Handle[T any](r *Registry, name string, fn func(ctx context.Context, payload T) error, ...)
- func Module(o Options) *dygo.Module
- func SkipRetry(err error) error
- func WithJob(ctx context.Context, j *Job) context.Context
- type AddOption
- type Driver
- type FailedJob
- type HandleOption
- type Job
- type Memory
- type MemoryOptions
- type Options
- type Processor
- type Queue
- func (q *Queue) Add(ctx context.Context, name string, payload any, opts ...AddOption) (string, error)
- func (q *Queue) BeforeApplicationShutdown(ctx context.Context, _ string) error
- func (q *Queue) Driver() Driver
- func (q *Queue) OnApplicationBootstrap(context.Context) error
- func (q *Queue) OnModuleInit(context.Context) error
- func (q *Queue) Register(p Processor) error
- func (q *Queue) Start() error
- func (q *Queue) Stop(ctx context.Context) error
- type Registry
Constants ¶
const DefaultQueue = "default"
DefaultQueue is the queue of jobs added without On.
Variables ¶
var ErrDuplicate = errors.New("queue: duplicate job id")
ErrDuplicate is returned when a job ID is already queued.
var ErrSkipRetry = errors.New("queue: skip retry")
ErrSkipRetry marks an error that must not be retried; see SkipRetry.
Functions ¶
func Handle ¶
func Handle[T any](r *Registry, name string, fn func(ctx context.Context, payload T) error, opts ...HandleOption)
Handle registers fn for jobs named name, on DefaultQueue unless the handler options say otherwise. The payload is decoded into a T; a payload that does not decode fails without a retry.
Types ¶
type AddOption ¶
type AddOption func(*Job)
AddOption configures one job.
type Driver ¶
type Driver interface {
// Enqueue stores j. It sets j.ID when empty; a driver may reject an ID
// that is already queued with [ErrDuplicate].
Enqueue(ctx context.Context, j *Job) error
// Work delivers jobs of the given queues to handle, at most concurrency
// at a time, until ctx is cancelled; then it waits for running handlers
// and returns. When handle returns an error the driver retries the job
// until MaxAttempts, then keeps it as failed. An error wrapped with
// [SkipRetry] fails the job at once.
Work(ctx context.Context, queues []string, concurrency int, handle func(ctx context.Context, j *Job) error) error
}
Driver stores and delivers jobs.
type HandleOption ¶
type HandleOption func(*handler)
HandleOption configures a handler.
func FromQueue ¶
func FromQueue(name string) HandleOption
FromQueue makes the handler consume queue name.
type Job ¶
type Job struct {
ID string
Queue string
Name string
Payload []byte
// Attempt is 1 on the first run.
Attempt int
MaxAttempts int
// RunAt is when the job may run first; zero means now.
RunAt time.Time
}
Job is one unit of work, as a driver stores it.
type Memory ¶
type Memory struct {
// contains filtered or unexported fields
}
Memory is a Driver that keeps jobs in the process. Jobs are lost when the process stops.
func (*Memory) Pending ¶
Pending returns the number of jobs queued, waiting for a retry or delay, or running.
type MemoryOptions ¶
type MemoryOptions struct {
// Backoff is the wait before attempt n+1 after attempt n failed. The
// default doubles from one second up to one minute.
Backoff func(attempt int) time.Duration
// KeepFailed bounds how many failed jobs Failed returns (100 by default).
KeepFailed int
}
MemoryOptions configures NewMemory.
type Options ¶
type Options struct {
// Driver stores the jobs. The default is NewMemory(MemoryOptions{}).
Driver Driver
// Concurrency is how many jobs run at once (10 by default).
Concurrency int
// Attempts is the default number of attempts (3 by default).
Attempts int
// NoWorker only produces jobs; another process handles them.
NoWorker bool
// OnError gets every failed attempt; the default ignores them.
OnError func(ctx context.Context, j *Job, err error)
}
Options configures Module.
type Processor ¶
type Processor interface {
Process(r *Registry)
}
Processor is a provider that handles jobs. Process is called once, during OnModuleInit of the queue.
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
Queue adds jobs and, unless Options.NoWorker is set, runs a worker for the registered handlers from OnApplicationBootstrap until BeforeApplicationShutdown, which waits for running jobs.
func (*Queue) Add ¶
func (q *Queue) Add(ctx context.Context, name string, payload any, opts ...AddOption) (string, error)
Add enqueues a job named name with payload encoded as JSON and returns its ID.
func (*Queue) BeforeApplicationShutdown ¶
BeforeApplicationShutdown stops the worker and waits for running jobs.
func (*Queue) OnApplicationBootstrap ¶
OnApplicationBootstrap starts the worker.
func (*Queue) OnModuleInit ¶
OnModuleInit registers every Processor.
Source Files