Docs / RabbitMQ Streams

RabbitMQ Streams

Native Stream publishing and consumption, super streams, filtering, and durable outbox/inbox deduplication

go.digitalxero.dev/mq-rmqstream/v2 uses RabbitMQ’s native Stream protocol. It is a separate backend from the AMQP module and requires Go 1.25 or newer. Importing it registers rabbitmq-stream:// and rabbitmq-stream+tls:// for both producers and consumers.

Publishing succeeds only after broker confirmation. Consumers acknowledge by storing progress, not deleting records. Only the settled prefix of delivered records can advance the checkpoint; unfinished earlier work prevents later handlers from skipping it. A crash can replay records, and stream retention can remove records independently of consumption. Design handlers for at-least-once processing.

Module availability and evaluating this change

The Stream module and optional Redis adapter are introduced by the broker extensions change. Their new vanity import paths require the supporting go-pkgs mapping branch to be deployed, followed by the mqutils module release. These two endpoints were not yet live when this change was prepared. Existing backend import paths are unaffected.

Evaluate the implementation from its canonical repository with local module replacements while the changes are under review:

git clone --branch codex/broker-extensions https://gitlab.com/digitalxero/mqutils.git
cd mqutils/_examples/rmqstream_dedup
go test -race ./...

The runnable examples already replace dependencies with this checkout. For an existing application, point all three modules at the corresponding directories in your checkout (substitute its absolute path):

go mod edit -require=go.digitalxero.dev/mqutils/v2@v2.2.0
go mod edit -require=go.digitalxero.dev/mq-rmqstream/v2@v2.0.0
go mod edit -require=go.digitalxero.dev/mq-rmqstream-dedupredis/v2@v2.0.0
go mod edit -replace=go.digitalxero.dev/mqutils/v2=/absolute/path/mqutils
go mod edit -replace=go.digitalxero.dev/mq-rmqstream/v2=/absolute/path/mqutils/rmqstream
go mod edit -replace=go.digitalxero.dev/mq-rmqstream-dedupredis/v2=/absolute/path/mqutils/rmqstream/dedupredis
go mod tidy

The third requirement/replacement is needed only when using the Redis adapter. Core Stream consumers can use an application-supplied store and have no mandatory Redis dependency.

A single stream

The examples below use rmqstream for go.digitalxero.dev/mq-rmqstream/v2 and types for go.digitalxero.dev/mqutils/v2/types.

producer, err := rmqstream.NewProducerBuilder().
    WithURL("rabbitmq-stream://guest:guest@localhost:5552").
    WithDestination("orders").
    WithAutoDeclare(true).
    Build(ctx)
if err != nil {
    return err
}
defer producer.Close()
if err := producer.Start(ctx); err != nil {
    return err
}
message := rmqstream.NewMessageBuilder().
    WithMessageId("order-42").
    WithRoutingKey("customer-7").
    WithContentType("application/json").
    WithBody([]byte(`{"id":42}`)).
    Build()
return producer.PublishMsg(ctx, "orders", "customer-7", message)

Producer Build validates; Start connects. Stream consumer Build validates; Run connects and consumes until shutdown. A consumer instance runs once; create a new instance for a subsequent run. Closing is idempotent.

consumer, err := rmqstream.NewConsumerBuilder().
    WithURL("rabbitmq-stream://guest:guest@localhost:5552").
    WithDestination("orders").
    WithConsumerGroup("billing-v1").
    WithAutoReconnect(true).
    WithInitialOffset("first").
    WithHandler(func(ctx context.Context, msg types.Message) error {
        return processOrder(ctx, msg.Body())
    }).
    Build(ctx)
if err != nil {
    return err
}
defer consumer.Close()
return consumer.Run(ctx)

The consumer identity is durable. When a checkpoint exists it resumes from the stored safe offset plus one; initial_offset selects the starting position for a new identity. Offset numbers can contain gaps. A successful Ack means progress was submitted to the broker; interruption before durable checkpoint persistence can still cause redelivery.

rmqstream.Message extends the shared message interface with Offset(), Partition(), and an ownership Context(). Acknowledgments from an expired connection or partition ownership session cannot advance its replacement.

Super streams and ownership

Declare a three-partition super stream with hash routing:

producer, err := rmqstream.NewProducerBuilder().
    WithURL(streamURL).
    WithDestination("orders-partitioned").
    WithStreamMode("super_stream").
    WithRoutingStrategy("hash").
    WithPartitions(3).
    WithAutoDeclare(true).
    Build(ctx)

For binding-key routing, use WithRoutingStrategy("key") and WithBindingKeys([]string{"emea", "americas"}) instead of WithPartitions. Message.RoutingKey() is the default route. A typed WithRoutingExtractor(func(types.Message) string) can supply it instead. Missing routes are errors. Ordering is per physical partition, not across the entire super stream; application concurrency can also reorder handler completion. Existing partition order and bindings must match. Auto-declaration also reserves three empty ordinary streams named mqutils-topology-<name-hash>-0, -1, and -2. Their immutable declaration settings encode the ordered binding keys and retention settings, so competing incompatible creators fail before changing application topology. These resources require configure permission and persist after Close. Keep them for the topology lifetime; remove them only as part of an intentional topology deletion/reset, after all users stop and the original topology is removed. External administrative changes bypass this reservation, so applications still need exclusive topology ownership. These markers are separate from the per-consumer filter markers below.

Consumers use WithStreamMode("super_stream") and the same logical destination. Members sharing a consumer_group use single active consumer ownership per partition; super-stream consumption enables that ownership automatically. WithSingleActiveConsumer(true) also enables it for an individual stream. Each physical stream keeps an independent safe checkpoint. Ownership loss cancels its in-flight work and invalidates its old acknowledgments.

Named ordinary producers are used per physical partition, so super streams can combine routing, filtering, and broker publishing deduplication. A failure in one partition does not authorize checkpointing unfinished work in another.

Filtering and durable filter identities

WithFilterHeader("region") on a producer extracts a string header as the broker filter value. WithFilterExtractor(func(types.Message) string) supports a typed extraction function. The adapter stores that value as mqutils.filter and preserves it during retry/dead-letter forwarding.

producerBuilder := rmqstream.NewProducerBuilder().
    WithURL(streamURL).WithDestination("orders").
    WithFilterHeader("region")

consumerBuilder := rmqstream.NewConsumerBuilder().
    WithURL(streamURL).WithDestination("orders").
    WithConsumerGroup("eu-billing-v1").
    WithFilterValues([]string{"eu"}).
    WithMatchUnfiltered(false).
    WithFilterPredicate("paid-orders-v1", func(msg types.Message) bool {
        return msg.Headers()["payment_status"] == "paid"
    }).
    WithHandler(processOrderMessage)

The broker’s Bloom filter can produce false positives. Exact value matching and the optional predicate run inside the adapter before application handlers. All delivered records, including excluded records, participate in settlement. A tail of excluded deliveries can advance progress once earlier included work finishes; filtering never allows a numeric offset gap to hide unfinished work.

The exact predicate requires a stable contract name. Change it whenever predicate semantics change. Filter values, unfiltered-message policy, header selection, and predicate contract are bound to the durable consumer identity. Reusing the identity with incompatible settings fails; choose a new consumer_group to change which historical messages are processed.

The adapter records this contract using three reserved empty streams named mqutils-filter-<identity-hash>-0, -1, and -2 for each consumer-group/physical-stream pair. Their immutable declaration settings encode the filter fingerprint; broker declaration equivalence prevents conflicting contracts from racing. The markers never contain messages and do not change source-stream retention. Consumers need topology declaration/discovery permissions even when application auto_declare is false and filtering is not enabled. Keep these metadata resources while retaining the consumer identity; Close does not remove them. Delete all three markers and the associated checkpoints only when retiring the identity or explicitly resetting it with an intentional replay plan.

Filtering requires a broker supporting Stream filtering (RabbitMQ 3.13 or newer). Unsupported feature configuration is reported explicitly. The adapter disables subentry aggregation and rejects sub_entry_size; application handler batching remains available.

Broker publishing deduplication

Broker deduplication suppresses repeat stream writes using a producer name and publishing-sequence high-water mark on each physical stream. It does not use MessageId as an arbitrary idempotency key.

Configure WithProducerName("orders-outbox-v1"). Persist an event’s producer name, physical partition, positive sequence, immutable payload, and properties before publishing it:

partition, err := producer.ResolvePartition(ctx, message)
if err != nil {
    return err
}
// Persist the event, partition, producer identity, and next sequence in your
// durable outbox before calling PublishSequence. Never derive sequence from a hash.
if err := saveOutbox(ctx, producerName, partition, sequence, message); err != nil {
    return err
}
return producer.PublishSequence(ctx, partition, sequence, message)

On restart or a lost confirmation, use ReplaySequence with the same physical partition, sequence, producer identity, payload, and properties. Do not ask the router to choose a new partition for a replay. LastPublishingID returns the broker watermark (0 for a new identity); compare it with durable application history before assigning new numbers. A watermark alone cannot recover which business event used each sequence.

A producer reference/physical-stream pair requires one active application owner. The module detects conflicting owners within one process; separate processes must coordinate externally. RabbitMQ does not fence competing named publishers. Reusing a sequence for different contents is invalid; after a process restart the application’s durable ledger is required to detect that conflict. Never hash a MessageId into a sequence or blindly allocate a new sequence after uncertainty.

ErrSequenceOrder identifies invalid sequence progression; ErrSequenceConflict identifies conflicting reuse known to the current producer. PublishUncertainError errors expose the unresolved physical partition and sequence. Keep their durable outbox entry unresolved until the original identity is reconciled/replayed. Generic Publish/PublishMsg does not provide event deduplication across process restarts without this durable association.

The runnable outbox/inbox example uses SQLite transactions to persist publication identity and immutable content before sending. It retains confirmed history, refuses broker/ledger disagreement, and replays pending rows before allocating new sequences. A durable ownership row requires explicit token-specific recovery after confirming an old process has stopped. All publishers using that identity must share the same database.

Application message deduplication

Application deduplication is separately opt-in through WithDeduplicator. A DeduplicationStore atomically claims, renews, completes, and releases logical message keys using ownership tokens and fingerprints.

// dedupredis imports go.digitalxero.dev/mq-rmqstream-dedupredis/v2.
store, err := dedupredis.New().WithURL(redisURL).Build(ctx)
if err != nil {
    return err
}
defer store.Close()
dedup, err := rmqstream.NewDeduplicator().
    WithStore(store).
    WithNamespace("billing").
    WithConsumerGroup("billing-v1").
    WithRetention(7 * 24 * time.Hour).
    Build()
if err != nil {
    return err
}
consumer, err := rmqstream.NewConsumerBuilder().
    WithURL(streamURL).WithDestination("orders").
    WithConsumerGroup("billing-v1").
    WithDeduplicator(dedup).
    WithHandler(processOrderMessage).
    Build(ctx)

Namespace and consumer group are explicit, and retention must be specified. The default logical key is MessageId; missing keys fail processing rather than silently substituting a random ID. WithKeyExtractor supplies another typed key. The store key includes namespace, group, and logical key, excluding physical partition. Use a destination-specific namespace for destination-local suppression.

The default fingerprint includes body, application properties and headers, content type/encoding, and routing/reply metadata. Stream retry counters and original offsets are excluded. A logical key reused with different contents returns ErrDeduplicationConflict. WithFingerprintExtractor customizes the application contract.

A completed duplicate skips the handler and can settle. An in-progress duplicate waits. Newly owned claims renew during long handlers (default lease 30s), and lost ownership cancels the handler context. Stale tokens cannot complete or release newer claims. Handlers must honor cancellation; the library cannot fence arbitrary external effects of code that ignores it.

Manual Ack is gated: it cannot advance a source checkpoint before successful handler return and durable completion. Manual Nack makes the handler attempt fail. Failed handlers release owned claims for retry. On ambiguous completion, the middleware retries that same conditional completion without immediately rerunning the handler. Store failures preserve unresolved deliveries.

Batch handlers receive only newly claimed, unique logical keys in original application order. Repeated keys within the batch coalesce. Completed duplicates settle separately. A failed batch releases its owned claims; partial completion failures do not undo completed entries or immediately rerun their handlers.

The optional Redis module requires maxmemory-policy=noeviction and CONFIG GET permission to verify it. Do not externally delete/expire its keys. Server restart durability depends on Redis persistence/fsync and replication settings; a Redis instance with persistence disabled loses this state on restart. A supplied WithClient remains owned by its caller; closing an adapter-created client is the adapter’s responsibility. Known go-redis clients must enable ContextTimeoutEnabled and finite dial/read/write/pool timeouts. Cluster validation checks all known primary and replica nodes for noeviction; maintain that policy when adding nodes. Redis Ring is rejected because failed-shard rehashing can route a logical key to a different independent store. Custom UniversalClient wrappers must enforce context deadlines, bounded I/O, and the policy on every backend they may use; their hidden options/topology cannot be inspected.

Finite retention permits duplicate processing after expiry. There remains a crash window between an external effect and saving completion. The SQLite inbox example closes that window for its balance update by committing the inbox receipt and update together in one database transaction. Redis suppression alone does not make arbitrary database/API effects exactly once.

Configuration reference

Backend-specific settings are supplied with WithConfig(*viper.Viper); typed methods cover the common cases and function-valued hooks.

Without a dead-letter stream or explicit discard, retry exhaustion stops the consumer without advancing the failed record. Failed forwarding preserves replayability. Retried/dead-lettered messages preserve filter and retry metadata.

Use rabbitmq-stream+tls:// for TLS; skip_verify only relaxes certificate verification. Both certificate and key are required for client authentication. AMQP exchange types, message priority, expiration, and subentry aggregation settings are rejected rather than treated as Stream features.

The Stream API reference and Redis adapter reference describe the complete interfaces.