messaging

package
v0.58.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 8, 2026 License: MIT Imports: 34 Imported by: 0

Documentation

Overview

Package messaging provides a unified interface for message queue operations. It abstracts the underlying messaging implementation to allow for easy testing and future extensibility.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrNotConnected is returned by a publish when the client is not connected to the broker.
	ErrNotConnected = errors.New("not connected to AMQP broker")
	// ErrShutdown is returned when a publish is interrupted by client shutdown. The outbox
	// relay treats this (and context.Canceled) as a shutdown abort that must NOT advance an
	// event's retry_count — it is the one publish error the relay branches on.
	ErrShutdown = errors.New("AMQP client is shutting down")
	// ErrPublishRetriesExhausted is returned by PublishToExchange once the bounded retry loop
	// reaches maxPublishAttempts. It wraps the last attempt's cause (one of ErrPublishNacked,
	// ErrPublishConfirmTimeout, or the raw publish error) so a caller can see WHY it gave up.
	ErrPublishRetriesExhausted = errors.New("amqp: publish retries exhausted")
	// ErrPublishNacked is the cause when the broker negatively acknowledged a publish. A
	// basic.nack on a publish-confirm is a transient broker condition (disk alarm, mirror
	// resync, failover), not a statement that the message is bad. These cause sentinels are
	// informational (logging / direct-publisher branching) — the outbox relay does NOT
	// classify on them (see ADR-033); it retries every publish failure.
	ErrPublishNacked = errors.New("amqp: publish nacked by broker")
	// ErrPublishConfirmTimeout is the cause when a confirmed publish never received an
	// ACK/NACK within connectionTimeout.
	ErrPublishConfirmTimeout = errors.New("amqp: publish confirmation timed out")
)
View Source 
var (
	// ErrPayloadUndecodable reports a message body that could not be decoded into
	// the consumer's payload type. Match it with errors.Is.
	ErrPayloadUndecodable = errors.New("messaging: payload could not be decoded")

	// ErrPayloadInvalid reports a payload that decoded but failed struct
	// validation. Match it with errors.Is.
	ErrPayloadInvalid = errors.New("messaging: payload failed validation")
)

Functions

func StartConsumeSpan added in v0.12.0 

func StartConsumeSpan(ctx context.Context, delivery *amqp.Delivery, queueName string) (context.Context, trace.Span)

StartConsumeSpan creates an OpenTelemetry span for message consumption. It extracts the trace context from the delivery headers and creates a child span. This should be called by consumers when processing messages. The returned context should be used for downstream operations, and the span must be ended when done.

Types

type AMQPClient 

type AMQPClient interface {
	Client

	// PublishToExchange publishes a message to a specific exchange with routing key.
	PublishToExchange(ctx context.Context, options PublishOptions, data []byte) error

	// ConsumeFromQueue consumes messages from a queue with specific options.
	ConsumeFromQueue(ctx context.Context, options ConsumeOptions) (<-chan amqp.Delivery, error)

	// DeclareQueue declares a queue from the given declaration.
	// ctx is checked before the broker operation (amqp091 declares are not context-aware on the wire).
	// The declaration's Args carries optional AMQP queue arguments (x-dead-letter-exchange, x-queue-type, ...).
	DeclareQueue(ctx context.Context, queue *QueueDeclaration) error

	// DeclareExchange declares an exchange from the given declaration (ctx: pre-flight check, as DeclareQueue).
	DeclareExchange(ctx context.Context, exchange *ExchangeDeclaration) error

	// BindQueue binds a queue to an exchange from the given declaration (ctx: pre-flight check, as DeclareQueue).
	BindQueue(ctx context.Context, binding *BindingDeclaration) error
}

AMQPClient extends the basic Client interface with AMQP-specific functionality. This allows for more advanced AMQP features while maintaining the simple interface.

type AMQPClientImpl 

type AMQPClientImpl struct {
	// contains filtered or unexported fields
}

AMQPClientImpl provides an AMQP implementation of the messaging client interface. It includes automatic reconnection, retry logic, and AMQP-specific features.

func NewAMQPClient 

func NewAMQPClient(brokerURL string, log logger.Logger, opts ...ClientOption) *AMQPClientImpl

NewAMQPClient creates a new AMQP client instance. It automatically attempts to connect to the broker and handles reconnections. Optional Option values (e.g. WithConnectionTimeout) override the defaults.

func (*AMQPClientImpl) BindQueue 

func (c *AMQPClientImpl) BindQueue(ctx context.Context, binding *BindingDeclaration) error

BindQueue binds a queue to an exchange from the given declaration (ctx: pre-flight check, see DeclareQueue).

func (*AMQPClientImpl) Close 

func (c *AMQPClientImpl) Close() error

Close gracefully shuts down the AMQP client.

func (*AMQPClientImpl) Consume 

func (c *AMQPClientImpl) Consume(ctx context.Context, destination string) (<-chan amqp.Delivery, error)

Consume starts consuming messages from the specified destination (queue name).

func (*AMQPClientImpl) ConsumeFromQueue 

func (c *AMQPClientImpl) ConsumeFromQueue(_ context.Context, options ConsumeOptions) (<-chan amqp.Delivery, error)

ConsumeFromQueue consumes messages from a queue with specific options.

func (*AMQPClientImpl) DeclareExchange 

func (c *AMQPClientImpl) DeclareExchange(ctx context.Context, exchange *ExchangeDeclaration) error

DeclareExchange declares an exchange from the given declaration (ctx: pre-flight check, see DeclareQueue).

func (*AMQPClientImpl) DeclareQueue 

func (c *AMQPClientImpl) DeclareQueue(ctx context.Context, queue *QueueDeclaration) error

DeclareQueue declares a queue from the given declaration. ctx is honored as a pre-flight check: amqp091 declare/bind operations are not context-aware on the wire, so a canceled context fails fast before the call.

func (*AMQPClientImpl) IsReady 

func (c *AMQPClientImpl) IsReady() bool

IsReady returns true if the client is connected and ready to send/receive messages.

func (*AMQPClientImpl) Publish 

func (c *AMQPClientImpl) Publish(ctx context.Context, destination string, data []byte) error

Publish sends a message to the specified destination (queue name). Uses default exchange ("") and destination as routing key.

func (*AMQPClientImpl) PublishToExchange 

func (c *AMQPClientImpl) PublishToExchange(ctx context.Context, options PublishOptions, data []byte) error

PublishToExchange publishes a message to a specific exchange with routing key.

type BindingDeclaration 

type BindingDeclaration struct {
	Queue      string         // Queue name
	Exchange   string         // Exchange name
	RoutingKey string         // Routing key pattern
	NoWait     bool           // Do not wait for server confirmation
	Args       map[string]any // Additional arguments
}

BindingDeclaration defines a queue-to-exchange binding

func NewBinding added in v0.14.1 

func NewBinding(queue, exchange, routingKey string) *BindingDeclaration

NewBinding creates a binding declaration between a queue and exchange.

Parameters:

  • queue: Queue name to bind
  • exchange: Exchange name to bind to
  • routingKey: Routing key pattern (e.g., "order.*", "user.created")

type BrokerURLProvider added in v0.26.0 

type BrokerURLProvider interface {
	// BrokerURL returns the AMQP broker URL for the given key.
	// For single-tenant apps, key will be "". For multi-tenant, key will be the tenant ID.
	BrokerURL(ctx context.Context, key string) (string, error)
}

BrokerURLProvider provides per-key AMQP configurations. This interface abstracts where tenant-specific messaging configs come from.

type Client 

type Client interface {
	// Publish sends a message to the specified destination.
	// destination can be a queue name, exchange, or topic depending on the implementation.
	// Returns an error if the publish operation fails.
	Publish(ctx context.Context, destination string, data []byte) error

	// Consume starts consuming messages from the specified destination.
	// Returns a channel that delivers messages and an error if consumption setup fails.
	// Messages should be acknowledged by the consumer.
	Consume(ctx context.Context, destination string) (<-chan amqp.Delivery, error)

	// Close gracefully shuts down the messaging client.
	// It should clean up all connections and resources.
	Close() error

	// IsReady returns true if the client is connected and ready to send/receive messages.
	IsReady() bool
}

Client defines the interface for messaging operations. It provides a simple API for publishing and consuming messages while hiding the complexity of connection management, retries, and protocol-specific details.

type ClientFactory added in v0.9.0 

type ClientFactory func(string, logger.Logger) AMQPClient

ClientFactory creates AMQP clients from URLs

type ClientOption added in v0.42.0 

type ClientOption func(*AMQPClientImpl)

ClientOption configures an AMQPClientImpl at construction time.

func WithConnectionTimeout added in v0.42.0 

func WithConnectionTimeout(d time.Duration) ClientOption

WithConnectionTimeout overrides the per-publish broker confirmation timeout — the wait for an ACK/NACK after a confirmed publish (see PublishToExchange). Non-positive values are ignored, leaving the 30s default in place.

func WithMaxPublishAttempts added in v0.45.0 

func WithMaxPublishAttempts(n int) ClientOption

WithMaxPublishAttempts bounds the per-publish retry loop: after n failed attempts PublishToExchange returns ErrPublishRetriesExhausted instead of retrying forever. Non-positive values are ignored, leaving the default (5).

func WithReadyTimeout added in v0.49.0 

func WithReadyTimeout(d time.Duration) ClientOption

WithReadyTimeout bounds PublishToExchange's pre-flight wait for a not-yet-ready client to become ready before the publish attempt begins (see waitForReady). The wait does not consume a maxPublishAttempts slot. Non-positive values are ignored, leaving the 5s default in place.

func WithReconnectDelay added in v0.49.0 

func WithReconnectDelay(d time.Duration) ClientOption

WithReconnectDelay overrides the base of the full-jitter reconnect backoff: each wait is a uniform random sample in [0, min(base*2^attempt, max)), so this is the first attempt's upper bound, not a minimum spacing between attempts. Non-positive values are ignored, leaving the 5s default in place.

func WithReconnectMaxDelay added in v0.49.0 

func WithReconnectMaxDelay(d time.Duration) ClientOption

WithReconnectMaxDelay overrides the ceiling of the connection-reconnect backoff. The consumer re-subscribe loop (registry.go) keeps its own fixed cap. Non-positive values are ignored, leaving the 60s default in place.

func WithReinitDelay added in v0.49.0 

func WithReinitDelay(d time.Duration) ClientOption

WithReinitDelay overrides the delay before channel reinitialization after failure. Non-positive values are ignored, leaving the 2s default in place.

func WithResendDelay added in v0.49.0 

func WithResendDelay(d time.Duration) ClientOption

WithResendDelay overrides the wait between retries after a channel-level publish error only; broker NACKs retry on the fixed 100ms nackBackoff and confirmation timeouts retry immediately. Non-positive values are ignored, leaving the 5s default in place.

type ConsumeOptions 

type ConsumeOptions struct {
	Queue         string // Queue name to consume from
	Consumer      string // Consumer tag
	AutoAck       bool   // Auto-acknowledge messages
	Exclusive     bool   // Exclusive consumer
	NoLocal       bool   // No-local flag
	NoWait        bool   // No-wait flag
	PrefetchCount int    // RabbitMQ prefetch count (0 or negative = default to 1)
}

ConsumeOptions contains options for consuming messages with AMQP-specific features.

type ConsumerDeclaration 

type ConsumerDeclaration struct {
	Queue         string         // Queue to consume from
	Consumer      string         // Consumer tag
	AutoAck       bool           // Automatically acknowledge messages
	Exclusive     bool           // Exclusive consumer
	NoLocal       bool           // Do not deliver to the connection that published
	NoWait        bool           // Do not wait for server confirmation
	EventType     string         // Event type identifier
	Description   string         // Human-readable description
	Handler       MessageHandler // Message handler (optional for documentation-only declarations)
	Workers       int            // Number of concurrent workers (0 = auto-scale to NumCPU*4, >0 = explicit)
	PrefetchCount int            // RabbitMQ prefetch count (0 = auto-scale to Workers*10, capped at 500)
}

ConsumerDeclaration defines what a module consumes and how to handle messages

func DeclareTypedConsumer added in v0.56.0 

func DeclareTypedConsumer[T any](decls *Declarations, opts *ConsumerOptions, fn func(context.Context, T) error) *ConsumerDeclaration

DeclareTypedConsumer registers a consumer whose handler is built by NewTypedHandler from fn, so T is inferred from fn and never spelled out.

The queue is not declared here: pass it to DeclareQueue or DeclareQueueWithDLQ separately, exactly as an untyped DeclareConsumer with a nil queue does. A consumer naming a queue nobody declared surfaces at Declarations.Validate() as "consumer references non-existent queue", not here.

It panics on a nil decls, a nil opts, or an opts that already carries a Handler — all three are declaration-time wiring mistakes, and the package already fails startup that way for duplicate consumer registrations.

Failure and concurrency semantics are NewTypedHandler's: decode and validation failures return a *PayloadError that nacks WITHOUT requeue, so pair the queue with DeclareQueueWithDLQ, and fn must be safe for concurrent use.

func DeclareTypedConsumerWithMeta added in v0.57.0 

func DeclareTypedConsumerWithMeta[T any](decls *Declarations, opts *ConsumerOptions, fn func(context.Context, T, Metadata) error) *ConsumerDeclaration

DeclareTypedConsumerWithMeta is DeclareTypedConsumer for a metadata-aware fn. Same panics, same queue rules, same PayloadError semantics.

func NewConsumer added in v0.14.1 

func NewConsumer(opts *ConsumerOptions) *ConsumerDeclaration

NewConsumer creates a consumer declaration from options.

type ConsumerOptions added in v0.14.1 

type ConsumerOptions struct {
	Queue         string         // Queue name to consume from
	Consumer      string         // Consumer tag
	EventType     string         // Event type identifier
	Description   string         // Human-readable description
	Handler       MessageHandler // Message handler (optional for documentation-only declarations)
	AutoAck       bool           // Automatically acknowledge messages (default: false)
	Exclusive     bool           // Exclusive consumer (default: false)
	NoLocal       bool           // Don't deliver to the connection that published (default: false)
	Workers       int            // Number of concurrent workers (0 = auto-scale to NumCPU*4, >0 = explicit)
	PrefetchCount int            // RabbitMQ prefetch count (0 = auto-scale to Workers*10, capped at 500)
}

ConsumerOptions contains configuration for creating a consumer declaration.

type DeadLetterSpec added in v0.53.0 

type DeadLetterSpec struct {
	// Exchange is the dead-letter exchange name. Empty derives "<queue>.dlx".
	// The exchange is declared as a durable fanout so the parking queue
	// receives every dead-lettered message regardless of routing key.
	Exchange string

	// ParkingQueue is the queue that collects dead-lettered messages.
	// Empty derives "<queue>.dlq". Declared with the production defaults of
	// NewQueue (durable, non-exclusive).
	ParkingQueue string

	// RoutingKey, when non-empty, is set as x-dead-letter-routing-key so
	// dead-lettered messages are re-published with it instead of their
	// original routing key. Rarely needed with the fanout DLX default.
	RoutingKey string
}

DeadLetterSpec configures the declarative dead-letter opt-in for a queue. Zero-value fields get derived defaults; see DeclareQueueWithDLQ.

type DeclarationStats added in v0.9.0 

type DeclarationStats struct {
	Exchanges  int
	Queues     int
	Bindings   int
	Publishers int
	Consumers  int
}

DeclarationStats holds counts of each declaration type.

type Declarations added in v0.9.0 

type Declarations struct {
	Exchanges  map[string]*ExchangeDeclaration
	Queues     map[string]*QueueDeclaration
	Bindings   []*BindingDeclaration
	Publishers []*PublisherDeclaration
	// contains filtered or unexported fields
}

Declarations stores messaging infrastructure declarations made by modules at startup. This is a pure data structure (no client dependencies) that can be validated once and replayed to multiple per-tenant registries.

func NewDeclarations added in v0.9.0 

func NewDeclarations() *Declarations

NewDeclarations creates a new empty declarations store.

func (*Declarations) Clone added in v0.9.0 

func (d *Declarations) Clone() *Declarations

Clone creates a deep copy of the declarations. This is useful for creating per-tenant copies during replay.

func (*Declarations) Consumers added in v0.9.0 

func (d *Declarations) Consumers() []*ConsumerDeclaration

Consumers returns all consumer declarations in registration order. Used internally by ReplayToRegistry and for observability/metrics.

func (*Declarations) DeclareBinding added in v0.14.1 

func (d *Declarations) DeclareBinding(queue, exchange, routingKey string) *BindingDeclaration

DeclareBinding creates and registers a binding in one step. Returns the created binding declaration for reference.

func (*Declarations) DeclareConsumer added in v0.14.1 

func (d *Declarations) DeclareConsumer(opts *ConsumerOptions, queue *QueueDeclaration) *ConsumerDeclaration

DeclareConsumer creates and registers a consumer in one step.

A non-nil queue is registered, merging with any existing declaration of the same name; an incompatible shape keeps the incumbent and becomes a startup conflict (see RegisterQueue). This hybrid approach allows consumers to optionally declare their dependencies.

Usage:

  • Pass nil if queue is already registered separately
  • Pass queue declaration to auto-register (convenience for simple cases)

func (*Declarations) DeclarePublisher added in v0.14.1 

func (d *Declarations) DeclarePublisher(opts *PublisherOptions, exchange *ExchangeDeclaration) *PublisherDeclaration

DeclarePublisher creates and registers a publisher in one step.

If exchange is non-nil and not already registered, it will be automatically registered. This hybrid approach allows publishers to optionally declare their dependencies.

Usage:

  • Pass nil if exchange is already registered separately
  • Pass exchange declaration to auto-register (convenience for simple cases)

func (*Declarations) DeclareQueue added in v0.14.1 

func (d *Declarations) DeclareQueue(name string) *QueueDeclaration

DeclareQueue creates and registers a queue in one step. Returns the created queue declaration for reference.

func (*Declarations) DeclareQueueWithDLQ added in v0.53.0 

func (d *Declarations) DeclareQueueWithDLQ(name string, dl *DeadLetterSpec) *QueueDeclaration

DeclareQueueWithDLQ declares a queue whose failed deliveries are parked instead of dropped: the framework's nack-without-requeue on handler error (see wiki/messaging.md) dead-letters into the spec's exchange, and the parking queue bound to it retains the message with the x-death header. Lowers to ordinary exchange/queue/binding declarations plus queue Args, so per-tenant replay, validation, and topology hashing behave as if declared by hand. Returns the primary queue declaration.

func (*Declarations) DeclareTopicExchange added in v0.14.1 

func (d *Declarations) DeclareTopicExchange(name string) *ExchangeDeclaration

DeclareTopicExchange creates and registers a topic exchange in one step. Returns the created exchange declaration for reference.

func (*Declarations) Hash added in v0.15.0 

func (d *Declarations) Hash() uint64

Hash generates a deterministic hash of all declarations. This is used by Manager to detect duplicate replay attempts (idempotency). The hash is stable across multiple calls and does not include handler functions.

func (*Declarations) IsEmpty added in v0.30.0 

func (d *Declarations) IsEmpty() bool

IsEmpty reports whether no declarations have been registered.

func (*Declarations) RegisterBinding added in v0.9.0 

func (d *Declarations) RegisterBinding(b *BindingDeclaration)

RegisterBinding adds a binding declaration to the store.

func (*Declarations) RegisterConsumer added in v0.9.0 

func (d *Declarations) RegisterConsumer(c *ConsumerDeclaration)

RegisterConsumer adds a consumer declaration to the store. Panics if a consumer with the same queue+consumer+event_type already exists.

func (*Declarations) RegisterExchange added in v0.9.0 

func (d *Declarations) RegisterExchange(e *ExchangeDeclaration)

RegisterExchange adds an exchange declaration to the store.

Re-declaring a name keeps the last declaration, unlike RegisterQueue: no framework helper builds a divergent exchange shape for a name a caller would also declare by hand. The one to watch is DeclareQueueWithDLQ's fanout DLX, which several primary queues may share — today every repeat is identical.

func (*Declarations) RegisterPublisher added in v0.9.0 

func (d *Declarations) RegisterPublisher(p *PublisherDeclaration)

RegisterPublisher adds a publisher declaration to the store.

func (*Declarations) RegisterQueue added in v0.9.0 

func (d *Declarations) RegisterQueue(q *QueueDeclaration)

RegisterQueue adds a queue declaration to the store.

Re-declaring a name merges when the shapes are compatible, so DeclareQueueWithDLQ and DeclareQueue on one name compose instead of whichever ran last silently dropping the other's dead-letter args. An incompatible re-declaration keeps the incumbent and records a conflict that Validate reports, because declaration order across modules is invisible at any single call site. That report prints the contested Args values verbatim into a startup error, which the logger's key-based filter cannot mask — queue Args are broker topology and must not carry secrets.

func (*Declarations) ReplayToRegistry added in v0.9.0 

func (d *Declarations) ReplayToRegistry(reg RegistryInterface) error

ReplayToRegistry applies all declarations to a runtime registry. The order is important: exchanges first, then queues, then bindings, then publishers/consumers.

func (*Declarations) Stats added in v0.9.0 

func (d *Declarations) Stats() DeclarationStats

Stats returns counts of each declaration type.

func (*Declarations) Validate added in v0.9.0 

func (d *Declarations) Validate() error

Validate checks the integrity of all declarations. It ensures that references between declarations are valid.

type ExchangeDeclaration 

type ExchangeDeclaration struct {
	Name       string         // Exchange name
	Type       string         // Exchange type (direct, topic, fanout, headers)
	Durable    bool           // Survive server restart
	AutoDelete bool           // Delete when no longer used
	Internal   bool           // Internal exchange
	NoWait     bool           // Do not wait for server confirmation
	Args       map[string]any // Additional arguments
}

ExchangeDeclaration defines an exchange to be declared

func NewTopicExchange added in v0.14.1 

func NewTopicExchange(name string) *ExchangeDeclaration

NewTopicExchange creates a topic exchange with production-safe defaults. Topic exchanges route messages based on routing key patterns (e.g., "order.*", "user.#").

Production defaults:

  • Durable: true (survives broker restart)
  • AutoDelete: false (won't delete when unused)
  • Internal: false (can be published to directly)
  • NoWait: false (waits for broker confirmation)

type Manager added in v0.9.0 

type Manager struct {
	// contains filtered or unexported fields
}

Manager manages AMQP clients by string keys with different lifecycle strategies. Publishers are cached with idle eviction (can be recreated easily). Consumers are long-lived (must stay alive to receive messages). The manager is key-agnostic - it doesn't know about tenants, just manages named clients.

The publisher side is a thin adapter over internal/resourcepool.Pool, which owns the ADR-032 lease/evict/close protocol (seed leases, LRU eviction, idle cleanup, and the closed-pool guard). The manager keeps only the AMQP-specific client creation. The consumer side is long-lived and managed directly (not pool-shaped).

func NewMessagingManager added in v0.9.0 

func NewMessagingManager(resourceSource BrokerURLProvider, log logger.Logger, opts ManagerOptions, clientFactory ClientFactory) *Manager

NewMessagingManager creates a new messaging manager

func (*Manager) Close added in v0.9.0 

func (m *Manager) Close() error

Close closes all clients and stops cleanup. Publisher closes go through the pool (which stops its own cleanup loop and joins every per-publisher close failure); consumer closes are handled directly. Every failure from BOTH sides is surfaced under the historical "errors closing messaging clients" prefix.

func (*Manager) EnsureConsumers added in v0.9.0 

func (m *Manager) EnsureConsumers(ctx context.Context, key string, decls *Declarations) error

EnsureConsumers creates and starts consumers for the given key using the provided declarations. This should be called once per key to set up long-lived consumers. Subsequent calls for the same key are idempotent.

Concurrent calls for the same key collapse onto one setup pass, but each caller waits on ITS OWN context. DoChan (not Do) is what makes that possible: Do blocks uncancelably, so a caller whose budget was already spent still sat through the full setup — bounded by infraSetupTimeout (45s) while the realistic caller is a lazy first-touch request carrying a ~5s deadline. Giving up does NOT cancel the setup: ensureConsumersInternal runs on a WithoutCancel budget, so it completes and installs the consumers for whoever asks next.

One testing caveat: singleflight neither recovers nor forwards a runtime.Goexit from the shared call, so a test double that calls t.Fatal or require.* inside the setup path hangs every waiter. Use t.Errorf and return, as the repo's httptest handlers do.

func (*Manager) Publisher added in v0.19.0 

func (m *Manager) Publisher(ctx context.Context, key string) (AMQPClient, ReleaseFunc, error)

Publisher returns a publisher client for the given key plus a ReleaseFunc the caller must invoke when finished with it for the current unit of work (typically deferred). Publishers are cached with LRU eviction and lazy initialization; the lease prevents a publisher that is evicted while in use from being closed under an active caller (the #606 race). Once Close has run, Publisher fails closed rather than resurrecting a publisher (F22). On error the returned ReleaseFunc is nil — check err first.

func (*Manager) StartCleanup added in v0.9.0 

func (m *Manager) StartCleanup(interval time.Duration)

StartCleanup starts the background cleanup routine for idle publishers. A non-positive interval substitutes the documented 2-minute default.

func (*Manager) Stats added in v0.9.0 

func (m *Manager) Stats() map[string]any

Stats returns statistics about the messaging manager. Publisher counters come from the pool; active_consumers comes from the directly-managed consumer map.

func (*Manager) StopCleanup added in v0.9.0 

func (m *Manager) StopCleanup()

StopCleanup stops the background cleanup routine

func (*Manager) StopConsumers added in v0.42.0 

func (m *Manager) StopConsumers()

StopConsumers stops every consumer registry from accepting new messages (canceling their consume contexts) WITHOUT closing the underlying AMQP connections — Close does that. The framework calls this during shutdown before tearing down modules so it stops delivering fresh messages to modules that are about to shut down. Cancellation propagates to in-flight handlers via their context, but they are not synchronously joined here. Idempotent: Registry.StopConsumers guards on its active flag, so a subsequent Close (which also stops consumers) is safe.

type ManagerOptions added in v0.9.0 

type ManagerOptions struct {
	MaxPublishers int           // Maximum number of publisher clients to keep cached
	IdleTTL       time.Duration // Time after which idle publishers are evicted
	// ConnectionTimeout is the per-publish broker confirmation timeout applied to
	// clients created by the default factory. Zero leaves the client default (30s).
	ConnectionTimeout time.Duration
	// MaxPublishAttempts bounds the per-publish retry loop for clients created by the
	// default factory. Zero (or negative) leaves the client default (5).
	MaxPublishAttempts int
	// ReadyTimeout bounds the pre-flight readiness wait for clients created by the
	// default factory. Zero (or negative) leaves the client default (5s).
	ReadyTimeout time.Duration
	// Reconnect delays for clients created by the default factory. Zero (or negative)
	// leaves the client defaults (5s/60s/2s/5s). See the WithReconnect*/WithResendDelay
	// option docs for each knob's exact scope (jitter semantics, publish-error-only).
	ReconnectDelay    time.Duration
	ReconnectMaxDelay time.Duration
	ReinitDelay       time.Duration
	ResendDelay       time.Duration
}

ManagerOptions configures the Manager

type MessageHandler 

type MessageHandler interface {
	// Handle processes a message and returns an error if processing fails.
	// If an error is returned, the message will be negatively acknowledged (nack).
	// If no error is returned, the message will be acknowledged (ack).
	// The delivery is passed by pointer for performance reasons.
	Handle(ctx context.Context, delivery *amqp.Delivery) error

	// EventType returns the event type this handler can process.
	// This is used for routing messages to the correct handler.
	EventType() string
}

MessageHandler defines the interface for processing consumed messages. Handlers should implement this interface to process specific message types.

func NewTypedHandler added in v0.56.0 

func NewTypedHandler[T any](eventType string, fn func(context.Context, T) error) MessageHandler

NewTypedHandler adapts a typed function to the MessageHandler contract: decode (JSON) → validate (go-playground struct tags) → fn. It is the consumer mirror of the typed HTTP handlers registered with server.POST.

Failures short-circuit with a *PayloadError, which the worker loop nacks WITHOUT requeue like any other handler error — decode and validation failures are not retryable, so pair the queue with DeclareQueueWithDLQ to park them. Match them with errors.Is against ErrPayloadUndecodable or ErrPayloadInvalid. fn's own error is returned unwrapped, so a consumer's errors.Is against its business sentinels still works.

The returned handler holds no mutable state and is safe to share across workers and tenants. fn must therefore be safe for concurrent use too.

func NewTypedHandlerWithMeta added in v0.57.0 

func NewTypedHandlerWithMeta[T any](eventType string, fn func(context.Context, T, Metadata) error) MessageHandler

NewTypedHandlerWithMeta is NewTypedHandler for consumers that also need delivery metadata — the outbox-dedup shape: read the x-outbox-event-id header via outbox.EventIDFromHeaders(meta.Headers()) and wrap the business logic in inbox.ProcessOnce. Failure and concurrency semantics are identical to NewTypedHandler; fn must be safe for concurrent use.

type Metadata added in v0.57.0 

type Metadata struct {
	// contains filtered or unexported fields
}

Metadata exposes read-only delivery facts to a typed consumer without widening fn's payload contract. It is a per-delivery value constructed by the adapter AFTER the nil-delivery guard; the zero value is inert (every accessor returns its zero). Headers returns the live amqp.Table — treat it as read-only, it is shared with the worker loop.

func (Metadata) EventType added in v0.57.0 

func (m Metadata) EventType() string

EventType returns the wire-level message type stamped by the publisher.

func (Metadata) Headers added in v0.57.0 

func (m Metadata) Headers() amqp.Table

Headers returns the AMQP delivery headers (e.g. for outbox.EventIDFromHeaders). Nil when no headers were published. The table is the delivery's own, not a copy — read it, do not mutate it. Values are publisher-controlled, so treat them as untrusted input on a queue fed from outside this service.

func (Metadata) Redelivered added in v0.57.0 

func (m Metadata) Redelivered() bool

Redelivered reports the broker's redelivery flag.

type PayloadError added in v0.56.0 

type PayloadError struct {
	// EventType is the declared consumer event type the body was routed to.
	EventType string

	// Stage is where the failure happened. It is exported to label logs and
	// metrics with "decode" or "validate"; for control flow, match errors.Is
	// against ErrPayloadUndecodable or ErrPayloadInvalid instead.
	Stage PayloadStage
	// contains filtered or unexported fields
}

PayloadError describes why a message body could not be turned into a typed payload.

SECURITY: AMQP bodies are partner PII/PCI data, so the framework's own rendering must stay free of them. Error() and Fields() are safe to log; Unwrap() is not. Error() composes its text from schema facts only — it never renders the wrapped cause verbatim, because every producer in reach echoes payload bytes in at least one shape:

  • json.UnmarshalTypeError.Value carries the raw literal ("number 1234.56") and, for integer-keyed maps, the raw key.
  • json.SyntaxError quotes the offending payload byte.
  • json.Decoder.DisallowUnknownFields reports the partner-supplied key verbatim.
  • validator namespaces interpolate map keys verbatim ("Limits[4111...]"), which is why the namespace list is unexported and redacted on read.

The rendering itself lives on the codec seam (codec.summarize), so a new codec (issue #346) must supply its own audited phrasing; until it does, the constructor substitutes the fail-closed phrase and the cause itself is never rendered.

func (*PayloadError) Error added in v0.56.0 

func (e *PayloadError) Error() string

func (*PayloadError) Fields added in v0.56.0 

func (e *PayloadError) Fields() []string

Fields returns the validator field namespaces that failed, e.g. ["CreateReq.Amount"]. It is empty for decode failures and for a nil receiver.

SECURITY: the bracketed span is redacted to [*] on the way out, and the result is a fresh slice, so the redaction survives whatever the caller does with it. This is the only read path onto the namespaces.

func (*PayloadError) Is added in v0.56.0 

func (e *PayloadError) Is(target error) bool

Is maps the stage onto its sentinel so consumers can discriminate the two failure modes without reading Stage.

func (*PayloadError) Unwrap added in v0.56.0 

func (e *PayloadError) Unwrap() error

Unwrap exposes the underlying decode or validation error so errors.As can reach the cause.

SECURITY: the returned error MAY carry payload-derived text — a rejected numeric literal, an offending byte, an unknown key, a map key. It is the deliberate escape hatch for a caller that needs the raw diagnostic; logging it is opt-in and on the caller.

type PayloadStage added in v0.56.0 

type PayloadStage string

PayloadStage names the half of the typed-payload pipeline that failed. 

const (
	PayloadStageDecode   PayloadStage = "decode"
	PayloadStageValidate PayloadStage = "validate"
)

Stage values carried by PayloadError.Stage. A PayloadError whose Stage is none of these matches neither sentinel.

type PublishOptions 

type PublishOptions struct {
	Exchange   string         // AMQP exchange name
	RoutingKey string         // AMQP routing key
	Headers    map[string]any // Message headers
	Mandatory  bool           // AMQP mandatory flag
	Immediate  bool           // AMQP immediate flag
}

PublishOptions contains options for publishing messages with AMQP-specific features.

type PublisherDeclaration 

type PublisherDeclaration struct {
	Exchange    string         // Target exchange
	RoutingKey  string         // Default routing key
	EventType   string         // Event type identifier
	Description string         // Human-readable description
	Mandatory   bool           // Message must be routed to a queue
	Immediate   bool           // Message must be delivered immediately
	Headers     map[string]any // Default headers
}

PublisherDeclaration defines what a module publishes

func NewPublisher added in v0.14.1 

func NewPublisher(opts *PublisherOptions) *PublisherDeclaration

NewPublisher creates a publisher declaration from options. If Headers is nil, an empty map is created.

type PublisherOptions added in v0.14.1 

type PublisherOptions struct {
	Exchange    string         // Target exchange name
	RoutingKey  string         // Routing key for messages
	EventType   string         // Event type identifier
	Description string         // Human-readable description
	Headers     map[string]any // Default headers (optional)
	Mandatory   bool           // Message must be routed to a queue (default: false)
	Immediate   bool           // Message must be delivered immediately (default: false)
}

PublisherOptions contains configuration for creating a publisher declaration.

type QueueDeclaration 

type QueueDeclaration struct {
	Name       string         // Queue name
	Durable    bool           // Survive server restart
	AutoDelete bool           // Delete when no consumers
	Exclusive  bool           // Only accessible by declaring connection
	NoWait     bool           // Do not wait for server confirmation
	Args       map[string]any // Additional arguments
}

QueueDeclaration defines a queue to be declared

func NewQueue added in v0.14.1 

func NewQueue(name string) *QueueDeclaration

NewQueue creates a queue with production-safe defaults.

Production defaults:

  • Durable: true (survives broker restart)
  • AutoDelete: false (won't delete when consumers disconnect)
  • Exclusive: false (can be accessed by multiple connections)
  • NoWait: false (waits for broker confirmation)

type Registry 

type Registry struct {
	// contains filtered or unexported fields
}

Registry manages messaging infrastructure declarations across modules. It ensures queues, exchanges, and bindings are properly declared before use. It also manages consumer lifecycle and handles message routing to handlers.

func NewRegistry 

func NewRegistry(client AMQPClient, log logger.Logger) *Registry

NewRegistry creates a new messaging registry

func (*Registry) Bindings added in v0.19.0 

func (r *Registry) Bindings() []*BindingDeclaration

Bindings returns all registered bindings (for testing/monitoring)

func (*Registry) Consumers added in v0.19.0 

func (r *Registry) Consumers() []*ConsumerDeclaration

Consumers returns all registered consumers (for documentation/monitoring)

func (*Registry) DeclareInfrastructure 

func (r *Registry) DeclareInfrastructure(ctx context.Context) error

DeclareInfrastructure declares all registered messaging infrastructure

func (*Registry) Exchanges added in v0.19.0 

func (r *Registry) Exchanges() map[string]*ExchangeDeclaration

Exchanges returns all registered exchanges (for testing/monitoring)

func (*Registry) Publishers added in v0.19.0 

func (r *Registry) Publishers() []*PublisherDeclaration

Publishers returns all registered publishers (for documentation/monitoring)

func (*Registry) Queues added in v0.19.0 

func (r *Registry) Queues() map[string]*QueueDeclaration

Queues returns all registered queues (for testing/monitoring)

func (*Registry) RegisterBinding 

func (r *Registry) RegisterBinding(declaration *BindingDeclaration)

RegisterBinding registers a binding for declaration

func (*Registry) RegisterConsumer 

func (r *Registry) RegisterConsumer(declaration *ConsumerDeclaration)

RegisterConsumer registers a consumer declaration

func (*Registry) RegisterExchange 

func (r *Registry) RegisterExchange(declaration *ExchangeDeclaration)

RegisterExchange registers an exchange for declaration

func (*Registry) RegisterPublisher 

func (r *Registry) RegisterPublisher(declaration *PublisherDeclaration)

RegisterPublisher registers a publisher declaration

func (*Registry) RegisterQueue 

func (r *Registry) RegisterQueue(declaration *QueueDeclaration)

RegisterQueue registers a queue for declaration

func (*Registry) StartConsumers 

func (r *Registry) StartConsumers(ctx context.Context) error

StartConsumers starts all registered consumers with handlers. This should be called after DeclareInfrastructure and before starting the main application.

func (*Registry) StopConsumers 

func (r *Registry) StopConsumers()

StopConsumers gracefully stops all running consumers.

func (*Registry) ValidateConsumer 

func (r *Registry) ValidateConsumer(queue string) bool

ValidateConsumer checks if a consumer is registered for the given queue

func (*Registry) ValidatePublisher 

func (r *Registry) ValidatePublisher(exchange, routingKey string) bool

ValidatePublisher checks if a publisher is registered for the given exchange/routing key

type RegistryInterface added in v0.8.0 

type RegistryInterface interface {
	// Registration methods
	RegisterExchange(declaration *ExchangeDeclaration)
	RegisterQueue(declaration *QueueDeclaration)
	RegisterBinding(declaration *BindingDeclaration)
	RegisterPublisher(declaration *PublisherDeclaration)
	RegisterConsumer(declaration *ConsumerDeclaration)

	// Infrastructure lifecycle
	DeclareInfrastructure(ctx context.Context) error
	StartConsumers(ctx context.Context) error
	StopConsumers()

	// Accessor methods for testing/monitoring
	Exchanges() map[string]*ExchangeDeclaration
	Queues() map[string]*QueueDeclaration
	Bindings() []*BindingDeclaration
	Publishers() []*PublisherDeclaration
	Consumers() []*ConsumerDeclaration

	// Validation methods
	ValidatePublisher(exchange, routingKey string) bool
	ValidateConsumer(queue string) bool
}

RegistryInterface defines the contract for messaging infrastructure management. This interface allows for easy mocking and testing of messaging infrastructure.

type ReleaseFunc added in v0.43.0 

type ReleaseFunc func()

ReleaseFunc releases a lease obtained from Publisher. Callers must invoke it (typically deferred) when finished with the publisher for the current unit of work. It is idempotent. Release does NOT close the shared publisher; it signals this borrower is done, so a publisher evicted while leased is closed only once its last lease is released. See ADR-032.

Source Files

View all Source files

Directories

Path Synopsis
internal
tracking

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.