queue

type BackoffFunc ¶

type BackoffFunc func(attempt int) time.Duration

BackoffFunc returns the delay before the given retry attempt. attempt is 1 after the first failure, 2 after the second, and so on.

type Clock ¶

type Clock interface {
	Now() time.Time
	NewTimer(d time.Duration) Timer
}

Clock abstracts time for deterministic testing. The default implementation uses the real time package; queuetest provides a controllable FakeClock.

Backends accept a Clock via the (unexported, internal) WithClock option in their respective sub-packages — exported only inside this module via internal helper functions to keep the public surface small.

func RealClock() Clock

type Consumer ¶

type Consumer interface {
	Close() error
}

Consumer is the lifetime handle for a subscription. Close stops claiming new jobs and waits for in-flight handlers up to the configured shutdown timeout.

type EnqueueConfig ¶

type EnqueueConfig struct {
	App       string
	Namespace string
	Priority  Priority
	// RunAt is the absolute time at which the job becomes eligible. Zero
	// means "now". WithDelay and WithRunAt both populate this field; WithRunAt
	// always wins (last-wins, but RunAt is also "stronger" semantically).
	RunAt time.Time
}

EnqueueConfig is the resolved configuration for a single Enqueue call. Backends populate the App, Namespace, and Priority defaults from their own service-level config, then apply EnqueueOptions to overlay caller intent.

type EnqueueOption ¶

type EnqueueOption interface {
	Apply(*EnqueueConfig)
}

EnqueueOption configures a single Enqueue call.

func WithDelay(d time.Duration) EnqueueOption

func WithEnqueueApp(app string) EnqueueOption

func WithEnqueueNamespace(ns string) EnqueueOption

func WithPriority(p Priority) EnqueueOption

func WithRunAt(t time.Time) EnqueueOption

type EnqueueOptionFunc ¶

type EnqueueOptionFunc func(*EnqueueConfig)

EnqueueOptionFunc adapts a function to EnqueueOption.

func (f EnqueueOptionFunc) Apply(cfg *EnqueueConfig)

type Enqueuer ¶

type Enqueuer interface {
	Enqueue(ctx context.Context, topic string, payload []byte, opts ...EnqueueOption) error
}

Enqueuer enqueues jobs onto a topic.

type Handler ¶

type Handler func(ctx context.Context, job *Job) error

Handler processes a single job. Returning nil acknowledges the job and the backend removes it from the queue. Returning a non-nil error schedules a retry according to the configured backoff and max retry count, unless the error matches ErrSkipRetry, in which case the job is routed straight to the dead letter topic on this attempt.

type Job ¶

type Job struct {
	// ID is the backend-assigned unique identifier (UUID for durable backends,
	// monotonic for inmem). Always non-empty when delivered to a Handler.
	ID string
	// Topic is the fully formatted topic name as it appears in the backend
	// (i.e. after Format).
	Topic string
	// Payload is the opaque body supplied to Enqueue.
	Payload []byte
	// Attempt is 1 on first delivery, 2 on the first retry, and so on.
	Attempt int
	// EnqueuedAt is the time the job was first enqueued (preserved across
	// retries).
	EnqueuedAt time.Time
}

Job is the unit of work delivered to a Handler.

type Priority ¶

type Priority int8

Priority controls relative ordering of jobs in the same topic. Higher priority jobs are delivered before lower priority jobs once they are eligible to run.

const (
	PriorityLow    Priority = -1
	PriorityNormal Priority = 0
	PriorityHigh   Priority = 1
)

type QueueConfig ¶

type QueueConfig struct {
	App       string
	Namespace string
}

QueueConfig is the service-level configuration shared by Enqueue and Subscribe.

type QueueOption ¶

type QueueOption interface {
	Apply(*QueueConfig)
}

QueueOption configures a backend constructor.

func WithApp(app string) QueueOption

func WithNamespace(ns string) QueueOption

type QueueOptionFunc ¶

type QueueOptionFunc func(*QueueConfig)

QueueOptionFunc adapts a function to QueueOption.

func (f QueueOptionFunc) Apply(cfg *QueueConfig)

type Service ¶

type Service interface {
	Enqueuer
	Subscriber
	// Close releases backend resources. Safe to call once; subsequent calls
	// return nil.
	Close(ctx context.Context) error
}

Service is the union of Enqueuer and Subscriber plus a backend Close. Backend implementations expose this as their public type.

type SubscribeConfig ¶

type SubscribeConfig struct {
	App       string
	Namespace string

	Concurrency       int
	MaxRetries        int
	PollInterval      time.Duration
	VisibilityTimeout time.Duration
	ShutdownTimeout   time.Duration

	// DeadLetterTopic is the topic that exhausted jobs are routed to. Zero
	// value means "<topic>.dead".
	DeadLetterTopic string

	// Backoff returns the delay before the given retry attempt. Zero value
	// means DefaultBackoff.
	Backoff BackoffFunc
}

SubscribeConfig is the resolved configuration for a single Subscribe call.

type SubscribeOption ¶

type SubscribeOption interface {
	Apply(*SubscribeConfig)
}

SubscribeOption configures a single Subscribe call.

func WithBackoff(b BackoffFunc) SubscribeOption

func WithConcurrency(n int) SubscribeOption

func WithDeadLetterTopic(topic string) SubscribeOption

func WithMaxRetries(n int) SubscribeOption

func WithPollInterval(d time.Duration) SubscribeOption

func WithShutdownTimeout(d time.Duration) SubscribeOption

func WithSubscribeApp(app string) SubscribeOption

func WithSubscribeNamespace(ns string) SubscribeOption

func WithVisibilityTimeout(d time.Duration) SubscribeOption

type SubscribeOptionFunc ¶

type SubscribeOptionFunc func(*SubscribeConfig)

SubscribeOptionFunc adapts a function to SubscribeOption.

func (f SubscribeOptionFunc) Apply(cfg *SubscribeConfig)

type Subscriber ¶

type Subscriber interface {
	Subscribe(ctx context.Context, topic string, h Handler, opts ...SubscribeOption) (Consumer, error)
}

Subscriber registers a Handler against a topic. The returned Consumer controls the lifetime of the worker(s).

type Timer ¶

type Timer interface {
	C() <-chan time.Time
	Stop() bool
	Reset(d time.Duration) bool
}

Timer mirrors time.Timer minus the parts queue does not need. Stop returns true if the call stops the timer; false if the timer already fired.