README
¶
Queue
Use arkoren.dev/core/service/queue to hand typed work to a consumer without
binding application handlers to a broker. Install with go get arkoren.dev/core.
The queue guides
cover driver ownership, producers, and workers.
Declare a shared job in application code. Producers and workers must agree on its name and payload type; worker policy belongs to the registered definition:
job := queue.NewJob[string]("send", func(job *queue.JobBuilder) {
job.Retry(3)
job.Timeout(time.Minute)
})
With an existing dispatcher, send a value:
err := dispatcher.Dispatch(ctx, job, "hello")
In your worker's composition callback, use an existing consumer and your
sendText(context.Context, string) error handler:
compose.Require(consumer.Handle(job, sendText))
compose.Run(consumer.Run)
Own the selected driver's startup and cleanup separately. Success from Dispatch
means driver acceptance, not execution. Handlers return nil for success; wrap
only non-nil failures with queue.Permanent. Permanent(nil) is still a terminal
failure, not success.
Consumer cancellation requests retry without applying the immediate attempt
cap. It does not discount the driver's delivery count: a later ordinary failure
can exhaust the cap. Settlement receives a fresh 30-second context after
execution, independent of consumer cancellation. Drivers and synchronous event
listeners must cooperate with that context; Go code is not forcibly terminated.
JobFailed is emitted only after terminal settlement succeeds.
The memory driver owns payload and metadata copies, checks cancellation before
reservation, and discards queued work on Stop. Its Stop waits only until its
context ends; an admitted handler can still be running. See the
driver comparison
for durability and backend failure handling, and the
queue API for declarations.
Documentation
¶
Overview ¶
Package queue provides transport-neutral asynchronous queue dispatch and execution.
A Job is an immutable definition: its name, queue, retry attempts, backoff, and per-attempt timeout. A Dispatcher sends a payload for a job, and a Consumer registers one handler per job and runs every queue with a handled job through Consumer.Run:
var SendWelcome = queue.NewJob[Welcome]("send-welcome", func(job *queue.JobBuilder) {
job.OnQueue("emails")
job.Retry(5)
job.Backoff(time.Second, 2, time.Minute)
})
Dispatch from a request or service operation, using an existing dispatcher:
if err := dispatcher.Dispatch(ctx, SendWelcome, Welcome{UserID: 1}); err != nil {
return err
}
Register the worker handler and runner during application composition:
compose.Require(consumer.Handle(SendWelcome, sendMail)) compose.Run(consumer.Run)
Always construct definitions with NewJob. A zero Job is invalid: Job.Err reports it, and Consumer.Handle and Dispatcher.Dispatch reject it with ErrInvalidName before registration or payload encoding.
Configuration is fixed at construction through role-specific JobBuilder, ConsumerBuilder, and DispatcherBuilder callbacks. Callbacks run directly once, synchronously in order on one fresh draft; supplied nil callbacks panic, while expanded nil or empty slices supply no callbacks. Constructors detach completed drafts. ConsumerBuilder.Queues also copies its input slice. Reusing a callback does not share mutable restrictions between consumers. Each Queues call replaces the restriction; an empty list restores consumption of all handled queues. Telemetry and event-bus pointers remain shared services, and nil telemetry follows telemetry.Default per operation.
A handler settles a delivery by returning: nil acknowledges, an ordinary error retries within the job's limits, and RetryAfter, Reject, and Permanent choose the outcome explicitly. Consumer cancellation requests retry without applying the immediate attempt cap, but does not reset or discount the driver's reported delivery count. Settlement uses a fresh 30-second context after execution, retaining context values independently of consumer cancellation. Drivers must honor that budget; synchronous event listeners must also cooperate with their context.
Message.Metadata travels beside each payload; when telemetry is enabled Dispatcher.Dispatch injects the trace context into it and Consumer continues that trace in a consumer span per delivery attempt. Both component builders expose Telemetry and System settings; dispatchers have no consumer- only Queues, UnknownNameAttempts, or Events settings.
With ConsumerBuilder.Events, a Consumer publishes, for every driver, JobFailed ("queue.job.failed") when it fails a delivery for good, JobRetried ("queue.job.retried") when it schedules another attempt, JobRejected ("queue.job.rejected") when a job rejects its message, and JobTimedOut ("queue.job.timed-out") when an attempt errors after its timeout elapsed. A timeout is followed by the event for its settlement, usually JobRetried.
Package queue provides transport-neutral asynchronous queue dispatch and execution.
Index ¶
- Constants
- Variables
- func Permanent(err error) error
- type Consumer
- type ConsumerBuilder
- func (consumer *ConsumerBuilder) Events(bus *event.Bus) *ConsumerBuilder
- func (consumer *ConsumerBuilder) Queues(queues ...string) *ConsumerBuilder
- func (consumer *ConsumerBuilder) System(system string) *ConsumerBuilder
- func (consumer *ConsumerBuilder) Telemetry(service *telemetry.Telemetry) *ConsumerBuilder
- func (consumer *ConsumerBuilder) UnknownNameAttempts(attempts int) *ConsumerBuilder
- type ConsumerBuilderFunc
- type ConsumerDriver
- type Delivery
- type DeliveryHandler
- type Dispatcher
- type DispatcherBuilder
- type DispatcherBuilderFunc
- type DispatcherDriver
- type Job
- type JobBuilder
- func (job *JobBuilder) Backoff(initial time.Duration, factor float64, maximum time.Duration) *JobBuilder
- func (job *JobBuilder) BackoffFunc(fn func(attempt int) time.Duration) *JobBuilder
- func (job *JobBuilder) Description(text string) *JobBuilder
- func (job *JobBuilder) OnQueue(queue string) *JobBuilder
- func (job *JobBuilder) Retry(attempts int) *JobBuilder
- func (job *JobBuilder) Timeout(duration time.Duration) *JobBuilder
- type JobBuilderFunc
- type JobFailedEvent
- type JobRejectedEvent
- type JobRetriedEvent
- type JobTimedOutEvent
- type Message
- type RejectedError
- type RetryAfterError
Examples ¶
Constants ¶
const ( // MaxMetadataEntries is the most metadata entries a driver must carry. // Larger maps fall outside the portable round-trip guarantee. // // if len(message.Metadata) > queue.MaxMetadataEntries { // return errors.New("too many metadata entries") // } MaxMetadataEntries = 16 // MaxMetadataKeyLength is the longest metadata key a driver must carry, // in bytes. // // if len(metadataKey) > queue.MaxMetadataKeyLength { // return errors.New("metadata key too long") // } MaxMetadataKeyLength = 64 // MaxMetadataSize is the most bytes of keys and values together a driver // must carry. // This counts key and value bytes together, not just encoded payload bytes. // // if metadataBytes > queue.MaxMetadataSize { // return errors.New("metadata too large") // } MaxMetadataSize = 4096 )
Limits of the Message.Metadata round-trip guarantee.
const DefaultQueue = "default"
DefaultQueue is the queue used by jobs created without JobBuilder.OnQueue. It is a literal routing destination, not an instruction to a native broker.
job := queue.NewJob[string]("send")
fmt.Println(job.Queue() == queue.DefaultQueue)
Variables ¶
var ErrDuplicateName = errors.New("duplicate job name")
ErrDuplicateName reports a second handler for one job name. The original registration remains unchanged.
if err := consumer.Handle(job, send); errors.Is(err, queue.ErrDuplicateName) {
return err
}
var ErrInvalidName = errors.New("invalid job name")
ErrInvalidName reports an invalid job. Consumer.Handle and Dispatcher.Dispatch return it, wrapping Job.Err, for a job whose name or settings are invalid.
err := dispatcher.Dispatch(ctx, queue.NewJob[string]("Bad Name"), "hello")
if errors.Is(err, queue.ErrInvalidName) {
return err
}
var ErrNoQueues = errors.New("no queues to consume")
ErrNoQueues reports that Consumer.Run has no queue to consume: no handler is registered, or ConsumerBuilder.Queues excludes every handled queue.
consumer := queue.NewConsumer(driver) err := consumer.Run(ctx) fmt.Println(errors.Is(err, queue.ErrNoQueues)) // true
var ErrPermanent = errors.New("permanent job failure")
ErrPermanent matches, under errors.Is, errors created by Permanent. It classifies a handler outcome, not a confirmed broker settlement.
err := queue.Permanent(errors.New("unsupported template"))
fmt.Println(errors.Is(err, queue.ErrPermanent)) // true
var ErrUnknownName = errors.New("unknown job name")
ErrUnknownName reports a job name the consumer has no handler for. The consumer retries such deliveries later, for example while a rolling deploy registers the handler on other workers.
if errors.Is(err, queue.ErrUnknownName) {
fmt.Println("check worker registration and rollout")
}
var JobFailed = event.NewTopic[JobFailedEvent]("queue.job.failed")
JobFailed is published when a Consumer fails a delivery for good: the job returned a Permanent error or exhausted its retries, the payload did not decode, or no handler took the name within the unknown-name attempts. Retries and rejections publish JobRetried and JobRejected instead. A failed terminal settlement publishes nothing; the broker outcome is not confirmed, and the settlement error is returned through Consumer.Run.
unsubscribe, err := bus.Listen(queue.JobFailed, recordFailure)
if err != nil {
return err
}
compose.Defer(unsubscribe)
var JobRejected = event.NewTopic[JobRejectedEvent]("queue.job.rejected")
JobRejected is published when a job returns a Reject error and the driver accepted the rejection, which drops the message without retrying and without failing it. Rejection is a deliberate outcome, so JobFailed is not published as well.
unsubscribe, err := bus.Listen(queue.JobRejected, recordRejection)
if err != nil {
return err
}
compose.Defer(unsubscribe)
var JobRetried = event.NewTopic[JobRetriedEvent]("queue.job.retried")
JobRetried is published when a Consumer hands a failed delivery back to the driver for another attempt and the driver accepted it. A timed-out attempt that is retried publishes JobTimedOut and then JobRetried. It is not published when the retries are exhausted (JobFailed is), when the consumer stops mid-job and releases the delivery, or for the driver's own redelivery of an unknown name.
unsubscribe, err := bus.Listen(queue.JobRetried, recordRetry)
if err != nil {
return err
}
compose.Defer(unsubscribe)
var JobTimedOut = event.NewTopic[JobTimedOutEvent]("queue.job.timed-out")
JobTimedOut is published when an attempt returned an error after the job's JobBuilder.Timeout elapsed. It describes the attempt and is published before the attempt is settled. Accepted settlement is followed by JobRetried when the job retries or by JobFailed when it is out of attempts (or by JobRejected if the job turned the error into a rejection). An attempt that finishes successfully after its deadline publishes nothing. A settlement error can prevent any subsequent settlement notification.
unsubscribe, err := bus.Listen(queue.JobTimedOut, recordTimeout)
if err != nil {
return err
}
compose.Defer(unsubscribe)
Functions ¶
func Permanent ¶
Permanent marks err as a permanent failure: the delivery is settled as failed and never retried, whatever attempts remain. A driver may archive or dead-letter it when its backend supports and is configured for that. The result matches ErrPermanent under errors.Is and still unwraps to err. Unlike broadcast.Permanent, Permanent(nil) is a non-nil terminal outcome. Check a fallible operation's error before wrapping it; return nil on success.
Example:
return queue.Permanent(fmt.Errorf("unsupported plan: %w", err))
Example ¶
ExamplePermanent distinguishes an explicit terminal outcome from success. Applications must check operation errors before wrapping them permanently.
package main
import (
"errors"
"fmt"
"arkoren.dev/core/service/queue"
)
func main() {
fmt.Println(queue.Permanent(nil) == nil)
fmt.Println(errors.Is(queue.Permanent(errors.New("unsupported template")), queue.ErrPermanent))
}
Output: false true
Types ¶
type Consumer ¶
type Consumer struct {
// contains filtered or unexported fields
}
Consumer decodes, handles, and settles deliveries received through a ConsumerDriver. It owns registration and per-delivery policy, not native driver lifecycle.
consumer := queue.NewConsumer(driver)
if err := consumer.Handle(job, send); err != nil {
return err
}
compose.Run(consumer.Run)
func NewConsumer ¶
func NewConsumer(driver ConsumerDriver, configure ...ConsumerBuilderFunc) *Consumer
NewConsumer creates a Consumer that consumes through driver. Register handlers with Consumer.Handle. Callbacks configure its telemetry: every delivery attempt emits a consumer span "process {queue}", parented to the trace context in Message.Metadata, and the messaging.client.consumed.messages and messaging.process.duration metrics when the telemetry is enabled. It costs nothing otherwise.
Callbacks run directly once, synchronously in order on one fresh draft. Nil entries panic; expanded nil or empty slices omit configuration. Construction performs no I/O. Publication detaches the draft and queue restrictions, but telemetry and event buses remain shared services. Nil telemetry resolves telemetry.Default at each operation; nil events publish nothing.
consumer := queue.NewConsumer(driver, func(consumer *queue.ConsumerBuilder) {
consumer.Telemetry(app.Telemetry())
})
func (*Consumer) Driver ¶
func (consumer *Consumer) Driver() ConsumerDriver
Driver returns the underlying ConsumerDriver. Raw consumption bypasses the facade's typed handlers and outcome events.
consumer := queue.NewConsumer(driver) fmt.Println(consumer.Driver() == driver)
func (*Consumer) DriverAs ¶
DriverAs returns the first driver of type T in the chain of the underlying driver. A driver that wraps another exposes it with an Unwrap() ConsumerDriver method, so wrappers such as prefixing or tracing drivers do not hide the concrete driver.
if driver, ok := consumer.DriverAs[*memoryqueue.Driver](); ok {
fmt.Printf("memory consumption: %T\n", driver)
}
func (*Consumer) Handle ¶
Handle registers fn as the handler of job. A nil result acknowledges the delivery. An ordinary error requests retry according to the job's JobBuilder.Retry and JobBuilder.Backoff settings. Use RetryAfter, Reject, or Permanent to request an explicit settlement. An invalid job returns ErrInvalidName wrapping Job.Err; registering one name twice returns ErrDuplicateName. Run discovers handled queues once at startup. A handler added later can handle subsequent deliveries on a queue already being consumed; a new queue is not consumed until the next Run. A nil fn returns an error without registering the job.
job := queue.NewJob[string]("send")
if err := consumer.Handle(job, func(ctx context.Context, text string) error {
return sendText(ctx, text)
}); err != nil {
return err
}
func (*Consumer) Run ¶
Run consumes and handles deliveries from every queue that has a handled job, with one ConsumerDriver.Consume call per queue, until ctx is canceled or any consume fails. The first failure cancels the others and is returned once they all stop; cancellation returns the error the driver reports for it. Use ConsumerBuilder.Queues to restrict the queues. It returns ErrNoQueues when there is nothing to consume.
Register a runner during application composition:
compose.Run(consumer.Run)
type ConsumerBuilder ¶
type ConsumerBuilder struct {
// contains filtered or unexported fields
}
ConsumerBuilder configures one consumer before publication. Constructors invoke callbacks directly, once and in order on the same draft. Retaining the draft does not permit changing the published consumer. Service pointers and function captures remain shared; concurrent draft mutation is unsupported.
consumer := queue.NewConsumer(driver, func(consumer *queue.ConsumerBuilder) {
consumer.Queues("emails").UnknownNameAttempts(20)
})
func (*ConsumerBuilder) Events ¶
func (consumer *ConsumerBuilder) Events(bus *event.Bus) *ConsumerBuilder
Events selects the shared bus for job outcomes and returns consumer. Nil publishes nothing; listeners never change the settlement outcome.
queue.NewConsumer(driver, func(consumer *queue.ConsumerBuilder) {
consumer.Events(bus)
})
func (*ConsumerBuilder) Queues ¶
func (consumer *ConsumerBuilder) Queues(queues ...string) *ConsumerBuilder
Queues replaces the restriction and returns consumer. It copies the input; publication copies it again. Empty removes the restriction. Run skips queues without handled jobs, even when named here.
queue.NewConsumer(driver, func(consumer *queue.ConsumerBuilder) {
consumer.Queues("emails")
})
func (*ConsumerBuilder) System ¶
func (consumer *ConsumerBuilder) System(system string) *ConsumerBuilder
System overrides messaging.system and returns consumer. Empty restores driver discovery, falling back to "_OTHER" when the driver has no system.
queue.NewConsumer(driver, func(consumer *queue.ConsumerBuilder) {
consumer.System("custom")
})
func (*ConsumerBuilder) Telemetry ¶
func (consumer *ConsumerBuilder) Telemetry(service *telemetry.Telemetry) *ConsumerBuilder
Telemetry chooses the shared telemetry service and returns consumer. Nil follows telemetry.Default at each operation, including later default changes.
queue.NewConsumer(driver, func(consumer *queue.ConsumerBuilder) {
consumer.Telemetry(nil)
})
func (*ConsumerBuilder) UnknownNameAttempts ¶
func (consumer *ConsumerBuilder) UnknownNameAttempts(attempts int) *ConsumerBuilder
UnknownNameAttempts sets the unknown-job retry cap and returns consumer. Zero uses ten, negative retries indefinitely, with a 30-second delay. Drivers must report Message.Attempts for the cap to take effect.
queue.NewConsumer(driver, func(consumer *queue.ConsumerBuilder) {
consumer.UnknownNameAttempts(20)
})
type ConsumerBuilderFunc ¶
type ConsumerBuilderFunc = func(*ConsumerBuilder)
ConsumerBuilderFunc edits one fresh consumer draft. Queue restrictions are detached at publication; telemetry and event services stay shared. Supplied nil callbacks panic, while expanded empty callback slices make no edits.
policy := queue.ConsumerBuilderFunc(func(consumer *queue.ConsumerBuilder) {
consumer.Queues("work")
})
consumer := queue.NewConsumer(driver, policy)
type ConsumerDriver ¶
type ConsumerDriver interface {
// Consume receives deliveries from queue until ctx is canceled or an error
// occurs. The handler is responsible for settling every accepted delivery.
// Cancellation ends Consume with an error matching the context error under
// [errors.Is], and a context that is already canceled is rejected the same
// way. Concurrent Consume calls on one queue compete for messages. The
// driver reports [Message.Attempts] on every delivery or on none.
//
// if err := driver.Consume(ctx, "work", handler); err != nil {
// return err
// }
Consume(ctx context.Context, queue string, handler DeliveryHandler) error
}
ConsumerDriver receives deliveries from a named queue. It must not acknowledge a delivery before handler returns successfully. The handler, not the driver's return value, selects delivery settlement.
consumer := queue.NewConsumer(driver)
if err := consumer.Handle(job, send); err != nil {
return err
}
compose.Run(consumer.Run)
type Delivery ¶
type Delivery interface {
// Message returns the job message being delivered.
//
// message := delivery.Message()
// fmt.Println(message.Name, message.Attempts)
Message() Message
// Acknowledge permanently completes this delivery.
//
// if err := delivery.Acknowledge(ctx); err != nil {
// return err
// }
Acknowledge(ctx context.Context) error
// Retry requests another delivery of the same message, with Attempts
// incremented when the driver reports it. A zero delay delegates timing to
// the driver's configured default. A positive delay is a lower bound: the
// message is not delivered again earlier, though drivers may round it up.
//
// if err := delivery.Retry(ctx, time.Minute); err != nil {
// return err
// }
Retry(ctx context.Context, delay time.Duration) error
// Reject permanently rejects this delivery. The driver's broker may
// dead-letter rejected deliveries according to its configured policy;
// brokers that implement rejection as redrive may deliver the message
// again until that policy moves it.
//
// if err := delivery.Reject(ctx); err != nil {
// return err
// }
Reject(ctx context.Context) error
// Fail requests terminal failure handling for this delivery: the message
// is not delivered again. The driver's broker may archive or dead-letter
// it according to its configured policy.
//
// if err := delivery.Fail(ctx, cause); err != nil {
// return err
// }
Fail(ctx context.Context, err error) error
}
Delivery is a broker-reserved job message. Its settlement methods hide transport details such as SQS receipt handles and AMQP delivery tags. A delivery is settled exactly once: a second settlement call, concurrent or not, returns an error and changes nothing. Settlement failures reach the consumer driver; an unsuccessful settlement does not confirm the requested outcome. Contexts bound settlement operations.
if err := delivery.Acknowledge(ctx); err != nil {
return err
}
type DeliveryHandler ¶
DeliveryHandler handles one broker-reserved Delivery. Returning an error reports a consumer or settlement failure to the consumer driver. Successful handlers must settle the delivery; returning nil alone is not an acknowledgement. Use Consumer for typed execution and settlement policy.
handler := queue.DeliveryHandler(func(ctx context.Context, delivery queue.Delivery) error {
return delivery.Acknowledge(ctx)
})
if err := driver.Consume(ctx, "work", handler); err != nil {
return err
}
type Dispatcher ¶
type Dispatcher struct {
// contains filtered or unexported fields
}
Dispatcher JSON-encodes job payloads and dispatches their transport envelopes through a DispatcherDriver. It does not own native driver lifecycle or start a consumer.
dispatcher := queue.NewDispatcher(driver)
if err := dispatcher.Dispatch(ctx, queue.NewJob[string]("send"), "hello"); err != nil {
return err
}
func NewDispatcher ¶
func NewDispatcher(driver DispatcherDriver, configure ...DispatcherBuilderFunc) *Dispatcher
NewDispatcher creates a Dispatcher that delegates to driver. Callbacks configure its telemetry: every dispatch emits a producer span "send {queue}" and the messaging.client.sent.messages and messaging.client.operation.duration metrics, and injects the trace context into Message.Metadata, when the telemetry is enabled. It costs nothing otherwise. Payloads and metadata values are never recorded.
Callbacks run directly once, synchronously in order on one fresh draft. Nil entries panic; expanded nil or empty slices omit configuration. Publication detaches the draft; telemetry remains a shared service, resolved dynamically through telemetry.Default when nil. Construction performs no I/O.
dispatcher := queue.NewDispatcher(driver, func(dispatcher *queue.DispatcherBuilder) {
dispatcher.Telemetry(app.Telemetry())
})
func (*Dispatcher) Dispatch ¶
Dispatch JSON-encodes payload and sends it to the queue of job with a new random message ID. It returns ErrInvalidName, wrapping Job.Err, for an invalid job. With enabled telemetry the message carries the trace context of the dispatch span in Message.Metadata. Encoding errors and driver failures are returned without losing their causes; a successful call confirms acceptance, not handler completion. Invalid jobs are rejected before encoding. The facade does not start or stop the driver.
job := queue.NewJob[string]("send")
if err := dispatcher.Dispatch(ctx, job, "hello"); err != nil {
return err
}
func (*Dispatcher) Driver ¶
func (dispatcher *Dispatcher) Driver() DispatcherDriver
Driver returns the underlying DispatcherDriver. Raw calls bypass typed encoding and facade telemetry.
dispatcher := queue.NewDispatcher(driver) fmt.Println(dispatcher.Driver() == driver)
func (*Dispatcher) DriverAs ¶
func (dispatcher *Dispatcher) DriverAs[T any]() (T, bool)
DriverAs returns the first driver of type T in the chain of the underlying driver. A driver that wraps another exposes it with an Unwrap() DispatcherDriver method, so wrappers such as prefixing or tracing drivers do not hide the concrete driver.
if driver, ok := dispatcher.DriverAs[*memoryqueue.Driver](); ok {
fmt.Printf("memory dispatch: %T\n", driver)
}
type DispatcherBuilder ¶
type DispatcherBuilder struct {
// contains filtered or unexported fields
}
DispatcherBuilder configures dispatcher telemetry before publication. It deliberately exposes no consumer-only settings. Callbacks share one draft and run directly once in order; retained drafts cannot change a dispatcher.
dispatcher := queue.NewDispatcher(driver, func(dispatcher *queue.DispatcherBuilder) {
dispatcher.System("custom")
})
func (*DispatcherBuilder) System ¶
func (dispatcher *DispatcherBuilder) System(system string) *DispatcherBuilder
System overrides messaging.system and returns dispatcher. Empty restores driver discovery, falling back to "_OTHER" for unnamed drivers.
queue.NewDispatcher(driver, func(dispatcher *queue.DispatcherBuilder) {
dispatcher.System("custom")
})
func (*DispatcherBuilder) Telemetry ¶
func (dispatcher *DispatcherBuilder) Telemetry(service *telemetry.Telemetry) *DispatcherBuilder
Telemetry selects the shared telemetry service and returns dispatcher. Nil follows telemetry.Default dynamically at each dispatch, not at construction.
queue.NewDispatcher(driver, func(dispatcher *queue.DispatcherBuilder) {
dispatcher.Telemetry(nil)
})
type DispatcherBuilderFunc ¶
type DispatcherBuilderFunc = func(*DispatcherBuilder)
DispatcherBuilderFunc edits a dispatcher draft once during construction. It configures only producer settings and cannot change consumer policy; publication detaches the draft without cloning service dependencies.
policy := queue.DispatcherBuilderFunc(func(dispatcher *queue.DispatcherBuilder) {
dispatcher.System("custom")
})
dispatcher := queue.NewDispatcher(driver, policy)
type DispatcherDriver ¶
type DispatcherDriver interface {
// Dispatch adds message to its named queue. It returns an error when
// message.Queue is empty and the context error when ctx is already
// canceled. The driver delivers Name, Queue, Payload, and Metadata
// unchanged to consumers (see [Message.Metadata]). Message.ID is stable across redeliveries of one message, but a
// driver may replace the dispatched value with a broker identifier.
// Attempts is reported by the consumer side and ignored here.
//
// if err := driver.Dispatch(ctx, message); err != nil {
// return err
// }
Dispatch(ctx context.Context, message Message) error
}
DispatcherDriver dispatches raw Message values. A successful dispatch confirms only that the driver accepted the message, not that a job handler processed it.
var transport queue.DispatcherDriver = memoryqueue.New()
dispatcher := queue.NewDispatcher(transport)
if err := dispatcher.Dispatch(ctx, queue.NewJob[string]("send"), "hello"); err != nil {
return err
}
type Job ¶
type Job[T any] struct { // contains filtered or unexported fields }
Job names one kind of job and fixes the payload type T it carries, together with the queue, retry, and timeout settings that are application code rather than transport data. Create it with NewJob and configure it through a JobBuilder callback before publication.
A Job is an immutable value: its builder is detached at publication, and the first error found is kept and reported by Job.Err. Consumer.Handle and Dispatcher.Dispatch reject a job whose Err is not nil with ErrInvalidName. Always create jobs with NewJob; the zero value has no name and no queue, and Job.Err reports it as invalid.
Example:
var SendWelcome = queue.NewJob[Welcome]("send-welcome", func(job *queue.JobBuilder) {
job.OnQueue("emails")
job.Retry(5)
job.Backoff(time.Second, 2, time.Minute)
job.Timeout(30 * time.Second)
})
func NewJob ¶
func NewJob[T any](jobName string, configure ...JobBuilderFunc) Job[T]
NewJob returns the job called jobName that carries payloads of type T and runs on DefaultQueue. The name is the stable identifier used to find the handler of a delivery; it must satisfy the shared identifier rule of 1-64 lowercase letters, digits, or hyphens with no hyphen at either end. An invalid name is reported by Job.Err rather than by a panic. Callbacks run directly once, synchronously and in order on one fresh draft, even when the name is invalid. Nil entries panic; expanded nil or empty slices supply no callbacks. The first validation error is retained across callbacks. Retained drafts cannot change the published job; closure captures stay shared.
job := queue.NewJob[string]("email", func(job *queue.JobBuilder) {
job.Retry(3)
})
if err := job.Err(); err != nil {
return err
}
Example ¶
ExampleNewJob shows the immutable published routing and attempt settings. Definition errors are checked before the value is registered or dispatched.
package main
import (
"fmt"
"time"
"arkoren.dev/core/service/queue"
)
func main() {
job := queue.NewJob[string]("send", func(job *queue.JobBuilder) {
job.OnQueue("work")
job.Retry(3)
job.Timeout(time.Minute)
})
if err := job.Err(); err != nil {
fmt.Println(err)
return
}
fmt.Println(job.Name(), job.Queue(), job.Attempts(), job.TimeoutValue())
}
Output: send work 3 1m0s
func (Job[T]) Attempts ¶
Attempts returns the total allowed attempts, or zero for the driver default. It counts the first delivery as an attempt when drivers report delivery counts.
job := queue.NewJob[string]("send", func(job *queue.JobBuilder) {
job.Retry(3)
})
fmt.Println(job.Attempts()) // 3
func (Job[T]) DescriptionValue ¶
DescriptionValue returns the free text set with JobBuilder.Description. An omitted description is empty and does not affect routing or execution.
job := queue.NewJob[string]("send", func(job *queue.JobBuilder) {
job.Description("Sends a reminder")
})
fmt.Println(job.DescriptionValue())
func (Job[T]) Err ¶
Err returns the first error found while building the job, or nil for a valid definition. The zero Job is invalid; only NewJob supplies its identity. Consumer.Handle and Dispatcher.Dispatch wrap this error with ErrInvalidName before registering a handler or encoding a payload.
job := queue.NewJob[string]("send-welcome")
if err := job.Err(); err != nil {
return err
}
func (Job[T]) Name ¶
Name returns the job name. It is empty for the invalid zero Job; NewJob preserves the declared identity.
job := queue.NewJob[string]("send")
fmt.Println(job.Name()) // send
func (Job[T]) Queue ¶
Queue returns the queue the job is dispatched to. NewJob defaults to DefaultQueue; a zero Job has no queue and must not be used.
job := queue.NewJob[string]("send")
fmt.Println(job.Queue()) // default
func (Job[T]) TimeoutValue ¶
TimeoutValue returns the bound on one attempt, or zero for no limit. Execution must cooperate with its canceled context; this is not preemption.
job := queue.NewJob[string]("send")
fmt.Println(job.TimeoutValue() == 0) // true
type JobBuilder ¶
type JobBuilder struct {
// contains filtered or unexported fields
}
JobBuilder configures retry policy and routing before an immutable Job is published. Every method returns this same draft. Retained drafts cannot change published jobs; custom backoff closures retain their own captures.
job := queue.NewJob[string]("email", func(job *queue.JobBuilder) {
job.OnQueue("emails").Retry(5)
})
func (*JobBuilder) Backoff ¶
func (job *JobBuilder) Backoff(initial time.Duration, factor float64, maximum time.Duration) *JobBuilder
Backoff sets the delay before the first retry, the factor that multiplies it after each retry, and the cap on the delay. A zero maximum means no cap and a factor below one counts as one (constant delay). A negative initial or maximum is an error. It replaces any earlier JobBuilder.BackoffFunc. NaN also normalizes to one; overflow saturates at the largest duration. It returns the same draft, preserving any earlier validation error.
queue.NewJob[string]("email", func(job *queue.JobBuilder) {
job.Backoff(time.Second, 2, time.Minute)
})
func (*JobBuilder) BackoffFunc ¶
func (job *JobBuilder) BackoffFunc(fn func(attempt int) time.Duration) *JobBuilder
BackoffFunc sets a custom delay before a retry. fn receives the number of the attempt about to run (always at least 2). Negative delays count as zero. A nil fn restores the driver default. It replaces any earlier JobBuilder.Backoff. It returns the same draft and does not clear an earlier error. The function and its captured state remain application-owned.
queue.NewJob[string]("email", func(job *queue.JobBuilder) {
job.BackoffFunc(nil)
})
func (*JobBuilder) Description ¶
func (job *JobBuilder) Description(text string) *JobBuilder
Description sets free text for people and returns the same draft. It does not affect delivery or clear an earlier validation error.
queue.NewJob[string]("email", func(job *queue.JobBuilder) {
job.Description("Sends welcome mail")
})
func (*JobBuilder) OnQueue ¶
func (job *JobBuilder) OnQueue(queue string) *JobBuilder
OnQueue sets the queue the job is dispatched to. An empty queue is an error. The default is DefaultQueue. It returns the same draft and preserves any earlier validation error.
queue.NewJob[string]("email", func(job *queue.JobBuilder) {
job.OnQueue("emails")
})
func (*JobBuilder) Retry ¶
func (job *JobBuilder) Retry(attempts int) *JobBuilder
Retry sets the total number of attempts, including the first one, before an ordinary handler error becomes a permanent failure. Zero uses the driver default and a negative value is an error. The limit is enforced only when the driver reports Message.Attempts; otherwise the driver's own redelivery limit applies. The same draft is returned; any earlier validation error is preserved.
queue.NewJob[string]("email", func(job *queue.JobBuilder) {
job.Retry(3)
})
func (*JobBuilder) Timeout ¶
func (job *JobBuilder) Timeout(duration time.Duration) *JobBuilder
Timeout bounds each attempt of the handler. An attempt that exceeds it has its context canceled and counts as a failed attempt; an attempt interrupted by consumer shutdown is retried without applying the attempt cap to that interruption. Reported delivery counts still advance and can exhaust the cap on a later ordinary failure. Zero means no limit and a negative value is an error. It returns the same draft and preserves earlier validation errors.
queue.NewJob[string]("email", func(job *queue.JobBuilder) {
job.Timeout(time.Minute)
})
type JobBuilderFunc ¶
type JobBuilderFunc = func(*JobBuilder)
JobBuilderFunc edits one fresh JobBuilder during NewJob. Policies run once in order even for invalid declarations; nil entries panic. Retaining the draft cannot mutate a published Job, but captured application state is shared.
policy := queue.JobBuilderFunc(func(job *queue.JobBuilder) {
job.Retry(3)
})
job := queue.NewJob[string]("send", policy)
if err := job.Err(); err != nil {
return err
}
type JobFailedEvent ¶
type JobFailedEvent struct {
// ID is the message identifier.
ID string
// Name is the job name.
Name string
// Queue is the queue the message was consumed from.
Queue string
// Attempts is the driver's reported delivery count, or zero when unavailable.
Attempts int
// Err is the reason: the job's error, a [Permanent] error, a payload
// decode error, or [ErrUnknownName].
Err error
}
JobFailedEvent is the payload of JobFailed. It reports an accepted terminal settlement without carrying job contents. Error causes remain shared application values, not sanitized wire messages.
unsubscribe, err := bus.Listen(queue.JobFailed, func(_ context.Context, outcome queue.JobFailedEvent) error {
fmt.Println(outcome.Name, outcome.Attempts)
return nil
})
if err != nil {
return err
}
compose.Defer(unsubscribe)
type JobRejectedEvent ¶
type JobRejectedEvent struct {
// ID is the message identifier.
ID string
// Name is the job name.
Name string
// Queue is the queue the message was consumed from.
Queue string
// Attempts is the driver's reported delivery count, or zero when unavailable.
Attempts int
// Err is the [Reject] error the job returned.
Err error
}
JobRejectedEvent is the payload of JobRejected. It describes the facade's accepted rejection, not a broker's dead-letter state.
unsubscribe, err := bus.Listen(queue.JobRejected, func(_ context.Context, outcome queue.JobRejectedEvent) error {
fmt.Println(outcome.Name, outcome.Attempts)
return nil
})
if err != nil {
return err
}
compose.Defer(unsubscribe)
type JobRetriedEvent ¶
type JobRetriedEvent struct {
// ID is the message identifier.
ID string
// Name is the job name.
Name string
// Queue is the queue the message was consumed from.
Queue string
// Attempts is the driver's reported delivery count, or zero when unavailable.
Attempts int
// Delay is how long the driver was asked to wait before redelivering:
// the [RetryAfter] delay, the job's backoff, or zero.
Delay time.Duration
// Err is the job error that caused the retry.
Err error
}
JobRetriedEvent is the payload of JobRetried. Delay is a request to the driver, not proof of an exact future execution time.
unsubscribe, err := bus.Listen(queue.JobRetried, func(_ context.Context, outcome queue.JobRetriedEvent) error {
fmt.Println(outcome.Name, outcome.Delay)
return nil
})
if err != nil {
return err
}
compose.Defer(unsubscribe)
type JobTimedOutEvent ¶
type JobTimedOutEvent struct {
// ID is the message identifier.
ID string
// Name is the job name.
Name string
// Queue is the queue the message was consumed from.
Queue string
// Attempts is the driver's reported delivery count, or zero when unavailable.
Attempts int
// Timeout is the job's per-attempt limit that elapsed.
Timeout time.Duration
}
JobTimedOutEvent is the payload of JobTimedOut. Timeout describes the elapsed attempt deadline, before settlement is attempted.
unsubscribe, err := bus.Listen(queue.JobTimedOut, func(_ context.Context, outcome queue.JobTimedOutEvent) error {
fmt.Println(outcome.Name, outcome.Timeout)
return nil
})
if err != nil {
return err
}
compose.Defer(unsubscribe)
type Message ¶
type Message struct {
// ID identifies the message. [Dispatcher.Dispatch] assigns a random ID;
// drivers that deduplicate, such as JetStream, use it as the message ID.
ID string `json:"id"`
// Name identifies the application job definition.
Name string `json:"name"`
// Queue identifies the queue that received the job.
Queue string `json:"queue"`
// Payload is the JSON encoding of the job value.
Payload []byte `json:"payload"`
// Attempts is the number of delivery attempts when the driver provides that
// information. A value of zero means it is unavailable.
Attempts int `json:"attempts"`
// Metadata carries string key-value pairs beside the payload, such as the
// trace context [Dispatcher.Dispatch] injects. Drivers deliver it to
// consumers unchanged: nil and empty are equivalent, and a message
// dispatched without metadata arrives with nil Metadata. The guarantee
// covers at most MaxMetadataEntries entries whose keys are 1 to
// MaxMetadataKeyLength bytes of [A-Za-z0-9._-] (case-sensitive) and whose
// values are valid UTF-8, with at most MaxMetadataSize bytes of keys and
// values together. Beyond those limits a driver may reject the message.
// Metadata is not part of the payload and never affects routing.
Metadata map[string]string `json:"metadata,omitzero"`
}
Message is the transport delivery envelope for a job. Retry settings are intentionally excluded because the consumer takes them from the job registered for Name. Raw callers own validation and encoding; Dispatcher creates these envelopes from immutable Job definitions. Metadata limits describe portable delivery.
message := queue.Message{Queue: queue.DefaultQueue, Name: "send", Payload: []byte(`"hello"`)}
if err := driver.Dispatch(ctx, message); err != nil {
return err
}
type RejectedError ¶
type RejectedError struct {
// contains filtered or unexported fields
}
RejectedError requests terminal rejection. Broker dead-lettering, if any, is determined by the driver's backend configuration. Create it with Reject. A nil cause is allowed; rejection remains an explicit settlement outcome.
err := queue.Reject(errors.New("invalid recipient"))
fmt.Println(err.Error())
func Reject ¶
func Reject(err error) *RejectedError
Reject requests terminal rejection. Broker dead-lettering, if any, is determined by the driver's backend configuration. See RejectedError.
Example:
return queue.Reject(errors.New("missing user"))
func (*RejectedError) Error ¶
func (rejectedError *RejectedError) Error() string
Error returns the underlying cause message, or a generic outcome message when no cause was provided. It does not confirm that a broker accepted rejection.
err := queue.Reject(errors.New("invalid recipient"))
fmt.Println(err.Error())
func (*RejectedError) Unwrap ¶
func (rejectedError *RejectedError) Unwrap() error
Unwrap returns the error that caused this outcome. A nil rejection cause remains nil; the explicit rejection is still non-nil.
cause := errors.New("invalid recipient")
err := queue.Reject(cause)
fmt.Println(errors.Is(err.Unwrap(), cause)) // true
type RetryAfterError ¶
type RetryAfterError struct {
// contains filtered or unexported fields
}
RetryAfterError requests another attempt after a delay. The delay is a requested wait, not a guarantee: drivers may be unable to schedule delayed retries exactly or at all. errors.Is and errors.AsType traverse its cause. Create it with RetryAfter. A nil cause is allowed; the result still requests retry rather than success.
err := queue.RetryAfter(errors.New("busy"), time.Second)
fmt.Println(err.RetryAfter()) // 1s
func RetryAfter ¶
func RetryAfter(err error, delay time.Duration) *RetryAfterError
RetryAfter requests another attempt after delay, subject to the job's attempt limit. The delay is a request, not a guarantee; see RetryAfterError. A negative delay counts as zero. Ordinary handler errors are retried already; use RetryAfter only to choose the delay.
Example:
return queue.RetryAfter(err, time.Minute)
func (*RetryAfterError) Error ¶
func (retryAfterError *RetryAfterError) Error() string
Error returns the underlying cause message, or a generic outcome message when no cause was provided. It describes the cause, not the requested delay or broker acceptance.
err := queue.RetryAfter(errors.New("busy"), time.Second)
fmt.Println(err.Error()) // busy
func (*RetryAfterError) RetryAfter ¶
func (retryAfterError *RetryAfterError) RetryAfter() time.Duration
RetryAfter returns the requested delay before retrying. It retains the original value; the consumer normalizes a negative delay to zero when selecting settlement. Native transport timing remains driver-owned.
err := queue.RetryAfter(errors.New("busy"), time.Second)
fmt.Println(err.RetryAfter()) // 1s
func (*RetryAfterError) Unwrap ¶
func (retryAfterError *RetryAfterError) Unwrap() error
Unwrap returns the error that caused this outcome. A nil cause stays nil even though RetryAfter itself is a non-nil outcome.
cause := errors.New("busy")
err := queue.RetryAfter(cause, time.Second)
fmt.Println(errors.Is(err.Unwrap(), cause)) // true
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
internal
|
|
|
execution
Package execution owns queue attempt execution and settlement policy.
|
Package execution owns queue attempt execution and settlement policy. |
|
measurement
Package measurement supplies allocation measurements and race-mode metadata for tests.
|
Package measurement supplies allocation measurements and race-mode metadata for tests. |
|
Package memoryqueue provides a non-durable, in-process queue driver for local development.
|
Package memoryqueue provides a non-durable, in-process queue driver for local development. |
|
internal/broker
Package broker owns memory queue admission, delivery and delayed retries.
|
Package broker owns memory queue admission, delivery and delayed retries. |
|
Package queuetest tests code that dispatches or consumes jobs.
|
Package queuetest tests code that dispatches or consumes jobs. |