Message Delivery

A registered listener and a broker subscription have different lifetimes. The consumer owns local listener routing; its driver owns receiving, retention, and settlement. Understanding that boundary lets you stop a service without mistaking cancellation for completed work, or a callback error for a retry.

This page assumes a configured driver and the OrderCreated topic. Publication acceptance remains separate from recipient handling: the publisher never receives listener errors.

Owning Registrations

Use the unsubscribe function when a component no longer needs notifications. With an existing consumer and application search indexer, retain the function only after successful registration:

unsubscribe, err := consumer.Listen(OrderCreatedTopic,
	func(ctx context.Context, order OrderCreated) error {
		return search.Index(ctx, order.OrderID)
	})
if err != nil {
	return err
}

The registration now belongs to your component. On its shutdown, remove it:

unsubscribe()
unsubscribe()

Both calls are safe; only the first removes the registration. Removal affects future delivery snapshots. A delivery that already captured the listener can still invoke it, so unsubscribe does not join or cancel an in-flight callback. Do not release callback dependencies until their active users have finished.

Registration and removal are concurrent-safe, and callbacks run outside the registration lock. Each registration is independent, even if two use the same topic or function. Removing one does not remove the others or close the driver’s subscription. Conversely, closing a transport does not remove the consumer’s local registrations.

Stopping Consumers

Prefer an application runner for a service-wide consumer. If a smaller component directly owns consumption, give it a cancelable context and retain the result so its owner can join the runner:

import "context"

runCtx, cancel := context.WithCancel(ctx)
result := make(chan error, 1)
go func() {
	result <- consumer.Run(runCtx)
}()

This starts consumption, not a portable readiness handshake. Do not use goroutine startup as proof that a non-retaining subscription can receive messages. When the component ends, cancel and wait:

import "errors"

cancel()
err := <-result
if !errors.Is(err, context.Canceled) {
	return err
}

The driver returns the context error when cancellation ends consumption; Run preserves driver errors. An earlier transport failure can therefore be the result instead. Listener failures are delivery outcomes and do not by themselves end this loop.

Joining Run does not establish that every callback has finished. In particular, memory deliveries already admitted retain context values without the consumer’s cancellation or deadline. NATS cancellation unsubscribes without waiting for a running callback. Bound work inside the callback when its dependencies need predictable cleanup:

import "time"

unsubscribe, err := consumer.Listen(OrderCreatedTopic,
	func(ctx context.Context, order OrderCreated) error {
		work, cancel := context.WithTimeout(ctx, 5*time.Second)
		defer cancel()
		return search.Index(work, order.OrderID)
	})

The timeout only bounds operations that honor their context. If resource release requires proof that callbacks ended, track and join those callbacks in the owning component too. Driver Stop has provider-specific draining limits; follow driver ownership rather than assuming all drivers have the same shutdown contract. NATS and MQTT connection shutdown does not end their Consume context waits: cancel consumers separately.

Listener Ordering

For one delivery, the facade takes a snapshot and invokes every matching listener sequentially in registration order. An exact listener and a wildcard listener can both match. A failure does not skip later listeners. A delivery with no matching listeners succeeds; on SQS that permits deletion of the message.

Typed listeners decode the payload independently. A malformed payload can fail one listener’s decoding without preventing a different listener from running. Topics sharing a name must still agree on their intended JSON contract; separate decoding is not schema negotiation.

The facade joins all listener failures and returns the delivery result to the driver. Listener panics are recovered as errors, including a stack trace. One delivery’s ordering does not order different messages: transports can invoke several deliveries concurrently. Synchronize shared application state and do not assume publication order is a portable callback-order guarantee.

Pattern listeners share the envelope’s payload bytes and metadata map. Treat them as read-only; copy data that needs mutation or independent ownership.

Permanent Failures

If retrying the same input cannot help, mark that failure permanent. For example, reject an order notification without its required identifier before calling the indexer:

import "errors"

unsubscribe, err := consumer.Listen(OrderCreatedTopic,
	func(ctx context.Context, order OrderCreated) error {
		if order.OrderID == "" {
			return broadcast.Permanent(errors.New("missing order identifier"))
		}
		return search.Index(ctx, order.OrderID)
	})

The returned error matches broadcast.ErrPermanent under errors.Is and retains its original cause. Permanent(nil) returns nil. Invalid delivery names and typed payload-decoding failures are permanent automatically.

A joined delivery failure is permanent only if every failed listener is permanent. Successful listeners do not change that decision. If one failure is transient and another permanent, the delivery remains retryable. A redelivery runs every matching listener again, including listeners that previously succeeded. Use stable application identifiers and idempotent effects; a transport-generated ID is not a universal application deduplication key.

Permanent does not create retention, a retry policy, or a dead-letter destination. Non-retaining drivers do not retry either error class. The SQS broadcast driver deletes permanent failures rather than sending them to a dead-letter queue.

Observing Failures

To observe callback failures independently of transport policy, give the consumer your existing *event.Bus named bus:

consumer := broadcast.NewConsumer(driver, func(receiver *broadcast.ConsumerBuilder) {
	receiver.Events(bus)
})

Attach an observer on that bus. With an application logger, record the routing name without exporting the original error:

unsubscribe, err := bus.Listen(broadcast.ListenerFailed,
	func(ctx context.Context, failure broadcast.ListenerFailedEvent) error {
		logger.WarnContext(ctx, "broadcast listener failed", "name", failure.Name)
		return nil
	})
if err != nil {
	return err
}
defer unsubscribe()

The topic is broadcast.listener.failed. Each failed listener emits one ListenerFailedEvent in listener order, with ID, Name, and Err. ID is empty when the transport supplies none. Decode failures and recovered panics also emit events; successful deliveries emit none. Without Events, no bus notification is published. Publishers and drivers do not emit this event.

The event omits message payload and metadata, but Err retains application causes and recovered panic details. Treat it as sensitive diagnostic data, not safe external output. Observing the event cannot change delivery settlement, and it is not evidence of broker retention or acknowledgement. See built-in service events and monitoring privacy.

Transport Outcomes

Memory, Redis Pub/Sub, and Core NATS do not retry returned listener errors or retain offline notifications. Memory and NATS log handler failures; Redis does not turn those failures into retries. Subscribers must recover missed state another way, including after a process restart.

RabbitMQ uses auto-acknowledged, ephemeral consumer queues. Listener failure does not requeue a notification. Publication retries after a lost broker confirmation can nevertheless duplicate it. MQTT’s owned clean sessions likewise have no offline retention; callback errors do not request retries or change native acknowledgement. Protocol QoS does not guarantee exactly-once effects.

SQS deletes successfully handled messages, malformed envelopes, and deliveries whose failures are all permanent. Other failures remain for redelivery after visibility expires, subject to queue retention and redrive policies. Deletion failure can repeat an already successful effect. The driver does not renew visibility; size the timeout for the entire sequential batch. SNS acceptance does not prove delivery to every subscribed queue.

For complete transport configuration and cancellation limits, use the driver comparison. For every facade function, see the broadcast API.