Documentation
¶
Overview ¶
Package queue provides a lightweight job queue with in-memory and Redis drivers. Jobs are dispatched, stored in the driver, and processed by workers. Dead-letter handling and exponential back-off retries are built in.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func CanonicalName ¶
CanonicalName returns the canonical class name for a job value. If the job implements ClassNamer, that override wins. Otherwise it is the bare struct type name with any pointer and package qualifier stripped (e.g. *main.WelcomeEmailJob → "WelcomeEmailJob"). Dispatch stamps this name on the payload, so Register must be called with the same name.
queue.Register(queue.CanonicalName(&WelcomeEmailJob{}), func() queue.Job { return &WelcomeEmailJob{} })
func Register ¶
Register maps a class name to a factory for that job type. Call this once per job type, typically in an init() function. The class name must match what Dispatch stamps — the bare struct name, e.g. "WelcomeEmailJob" (see CanonicalName), or the job's Class() value if it implements ClassNamer. A raw package-qualified name will never match a dispatched job and every such job will dead-letter on Unmarshal.
queue.Register("WelcomeEmailJob", func() queue.Job { return &WelcomeEmailJob{} })
Types ¶
type ClassNamer ¶
type ClassNamer interface {
// Class returns the class name to use for this job type.
Class() string
}
ClassNamer is an optional interface a Job may implement to override the class name stamped on its payloads at dispatch time. Register the job under the same custom name.
type DispatchOpts ¶
type DispatchOpts struct {
MaxAttempts int
}
DispatchOpts customises job dispatch.
type Driver ¶
type Driver interface {
// Push enqueues a payload on the named queue.
Push(ctx context.Context, queue string, p *Payload) error
// Pop reserves and returns the next ready payload, or nil if empty/not ready.
Pop(ctx context.Context, queue string) (*Payload, error)
// Ack marks a previously popped payload as successfully processed and
// drops it from in-flight tracking.
Ack(ctx context.Context, p *Payload) error
// Dead moves a failed payload to the dead-letter queue (clearing its reservation).
Dead(ctx context.Context, p *Payload) error
// Release re-queues a payload with an updated AvailableAt for retry
// (clearing its reservation).
Release(ctx context.Context, p *Payload, delay time.Duration) error
}
Driver is the backend for storing and retrieving queue payloads.
Delivery contract: Pop reserves a payload rather than discarding it. The driver keeps tracking a popped payload until exactly one of Ack, Release, or Dead is called for it; if the worker crashes before that, the driver's reaper re-queues the payload once its visibility timeout expires. This gives at-least-once semantics instead of losing jobs on crash.
type Job ¶
type Job interface {
// Handle executes the job. Returning a non-nil error triggers retries.
Handle(ctx context.Context) error
}
Job is the interface every queueable job must implement.
type Manager ¶
type Manager struct {
// contains filtered or unexported fields
}
Manager dispatches and processes jobs.
func NewManager ¶
NewManager creates a Manager backed by the given driver.
func (*Manager) Dispatch ¶
func (m *Manager) Dispatch(ctx context.Context, queue string, job Job, delay time.Duration, opts ...DispatchOpts) error
Dispatch pushes a job onto the named queue with optional delay.
m.Dispatch(ctx, "default", &SendEmail{To: "user@example.com"}, 0)
m.Dispatch(ctx, "default", &SendEmail{...}, 5*time.Minute) // delayed
type Options ¶
type Options struct {
// Queues lists which queues to process and in priority order.
Queues []string
// Workers is the number of concurrent goroutines polling for jobs (default: 5).
Workers int
// PollInterval is how often workers poll for new jobs (default: 1s).
PollInterval time.Duration
// Logger (defaults to slog.Default).
Logger *slog.Logger
}
Options configures the Manager.
type Payload ¶
type Payload struct {
// Class is the fully-qualified job type name (used by the registry to reconstruct it).
Class string `json:"class"`
// Data is the JSON-encoded job struct.
Data json.RawMessage `json:"data"`
// ID is a unique job identifier.
ID string `json:"id"`
// Attempts is the number of times this job has been attempted.
Attempts int `json:"attempts"`
// MaxAttempts is the maximum number of attempts before moving to dead-letter.
MaxAttempts int `json:"max_attempts"`
// Queue is the queue name this job was dispatched on.
Queue string `json:"queue"`
// AvailableAt is the UTC time after which the job may be picked up.
AvailableAt time.Time `json:"available_at"`
}
Payload wraps a Job for wire transport (JSON encoding).
Source Files
Directories