mqutils
GitLab ↗

Docs / Core / Runner

Runner

The shared consumer loop: settlement, retry budgets, batching, reconnect and graceful shutdown

go
import "go.digitalxero.dev/mqutils/v2/runner"

Package runner provides the shared consumer processing loop used by every mqutils backend. It owns the concerns that are identical across message queue systems — dispatching messages to handlers, acknowledging based on the handler’s returned error, batch collection with a real flush timer, retry-budget drops with a dead-letter hook, reconnect with exponential backoff, and graceful shutdown draining — so backends only supply the broker-specific pieces via callbacks.

Backends construct a Runner with the Builder returned by New:

go
r, err := runner.New().
    WithName("amqp").
    WithConnect(connectFn).
    WithHandler(handler).
    WithAutoReconnect(true).
    WithConcurrency(1).
    WithGracefulShutdown(true, 30*time.Second).
    WithNackOnShutdown(true).
    Build()
if err != nil {
    return err
}
return r.Run(ctx)

Standalone runner concurrency defaults to one; backend consumers supply their effective message_channel_buffer default through WithConcurrency. The standalone default also applies when graceful shutdown is enabled. A terminal settlement or dead-letter handoff failure returns an error and cancels the subscription, leaving unresolved messages recoverable according to the backend. Handlers must respect context cancellation.

Following the Builder-IS-Implementation pattern, the value returned by New implements both Builder and Runner; Build validates the configuration and returns the receiver.

Index

funcAckWithRetry

go
func AckWithRetry(ctx context.Context, msg types.Message, logger *zap.Logger) error

AckWithRetry acknowledges msg, retrying transient failures with exponential backoff (3 attempts: 100/200/400ms). A message that is already acknowledged (types.ErrAlreadyAcknowledged) counts as success. logger may be nil.

funcNackWithRetry

go
func NackWithRetry(ctx context.Context, msg types.Message, logger *zap.Logger) error

NackWithRetry negatively acknowledges msg with the same retry behavior as AckWithRetry. logger may be nil.

typeBackOffFactory

BackOffFactory produces the backoff strategy used between reconnect attempts. A fresh strategy is created for every Run invocation.

go
type BackOffFactory func(ctx context.Context) backoff.BackOff

typeBuilder

Builder configures a Runner. Following the Builder-IS-Implementation pattern, the value returned by New implements both Builder and Runner; Build validates the configuration and returns the receiver.

go
type Builder interface {
    // WithName sets the backend name used for log scoping (e.g. "amqp").
    WithName(name string) Builder
    // WithLogger sets the logger. When unset, a no-op logger is used.
    WithLogger(logger *zap.Logger) Builder
    // WithConnect sets the connection callback. Required.
    WithConnect(fn ConnectFunc) Builder
    // WithHandler sets the single-message handler. Exactly one of
    // WithHandler or WithBatchHandler must be configured.
    WithHandler(fn types.HandlerFunc) Builder
    // WithBatchHandler sets the batch handler used together with WithBatch.
    WithBatchHandler(fn types.BatchHandlerFunc) Builder
    // WithBatch enables batch dispatch: messages accumulate until size is
    // reached or timeout elapses since the first message of the batch.
    WithBatch(size int, timeout time.Duration) Builder
    // WithConcurrency bounds simultaneous handlers or batches (default: 1).
    // Values greater than one allow out-of-order handler completion.
    WithConcurrency(n int) Builder
    // WithAutoReconnect enables reconnecting (via the connect callback) with
    // exponential backoff when the transport reports a lost connection.
    WithAutoReconnect(enabled bool) Builder
    // WithReconnectBackOff overrides the reconnect backoff strategy.
    WithReconnectBackOff(fn BackOffFactory) Builder
    // WithGracefulShutdown waits up to timeout for handlers on every terminal
    // path. Their contexts stay live during drain and are cancelled at its end.
    // Without graceful shutdown, handler cancellation is immediate.
    WithGracefulShutdown(enabled bool, timeout time.Duration) Builder
    // WithNackOnShutdown controls whether messages drained (but never handed
    // to a handler) during shutdown are negatively acknowledged. Backends
    // whose brokers redeliver automatically (visibility timeouts, offset
    // semantics) should leave this off.
    WithNackOnShutdown(enabled bool) Builder
    // WithDropDecision sets the retry-budget check applied before dispatch.
    WithDropDecision(fn DropDecisionFunc) Builder
    // WithOnDrop sets the hook invoked when a message is dropped for
    // exceeding its retry budget (e.g. dead-letter republish).
    WithOnDrop(fn DropFunc) Builder
    // Build validates the configuration and returns the Runner.
    Build() (Runner, error)
}

funcNew

go
func New() Builder

New returns a Builder for the shared consumer runner.

typeConnectFunc

ConnectFunc (re)establishes the backend connection and subscription. It is called once at startup and again on every reconnect attempt. It returns the message channel to consume from and the transport’s close-notification channel (types.Transport.ChannelClosed()).

go
type ConnectFunc func(ctx context.Context) (messages <-chan types.Message, closed chan error, err error)

typeDropDecisionFunc

DropDecisionFunc reports whether msg has exhausted its retry budget and must be dropped without processing. Backends encode their own counters (x-death/x-acquired-count, JetStream NumDelivered, SQS ApproximateReceiveCount, …).

go
type DropDecisionFunc func(msg types.Message) bool

typeDropFunc

DropFunc is invoked when a message is dropped for exceeding its retry budget, before the runner acknowledges it away. Backends plug dead-letter republishing in here. A returned error stops the runner without settling the original message, so the backend can redeliver it after recovery.

go
type DropFunc func(ctx context.Context, msg types.Message) error

typeRunner

Runner is the shared consumer processing loop.

go
type Runner interface {
    // Run blocks until ctx is cancelled (returns nil after any configured
    // graceful-shutdown drain) or an unrecoverable error occurs (initial
    // connect failure, reconnect budget exhausted, or connection loss with
    // auto-reconnect disabled).
    Run(ctx context.Context) error
}

Generated by gomarkdoc

↑ ↓ move↵ open/ opens search anywhere