babelqueue

Nov 1, 2026

Changes in this version

type App

+ func (a *App) DefaultQueue() string

+ func (a *App) PublishEnvelopeWithHeaders(ctx context.Context, urn string, data map[string]any, ...) (Envelope, error)

type AppOption

+ func WithDeadLetterObserver(fn DeadLetterObserver) AppOption

+ type DeadLetterEvent struct

+ DeadLetterQueue string

+ Queue string

+ Reason string

+ type DeadLetterObserver func(ctx context.Context, event DeadLetterEvent)

+ type HeaderRepublisher interface

+ PublishFromWithHeaders func(ctx context.Context, queue, body string, from *ReceivedMessage, ...) error

+ type NativeDeadLetterSettler interface

+ SettleNativeDeadLetter func(ctx context.Context, msg *ReceivedMessage, reason, description string) (deadLettered bool, err error)

Oct 26, 2026

Changes in this version

+ type DeadLetterNamer interface

+ DeadLetterQueueName func(queue, suffix string) string

+ type NativeSettler interface

+ SettleNative func(ctx context.Context, msg *ReceivedMessage, reason, description string) error

Oct 24, 2026

Changes in this version

type AppOption

+ func WithRetryBackoffMax(d time.Duration) AppOption

+ func WithRetryBackoffMultiplier(m float64) AppOption

+ func WithRetryJitter(j RetryJitter) AppOption

+ type DeadLetterer interface

+ DeadLetter func(ctx context.Context, msg *ReceivedMessage, dlqQueue, body string) error

type ReceivedMessage

+ Poison bool

+ type RetryJitter string

+ const JitterFull

+ const JitterNone

Oct 20, 2026

Changes in this version

+ var ErrNoIdentity = errors.New("babelqueue: envelope has no valid job or urn identity")

Oct 17, 2026

Changes in this version

+ var ErrDuplicateMetaKey = errors.New("babelqueue: meta key supplied more than once")

+ var ErrEmptyMetaKey = errors.New("babelqueue: meta key must not be empty")

+ var ErrForbiddenKey = errors.New("babelqueue: forbidden non-canonical envelope key (message-envelope.md §10)")

+ var ErrReservedMetaKey = errors.New("babelqueue: meta key collides with a core meta key")

type Option

+ func WithMeta(key string, value any) Option

Oct 12, 2026

Changes in this version

type DeadLetter

+ func (d *DeadLetter) UnmarshalJSON(raw []byte) error

type Meta

+ func (m *Meta) UnmarshalJSON(raw []byte) error

Oct 11, 2026

Changes in this version

+ type KeyedRepublisher interface

+ PublishFrom func(ctx context.Context, queue, body string, from *ReceivedMessage) error

Oct 10, 2026

Changes in this version

+ func TruncateDeclaredVersion(v string) string

type AppOption

+ func WithDeadLetterErrorHandler(fn func(ctx context.Context, msg *ReceivedMessage, err error)) AppOption

+ func WithRewindPause(d time.Duration) AppOption

+ type Rewinder interface

+ Rewind func(ctx context.Context, msg *ReceivedMessage) error

Oct 10, 2026

Changes in this version

type ReceivedMessage

+ RouteURN string

+ type RoutedPublisher interface

+ PublishRouted func(ctx context.Context, queue, body, routeURN string) error

Oct 3, 2026

Changes in this version

+ var ErrReleaseUnsupported = errors.New("babelqueue: in-place release is unsupported for this message")

type AppOption

+ func WithAckErrorHandler(fn func(ctx context.Context, msg *ReceivedMessage, err error)) AppOption

+ func WithReleaseErrorHandler(fn func(ctx context.Context, msg *ReceivedMessage, err error)) AppOption

+ func WithRetryBackoff(d time.Duration) AppOption

+ func WithUnknownURNReleaseDelay(d time.Duration) AppOption

type Envelope

+ func (e *Envelope) UnmarshalJSON(raw []byte) error

+ func (e Envelope) MarshalJSON() ([]byte, error)

+ func (e Envelope) Warnings() []string

type ReceivedMessage

+ DeliveryCount int

+ type Releaser interface

+ Release func(ctx context.Context, msg *ReceivedMessage, delay time.Duration) error

Jun 20, 2026

Changes in this version

+ func HeadersFromContext(ctx context.Context) map[string]string

type App

+ func (a *App) PublishWithHeaders(ctx context.Context, urn string, data map[string]any, ...) (string, error)

Jun 19, 2026

Changes in this version

+ const HeaderReplayBypass

+ func BypassExternalEffects(ctx context.Context, fn func() error) error

+ func IsReplay(ctx context.Context) bool

+ type HeaderPublisher interface

+ PublishWithHeaders func(ctx context.Context, queue, body string, headers map[string]string) error

type InMemoryTransport

+ func (t *InMemoryTransport) PublishWithHeaders(_ context.Context, queue, body string, headers map[string]string) error

type ReceivedMessage

+ Headers map[string]string

type RedriveItem

+ Bypassed bool

type RedriveOptions

+ Bypass bool

Jun 19, 2026

Changes in this version

+ type RedriveItem struct

+ From string

+ MessageID string

+ Reason string

+ Redriven bool

+ To string

+ TraceID string

+ URN string

+ type RedriveOptions struct

+ DryRun bool

+ Max int

+ Select func(Envelope) bool

+ Timeout time.Duration

+ ToQueue string

+ type RedriveResult struct

+ Items []RedriveItem

+ Redriven int

+ Skipped int

+ func Redrive(ctx context.Context, t Transport, dlq string, opts RedriveOptions) (RedriveResult, error)

Jun 6, 2026

Changes in this version

+ type App struct

+ func NewApp(transport Transport, opts ...AppOption) *App

+ func (a *App) Consume(ctx context.Context, queue ...string) error

+ func (a *App) Drain(ctx context.Context, queue string, max int) (int, error)

+ func (a *App) Handle(urn string, handler Handler)

+ func (a *App) Publish(ctx context.Context, urn string, data map[string]any, opts ...Option) (string, error)

+ func (a *App) Run(ctx context.Context) error

+ type AppOption func(*App)

+ func WithDeadLetter(enabled bool) AppOption

+ func WithDeadLetterQueue(queue string) AppOption

+ func WithDeadLetterSuffix(suffix string) AppOption

+ func WithDefaultQueue(queue string) AppOption

+ func WithMaxAttempts(n int) AppOption

+ func WithPollTimeout(d time.Duration) AppOption

+ func WithUnknownURNStrategy(strategy string) AppOption

+ type Handler func(ctx context.Context, env Envelope) error

+ type InMemoryTransport struct

+ func NewInMemoryTransport() *InMemoryTransport

+ func (t *InMemoryTransport) Ack(_ context.Context, _ *ReceivedMessage) error

+ func (t *InMemoryTransport) Pop(_ context.Context, queue string, _ time.Duration) (*ReceivedMessage, error)

+ func (t *InMemoryTransport) Publish(_ context.Context, queue, body string) error

+ func (t *InMemoryTransport) Size(queue string) int

+ type ReceivedMessage struct

+ Body string

+ Handle any

+ Queue string

+ type Transport interface

+ Ack func(ctx context.Context, msg *ReceivedMessage) error

+ Pop func(ctx context.Context, queue string, timeout time.Duration) (*ReceivedMessage, error)

+ Publish func(ctx context.Context, queue, body string) error

Jun 6, 2026

Changes in this version

+ const SchemaVersion

+ const SourceLang

+ const StrategyDeadLetter

+ const StrategyDelete

+ const StrategyFail

+ const StrategyRelease

+ var ErrEmptyURN = errors.New("babelqueue: a polyglot message must expose a stable, non-empty URN")

+ var ErrUnknownURN = errors.New("babelqueue: no handler is mapped for the message URN")

+ type DeadLetter struct

+ Attempts int

+ Error *string

+ Exception *string

+ FailedAt int64

+ Lang string

+ OriginalQueue string

+ Reason string

+ type Envelope struct

+ Attempts int

+ Data map[string]any

+ DeadLetter *DeadLetter

+ Job string

+ Meta Meta

+ TraceID string

+ func Annotate(env Envelope, reason, originalQueue string, attempts int, cause error) Envelope

+ func Decode(raw []byte) (Envelope, error)

+ func Make(urn string, data map[string]any, opts ...Option) (Envelope, error)

+ func (e Envelope) Accepts() bool

+ func (e Envelope) Encode() ([]byte, error)

+ func (e Envelope) URN() string

+ type Meta struct

+ CreatedAt int64

+ ID string

+ Lang string

+ Queue string

+ SchemaVersion int

+ type Option func(*makeConfig)

+ func WithQueue(queue string) Option

+ func WithTraceID(traceID string) Option