messaging

package
v0.63.0 Latest Latest
Warning

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

 Go to latest Published: Sep 5, 2026 License: MIT Imports: 40 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

View Source 
const (
	SealTenancyDisabled  = sealruntime.TenancyDisabled
	SealTenancyShared    = sealruntime.TenancyShared
	SealTenancyPerTenant = sealruntime.TenancyPerTenant
)
View Source 
const HeaderEventID = "x-outbox-event-id"

HeaderEventID is the AMQP header the outbox relay stamps with the event id a consumer dedups on. outbox.HeaderEventID aliases it; the constant lives here because Metadata.DedupKey reads it and outbox imports this package.

View Source 
const SealTagName = "seal"

SealTagName is the struct-tag key of the sealing family. Spelled here rather than imported from jose/sealed: the probe must not pull the codec into every build, and jose must not import messaging. messaging/sealed pins the two spellings together.

View Source 
const TenantStampHeader = tenantstamp.Header

TenantStampHeader is the header the framework writes the publishing context's tenant into, and the one a consumer reads it back from.

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 publishBytes 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")

	// ErrPayloadOpenRefused reports a sealed message the opener refused —
	// poison, like the two above; a provisioning-recoverable refusal
	// (SEAL_KID_UNKNOWN_GENERATION) still takes this path and keeps its code.
	// Match it with errors.Is.
	ErrPayloadOpenRefused = errors.New("messaging: sealed payload refused by the opener")
)
View Source 
var ErrInvalidEventID = errors.New("messaging: event id is outside the ledger grammar [A-Za-z0-9_-]{1,128}")

ErrInvalidEventID is returned when an event id headed for the inbox ledger is outside ^[A-Za-z0-9_-]{1,128}$ — absent, empty, over 128 bytes, or carrying any other byte. Match it with errors.Is. The wrapped message names the byte LENGTH, never the id: the value is publisher-controlled and this error reaches logs and spans. A handler returning it takes the standard poison path (nack without requeue → DLQ); the remedy is to re-mint conforming ids or to move the producer to the sealed typed door.

View Source 
var ErrInvalidPublishDestination = errors.New("amqp: publish destination exceeds the AMQP shortstr limit")

ErrInvalidPublishDestination is returned when a caller-supplied field that the AMQP wire format carries as a shortstr cannot fit one. Two doors return it: Publish/PublishToExchange, for the exchange and routing key of the basic.publish METHOD frame and the header keys of the CONTENT-HEADER frame beside it; and Declarations.Validate, for a declared name or routing key, which fails startup rather than the first publish. Match it with errors.Is; the wrapped message names the FIELD and its byte length, never the value — an over-long destination is usually built from request data, and this error reaches logs and spans.

The publish is refused outright rather than retried: the frame is unwritable whatever the broker's state, so a retry only re-tears the connection it just brought back. ErrPublishRetriesExhausted is deliberately NOT involved.

View Source 
var ErrManagerClosed = errors.New("messaging: manager closed")

ErrManagerClosed is returned by Manager's EnsureConsumers and Publisher methods once Close has been called, rather than resurrecting a consumer or publisher on a shut-down manager (backlog F22). Publisher additionally returns it from a zero-value Manager that was never built via NewMessagingManager. Callers can use errors.Is(err, ErrManagerClosed) to distinguish "manager is gone" from a per-key failure and decide whether to abort or fall back to a non-messaging path.

View Source 
var ErrNotSealTagged = errors.New("messaging: Seal on a type that carries no seal tags; a plain event goes to the outbox as a struct payload")

ErrNotSealTagged is returned by Publisher.Seal for a type that carries no seal tags.

View Source 
var ErrPublishDoorUnavailable = errors.New("messaging: client carries no byte publish door; publish through a framework-built client")

ErrPublishDoorUnavailable is returned when a publish is attempted through a client that carries no byte door: one built by app.Options.MessagingClientFactory, a hand-written AMQPClient, or a testing/mocks double. Match it with errors.Is. The remedy is to publish through a framework-built client (deps.Messaging) or, in a test, to swap the handle for messaging/testing's capture publisher.

View Source 
var ErrSealingNotLinked = sealruntime.ErrNotLinked

ErrSealingNotLinked is the startup error for a seal-tagged type declared in a build that never imported messaging/sealed.

View Source 
var ErrTenantStampConflict = tenantstamp.ErrConflict

ErrTenantStampConflict reports a publish whose tenant stamp was supplied by the caller and disagrees with the one the framework resolved. The framework is the stamp's only writer: a caller-supplied value is an unauthenticated claim to act for a tenant, so the publish fails rather than being silently overwritten.

The streams lane re-exports this same value, so errors.Is holds across lanes.

Functions

func ConfigureSealing added in v0.63.0 

func ConfigureSealing(rt *SealRuntime)

ConfigureSealing records the runtime facts sealing needs. The app calls it once before collecting declarations; a seal-tagged declaration collected before it fails Validate.

func IsSealTagged added in v0.63.0 

func IsSealTagged(t reflect.Type) bool

IsSealTagged reports whether t (pointers unwrapped) is a struct that carries a `seal` tag anywhere the typed publish door would refuse to publish as plaintext: on its own fields or the members an untagged embedded struct promotes (the depth jose/sealed.ScanType inspects), OR misplaced on a named nested field or a tagged embed (which DeclareTypedPublisher refuses outright). It is a probe, not a scan: the codec judges the declaration; the probe must say yes wherever the codec would speak or the door would fail closed. It is the one detector every lane guard shares — the typed publish door, the streams typed declarations (which refuse a sealed T in v1) and the outbox door (which refuses a sealed struct payload) — so a struct is "sealed" in exactly one way.

The outbox door asks on every Publish, so the answer is memoized per Go type: the set of payload types a process publishes is small and fixed, and a type's tags never change.

func IsSealedDedupKey added in v0.63.0 

func IsSealedDedupKey(key string) bool

IsSealedDedupKey reports whether key has the sealed `<SignFamily>:<jti>` shape. A header-sourced id can never satisfy it: `:` is outside the header grammar ValidateEventID enforces.

func IsSealedDelivery added in v0.63.0 

func IsSealedDelivery(ctx context.Context) bool

IsSealedDelivery reports whether ctx belongs to a delivery the sealed typed door opened — the framework's own marker, unreachable from a header or from consumer code, which is what lets the ledger door admit a sealed key from a sealed consumer while still refusing the same spelling from a header.

The marker travels with the handler's context: a handler that calls inbox.ProcessOnce from a goroutine or with a context NOT derived from the one it was handed (context.Background() instead of context.WithoutCancel(ctx)) loses it and gets ErrInvalidEventID — fail closed. Derive the context.

func RegisterSealCodec added in v0.63.0 

func RegisterSealCodec(c SealCodec)

RegisterSealCodec installs the sealing codec. A blank import of messaging/sealed does this from init; a second registration panics.

func ResolveTenantStamp added in v0.63.0 

func ResolveTenantStamp(ctx context.Context, headers map[string]any) (string, error)

ResolveTenantStamp is the publish-side stamp rule for a writer that persists a message for later delivery instead of publishing it now — the outbox, whose Publish must snapshot the tenant while the originating context is still live, the same way it snapshots the trace keys (ADR-087 §3).

It applies the exact rule stampingPublisher applies: a caller-supplied TenantStampHeader in headers is refused with ErrTenantStampConflict (equal value included), then the context tenant is the stamp. There is no pool key to fall back on, so the context is the only source; an empty stamp with a nil error means no tenant is in play and nothing must be written.

func ValidateDedupKey added in v0.63.0 

func ValidateDedupKey(ctx context.Context, key string) error

ValidateDedupKey checks a key at the ledger door, whichever door produced it: under a sealed delivery (IsSealedDelivery) a sealed `<SignFamily>:<jti>` key passes; everywhere, a header id passes under ValidateEventID's grammar. Anything else wraps ErrInvalidEventID and names the byte length only — so a publisher spelling a sealed key into x-outbox-event-id on an unsealed consumer is refused before the ledger, exactly as before sealing existed.

func ValidateEventID added in v0.63.0 

func ValidateEventID(id string) error

ValidateEventID checks id against the ledger grammar. Every framework path that turns a header into a ledger id runs it — Metadata.DedupKey here and inbox.ProcessOnce at the ledger door — so consumer code never has to.

func ValidatePublishDestination added in v0.61.0 

func ValidatePublishDestination(exchange, routingKey string, headers map[string]any) error

ValidatePublishDestination checks every caller-supplied shortstr a publish puts on the wire: the exchange and routing key, which travel in the basic.publish METHOD frame, and every header KEY, which travels in the CONTENT-HEADER frame that follows it (the same frame carrying CorrelationId, which ADR-070 already guards). One operation, two frames, one ceiling (nested tables included — a table's keys are shortstrs at every depth).

Length only. The charset is deliberately NOT checked: unlike the consume side (ADR-070, `[C60.17]`), where the value is a foreign publisher's, these are the service's OWN destinations, and a broker that dislikes one answers with a CHANNEL error — recoverable, and not the connection-wide failure this guard exists to prevent. Empty is legal: the default exchange and a fanout binding both use it.

It is exported for callers that record a destination now and publish it later — the outbox writes exchange, routing key and header keys to a ledger row, and a row the frame can never carry is better refused at the INSERT than parked by the relay after MaxRetries. They run the rule rather than restating the ceiling.

Types

type AMQPClient 

type AMQPClient interface {
	Client

	// 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. It is the module-facing type — what ModuleDeps.Messaging and scheduler.JobContext.Messaging return — and carries no byte publish door: the framework's own client and its stamping wrapper implement an unexported one that Publisher[T].Publish and the outbox relay reach (ADR-096).

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.

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 {
	// 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 consuming messages while hiding the complexity of connection management, retries, and protocol-specific details. It carries no publish method: a module publishes through the Publisher[T] handle that DeclareTypedPublisher returns, never by handing bytes to the client (ADR-096).

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 publishBytes). 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 publishBytes 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 publishBytes'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)
	// Args are per-consumer arguments forwarded to basic.consume
	// (x-stream-offset, x-priority, ...).
	Args map[string]any
}

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)
	Args          map[string]any // Per-consumer arguments forwarded to basic.consume (x-stream-offset, x-priority, ...)
	// TenantOptional lets this consumer run a delivery that carries no tenant stamp,
	// for a control-plane consumer whose events belong to no tenant. The default is
	// false — fail closed — and it never admits a stamp that is present but unusable.
	TenantOptional bool
}

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.

For a seal-tagged T this is the sealed typed door (ADR-097): every delivery is opened — verified, tid-judged, decrypted — BEFORE decode and validation, a refused one is a *PayloadError at PayloadStageOpen (nacked without requeue), and fn's Metadata carries the verified envelope, so meta.Sealed() is (envelope, true) and meta.DedupKey() is `<SignFamily>:<jti>`. A sealed consumer that cannot start — codec not linked, runtime not configured, no key material, a family missing a provisioned generation in its inherited role — fails Validate.

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)
	Args          map[string]any // Per-consumer arguments forwarded to basic.consume (x-stream-offset, x-priority, ...)
	// TenantOptional lets this consumer run a delivery that carries no tenant stamp
	// (a control-plane consumer). Default false: a consumer that needs a tenant fails
	// the delivery closed rather than running the handler without one.
	TenantOptional bool
}

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) DeclareStreamQueue added in v0.59.0 

func (d *Declarations) DeclareStreamQueue(name string, spec *StreamQueueSpec) *QueueDeclaration

DeclareStreamQueue declares a RabbitMQ stream queue (x-queue-type: stream): an append-only replicated log read non-destructively at a client-chosen offset, instead of a classic queue's destructive consume. Consumers pick a start position with the x-stream-offset consumer Arg (see wiki/messaging.md). A nil spec declares the queue with broker-default retention. Returns the registered 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 EventPublisher added in v0.63.0 

type EventPublisher[T any] interface {
	Publish(ctx context.Context, client AMQPClient, evt T) error
}

EventPublisher is the seam a module depends on when it wants to swap the handle in a test: *Publisher[T] satisfies it, and so does the capture double in messaging/testing. Production code declares the handle with DeclareTypedPublisher and stores it behind this interface.

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. The idle-publisher sweep starts here and stops in Close, so callers need not drive it (ADR-067); StartCleanup remains available and idempotent.

func (*Manager) Close added in v0.9.0 

func (m *Manager) Close() error

Close closes all clients and stops the idle-publisher sweep the constructor started. Publisher closes go through the pool (which stops its own cleanup loop and joins every per-publisher close failure); consumer closes are handled directly. A publisher client still borrowed by in-flight work is closed at its final release instead of by this call, and that deferred close failure (if any) is excluded from this return value — it is counted in Stats()["errors"] instead (wiki/migrations.md C581.3). Every failure returned here, 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 begins, Publisher fails closed rather than resurrecting a publisher (F22) — except a caller already mid-Publisher on a fresh client another borrower holds, who may still receive that live handle after Close returns; it closes exactly once, at its final release. 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. The constructor already started a sweep, so this is a no-op unless StopCleanup ran first (the pool's loop is single-instance).

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. Unlike Close it does not mark the manager closed and leaves the replay state intact, so a Stop is recoverable while a Close is terminal.

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
	// CleanupInterval is how often the idle-publisher sweep runs; <=0 uses the documented
	// 2-minute default. The manager starts that sweep itself at construction (ADR-067).
	CleanupInterval time.Duration
	// 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
	// TenantStamps makes consumers read the tenant stamp off each delivery and seed
	// the handler context with it. True only under multitenant.enabled together with
	// messaging.tenancy: shared — under per-tenant tenancy the replay key is already
	// the tenant, and in single-tenant mode there is none to read.
	TenantStamps bool
}

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: take meta.DedupKey() (the grammar-validated x-outbox-event-id, or an error to return) 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) DedupKey added in v0.63.0 

func (m Metadata) DedupKey() (string, error)

DedupKey returns the id the inbox ledger should be keyed on for this delivery. For a sealed consumer it is `<SignFamily>:<jti>` — the Logical sign family, never the concrete Generation, so a rotation does not re-open the replay window — composed from the verified envelope and never an error. For a plain typed consumer it is the grammar-validated x-outbox-event-id header; the error wraps ErrInvalidEventID when the header is absent, empty, over 128 bytes, or carries a byte outside [A-Za-z0-9_-]. Return it from the handler: the delivery is nacked without requeue, like any other poison message. AMQP header values arrive as string or []byte depending on the broker and client, so both are accepted.

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. Nil when no headers were published. For the outbox dedup key prefer DedupKey, which validates as it extracts. 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.

func (Metadata) Sealed added in v0.63.0 

func (m Metadata) Sealed() (SealedEnvelope, bool)

Sealed reports whether this delivery arrived through the sealed typed door and, when it did, what its verified envelope asserts. The answer is a property of the consumer TYPE, never of the message: a sealed consumer gets (envelope, true) for every delivery it runs — the opener refused every other one before the handler — and a plain typed consumer gets (zero, false) for every delivery, whatever headers the publisher wrote, so a handler branching on ok cannot be steered by a caller-written header.

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 "open", "decode" or "validate"; for control flow, match
	// errors.Is against ErrPayloadOpenRefused, 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. It is this lane's thin surface over payloaderr.Body, which owns the rendering rules and the SECURITY rationale behind them: Error() and Fields() are safe to log, Unwrap() is not.

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"], with every bracketed span redacted. It is empty for decode failures and for a nil receiver.

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. 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 = PayloadStage(payloaderr.StageDecode)
	PayloadStageValidate PayloadStage = PayloadStage(payloaderr.StageValidate)
	// PayloadStageOpen is a sealed message the opener refused before decode
	// (ADR-097): the rule's code is in Error() and the opener's own error is in
	// the chain for errors.As.
	PayloadStageOpen PayloadStage = PayloadStage(payloaderr.StageOpen)
)

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

type Publisher added in v0.63.0 

type Publisher[T any] struct {
	// contains filtered or unexported fields
}

Publisher is the module-facing handle DeclareTypedPublisher returns: a publisher bound at declaration time to ONE destination — the declared exchange, routing key, default headers and delivery flags — so a call site never re-spells them. It is the publish mirror of the typed consumer built by DeclareTypedConsumer.

Every field is set once at construction and read-only afterwards: one handle is shared by every goroutine of a module AND, under multi-tenant messaging, by every tenant's client, so any mutable state here would be a data race. Publish therefore hands the client a fresh headers map on every call and never writes back into the handle. The copy is one level deep, the same depth RegisterPublisher, Declarations.Clone and the stamping wrapper copy at: declared default headers are startup config, expected to be scalars (a string, a number, a bool), not nested tables a publish would mutate.

func DeclareTypedPublisher added in v0.63.0 

func DeclareTypedPublisher[T any](decls *Declarations, opts *PublisherOptions) *Publisher[T]

DeclareTypedPublisher registers a publisher exactly as DeclarePublisher does — same registry entry, same replay, validation and hash path — and returns a Publisher[T] handle bound to the declared destination. Keep the handle on the module (there is no deps accessor to look one up again) and publish through it from handlers and services.

The exchange is not declared here: pass it to DeclareTopicExchange separately, exactly as a DeclarePublisher with a nil exchange does.

It panics on a nil decls or a nil opts — both are declaration-time wiring mistakes, and the package already fails startup that way for the typed consumer entries.

func (*Publisher[T]) Publish added in v0.63.0 

func (h *Publisher[T]) Publish(ctx context.Context, client AMQPClient, evt T) error

Publish encodes evt and publishes it to the DECLARED exchange and routing key with the declared default headers, through client — the tenant-aware client a handler already holds (the getMessaging(ctx) idiom), so the framework's tenant stamping and trace injection apply unchanged.

A plain T is JSON-marshaled. A seal-tagged T is sealed (ADR-097) — once, here, before the client's retry loop, so every attempt and every redelivery carries the same bytes and the same signed jti. A caller-side retry after this call fails is a new seal and a new jti.

An encode or seal failure is returned wrapped and publishes nothing. Every other error is the client's own (ErrInvalidPublishDestination, ErrPublishRetriesExhausted, ...) and is returned unwrapped.

Safe for concurrent use: the handle is never written after construction and the client receives a fresh copy of the declared headers on every call.

func (*Publisher[T]) Seal added in v0.63.0 

func (h *Publisher[T]) Seal(ctx context.Context, evt T) ([]byte, error)

Seal returns the sealed wire bytes for evt without publishing them — the outbox lane persists them as-is (persisted-sealed, ADR-097) and the relay moves them byte-identical. A plain T has nothing to seal: it returns ErrNotSealTagged, and the event goes to the outbox as a struct payload the outbox already marshals.

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.

type SealCodec added in v0.63.0 

type SealCodec = sealruntime.Codec

SealCodec is what messaging/sealed registers.

type SealEnvelope added in v0.63.0 

type SealEnvelope = sealruntime.Envelope

SealEnvelope is the seam's envelope; SealedEnvelope is its Metadata twin.

type SealKeyStore added in v0.63.0 

type SealKeyStore = sealruntime.KeyStore

SealKeyStore is the app.KeyStore subset sealing needs.

type SealOpenRefusedError added in v0.63.0 

type SealOpenRefusedError = sealruntime.OpenRefusedError

SealOpenRefusedError is the seam's refusal, found in a PayloadError's chain.

type SealOpener added in v0.63.0 

type SealOpener = sealruntime.Opener

SealOpener opens one sealed delivery.

type SealOpenerProvider added in v0.63.0 

type SealOpenerProvider = sealruntime.OpenerProvider

SealOpenerProvider is the OPTIONAL consume side of a SealCodec.

type SealRuntime added in v0.63.0 

type SealRuntime = sealruntime.Runtime

SealRuntime is what the app configures at bootstrap.

func SealingRuntime added in v0.63.0 

func SealingRuntime() *SealRuntime

SealingRuntime returns the facts ConfigureSealing recorded, or nil before it ran.

type SealSpec added in v0.63.0 

type SealSpec = sealruntime.Spec

SealSpec is a codec's scanned declaration as messaging sees it: the two Logical kids.

type SealTenancy added in v0.63.0 

type SealTenancy = sealruntime.Tenancy

SealTenancy is the tenancy fact the opener's tid rule reads.

type SealTenantRule added in v0.63.0 

type SealTenantRule = sealruntime.TenantRule

SealTenantRule is the tid expectation the sealed door derives per delivery.

type SealedEnvelope added in v0.63.0 

type SealedEnvelope struct {
	// JTI is the token id the sealed dedup key `<SignFamily>:<jti>` is built from.
	JTI string
	// IssuedAt is the protected header's iat claim.
	IssuedAt time.Time
	// EventType is the event type asserted INSIDE the envelope, which the
	// framework has matched against the delivery's wire-level type.
	EventType string
	// TenantID is the tenant asserted inside the envelope (empty single-tenant).
	TenantID string
	// SignKid and SignFamily identify the verifying key and its key family.
	SignKid    string
	SignFamily string
	// EncKid identifies the key the envelope was decrypted with.
	EncKid string
}

SealedEnvelope is what a sealed (JWE-of-JWS) message's protected header asserts about itself once the framework has verified it. Plain data: a consumer that never seals reads this type without linking go-jose (the jose side has its own envelope type; the sealed door maps between them). It is reachable only through Metadata.Sealed, filled by the sealed typed door (DeclareTypedConsumerWithMeta on a seal-tagged T) and zero everywhere else.

type Sealer added in v0.63.0 

type Sealer = sealruntime.Sealer

Sealer turns one event into its sealed wire bytes.

type StreamQueueSpec added in v0.59.0 

type StreamQueueSpec struct {
	// MaxAge -> x-max-age ("<n>s"), truncated to whole seconds. A non-zero
	// sub-second value floors to "1s": second granularity is RabbitMQ's, so
	// anything briefer is inexpressible and would otherwise render "0s",
	// discarding the retention the caller asked for.
	MaxAge              time.Duration
	MaxLengthBytes      int64 // -> x-max-length-bytes
	MaxSegmentSizeBytes int64 // -> x-stream-max-segment-size-bytes
}

StreamQueueSpec configures retention for a stream queue. Zero-value fields are omitted (broker defaults apply). MaxAge is rendered as whole seconds.

Source Files

View all Source files

Directories

Path Synopsis
internal
delivery
Package delivery runs the delivery pipeline both messaging lanes share: everything that happens to one consumed message between "bytes arrived" and "outcome recorded" — trace extraction from the lane's carrier, the consumer span, the per-message lease scope, handler invocation, panic-to-error, one consumed record at completion, and the lane's own outcome line.
 Package delivery runs the delivery pipeline both messaging lanes share: everything that happens to one consumed message between "bytes arrived" and "outcome recorded" — trace extraction from the lane's carrier, the consumer span, the per-message lease scope, handler invocation, panic-to-error, one consumed record at completion, and the lane's own outcome line.
lanecontract
Package lanecontract holds the contract every messaging lane driving the delivery pipeline must satisfy, and the fixture that drives it.
 Package lanecontract holds the contract every messaging lane driving the delivery pipeline must satisfy, and the fixture that drives it.
payloaderr
Package payloaderr holds the payload-error core both messaging lanes build their typed consumers on: the decode and struct-validation steps, and a failure rendering that never echoes the body that caused it.
 Package payloaderr holds the payload-error core both messaging lanes build their typed consumers on: the decode and struct-validation steps, and a failure rendering that never echoes the body that caused it.
sealruntime
Package sealruntime is the link-time seam between the messaging package and the payload-sealing codec (ADR-097).
 Package sealruntime is the link-time seam between the messaging package and the payload-sealing codec (ADR-097).
tenantstamp
Package tenantstamp owns the tenant stamp: the carrier entry that carries a tenant identity between a producer and a consumer.
 Package tenantstamp owns the tenant stamp: the carrier entry that carries a tenant identity between a producer and a consumer.
tracking
Package sealed wires the jose/sealed codec into the messaging typed doors.
 Package sealed wires the jose/sealed codec into the messaging typed doors.
Package streams publishes to and consumes RabbitMQ streams over the native stream protocol (default port 5552, rabbitmq_stream plugin).
 Package streams publishes to and consumes RabbitMQ streams over the native stream protocol (default port 5552, rabbitmq_stream plugin).
Package testing provides test doubles for the messaging package's module-facing surface.
 Package testing provides test doubles for the messaging package's module-facing surface.

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.