mqutils
GitLab ↗

Docs / Backends / RabbitMQ Streams

RabbitMQ Streams

Confirmed publishing, durable offsets, super streams, filtering, and durable message deduplication

See the RabbitMQ Streams guide for configuration, ownership, filtering, durable outbox/inbox examples, and the current module availability note. This module requires Go 1.25 or newer. The optional Redis store has its own API reference.

go
import "go.digitalxero.dev/mq-rmqstream/v2"

Package rmqstream connects mqutils to RabbitMQ’s native Stream protocol.

Importing the package registers rabbitmq-stream:// and rabbitmq-stream+tls://. Producers await broker confirmations. Consumers checkpoint the settled delivery prefix, so concurrent handlers cannot commit past an earlier unfinished record. Acknowledgment stores progress; it does not delete stream data. Retention can remove data independently of consumption, and crash recovery can replay records.

Configure stream_mode=super_stream for partition discovery, hash or binding-key routing, and consumer groups with single active ownership per partition. Routing keys must be nonempty. There is no ordering guarantee between partitions. Producer filter_header and consumer filter_values enable Bloom filtering plus exact per-message filtering; match_unfiltered includes records without a value. Reusing a consumer_group with changed filters is rejected. Use a new identity.

ProducerName and PublishSequence/ReplaySequence support a durable application outbox. Persist identity, physical partition, sequence and immutable contents before sending. One process must exclusively own each producer reference/stream pair; RabbitMQ does not fence conflicting owners. The library rejects concurrent owners in the same process, but applications must coordinate across processes. Broker deduplication is sequence-based, not arbitrary MessageId deduplication.

WithDeduplicator separately enables application MessageId deduplication through an injected store. Exactly-once external side effects require a transactional inbox or another application-level idempotency mechanism.

Index

Variables

go
var (
    ErrDeduplicationConflict   = errors.New("deduplication key was reused with a different fingerprint")
    ErrDeduplicationClaimLost  = errors.New("deduplication claim ownership was lost")
    ErrDeduplicationKeyMissing = errors.New("deduplication requires a nonempty message key")
    ErrDeduplicationNack       = errors.New("deduplicated handler negatively acknowledged a message")
)

go
var (
    ErrClosed           = errors.New("rmqstream: closed")
    ErrSequenceConflict = errors.New("rmqstream: publishing sequence conflicts with previous event")
    ErrSequenceOrder    = errors.New("rmqstream: new publishing sequence must increase; use ReplaySequence for durable replay")
    ErrProducerOwned    = errors.New("rmqstream: producer identity is already owned in this process")
    ErrRetryExhausted   = errors.New("rmqstream: retry budget exhausted without dead-letter stream or explicit discard")
)

funcNewMessageBuilder

go
func NewMessageBuilder() types.MessageBuilder

NewMessageBuilder constructs an immutable message. Build returns this builder; do not mutate the builder afterward.

funcNewRMQStreamConsumer

go
func NewRMQStreamConsumer(v *viper.Viper) (types.Consumer, error)

NewRMQStreamConsumer is the registered Viper factory.

funcNewRMQStreamProducer

go
func NewRMQStreamProducer(v *viper.Viper) (types.Producer, error)

NewRMQStreamProducer is the registered Viper factory.

typeConsumer

Consumer consumes a stream or all assigned super-stream partitions.

go
type Consumer interface {
    types.Consumer
    Close() error
}

typeConsumerBuilder

ConsumerBuilder configures the consumer and implements the resulting runtime.

go
type ConsumerBuilder interface {
    WithConfig(*viper.Viper) ConsumerBuilder
    WithHandler(types.HandlerFunc) ConsumerBuilder
    WithBatchHandler(types.BatchHandlerFunc, int, time.Duration) ConsumerBuilder
    WithDeduplicator(Deduplicator) ConsumerBuilder
    WithGracefulShutdown(time.Duration) ConsumerBuilder
    WithFilterPredicate(string, func(types.Message) bool) ConsumerBuilder
    Build(context.Context) (Consumer, error)
    WithURL(string) ConsumerBuilder
    WithURLs([]string) ConsumerBuilder
    WithDestination(string) ConsumerBuilder
    WithConsumerGroup(string) ConsumerBuilder
    WithStreamMode(string) ConsumerBuilder
    WithPartitions(int) ConsumerBuilder
    WithBindingKeys([]string) ConsumerBuilder
    WithAutoDeclare(bool) ConsumerBuilder
    WithAutoReconnect(bool) ConsumerBuilder
    WithSingleActiveConsumer(bool) ConsumerBuilder
    WithInitialOffset(string) ConsumerBuilder
    WithFilterValues([]string) ConsumerBuilder
    WithMatchUnfiltered(bool) ConsumerBuilder
    WithConcurrency(int) ConsumerBuilder
    WithBuffer(int) ConsumerBuilder
    WithMaxRetries(int) ConsumerBuilder
    WithRetryDelay(time.Duration) ConsumerBuilder
    WithDeadLetterStream(string) ConsumerBuilder
    WithDiscardExhausted(bool) ConsumerBuilder
}

funcNewConsumerBuilder

go
func NewConsumerBuilder() ConsumerBuilder

typeDeduplicationStore

DeduplicationStore persists an atomic, fenced claim protocol. Stores must not evict active claims before lease expiry, or completed entries before retention. Claim returns owned for a new claim or the same live token, completed for a completed matching fingerprint, and neither when another live token owns it. A different fingerprint must return ErrDeduplicationConflict, including while a claim is active. Expired claims may be replaced atomically. Renew, Complete, and Release must reject stale tokens with ErrDeduplicationClaimLost. Complete must be idempotent for the same token and fingerprint, including a retry after an uncertain response; retries must not shorten retention. Renew on a matching completed token succeeds without modifying its expiry. Release never deletes a completed entry. Implementations must honor context deadlines.

go
type DeduplicationStore interface {
    Claim(ctx context.Context, key, token, fingerprint string, lease time.Duration) (owned, completed bool, err error)
    Renew(ctx context.Context, key, token string, lease time.Duration) error
    Complete(ctx context.Context, key, token, fingerprint string, retention time.Duration) error
    Release(ctx context.Context, key, token string) error
}

typeDeduplicator

Deduplicator protects both handler and manual acknowledgment paths. It does not own the store. A successful handler is completed durably before source acknowledgment. The handler context is canceled on lease/ownership loss. External side effects and completion are separate writes: use a transactional inbox with the side effect for atomic database processing. Handlers must honor cancellation; no deduplication store can fence arbitrary external effects.

go
type Deduplicator interface {
    Handle(context.Context, types.Message, types.HandlerFunc) error
    HandleBatch(context.Context, []types.Message, types.BatchHandlerFunc) error
}

typeDeduplicatorBuilder

DeduplicatorBuilder is the configuration interface for a Deduplicator. Namespace and consumer group are required. The default key excludes physical streams/partitions, suppressing duplicates throughout that application scope. Include a destination in the namespace for per-destination deduplication.

go
type DeduplicatorBuilder interface {
    WithStore(DeduplicationStore) DeduplicatorBuilder
    WithNamespace(string) DeduplicatorBuilder
    WithConsumerGroup(string) DeduplicatorBuilder
    WithLease(time.Duration) DeduplicatorBuilder
    WithRetention(time.Duration) DeduplicatorBuilder
    WithRetryDelay(time.Duration) DeduplicatorBuilder
    WithKeyExtractor(func(types.Message) (string, error)) DeduplicatorBuilder
    WithFingerprintExtractor(func(types.Message) (string, error)) DeduplicatorBuilder
    Build() (Deduplicator, error)
}

funcNewDeduplicator

go
func NewDeduplicator() DeduplicatorBuilder

NewDeduplicator creates a builder. Retention must be explicitly configured; after that finite window expires a repeated event may run its handler again. Do not modify a builder after Build or while its runtime is in use.

typeMessage

Message exposes the physical stream offset and delivery ownership context.

go
type Message interface {
    types.Message
    Offset() int64
    Partition() string
    Context() context.Context
}

typeProducer

Producer publishes confirmed messages and exposes durable publication replay. An application must ensure exclusive ownership of its producer_name across processes.

go
type Producer interface {
    types.Producer
    ResolvePartition(context.Context, types.Message) (string, error)
    LastPublishingID(context.Context, string) (int64, error)
    PublishSequence(context.Context, string, int64, types.Message) error
    ReplaySequence(context.Context, string, int64, types.Message) error
}

typeProducerBuilder

ProducerBuilder configures the producer. Build initializes this same value.

go
type ProducerBuilder interface {
    WithConfig(*viper.Viper) ProducerBuilder
    WithFilterExtractor(func(types.Message) string) ProducerBuilder
    WithRoutingExtractor(func(types.Message) string) ProducerBuilder
    Build(context.Context) (Producer, error)
    WithURL(string) ProducerBuilder
    WithURLs([]string) ProducerBuilder
    WithDestination(string) ProducerBuilder
    WithProducerName(string) ProducerBuilder
    WithStreamMode(string) ProducerBuilder
    WithRoutingStrategy(string) ProducerBuilder
    WithPartitions(int) ProducerBuilder
    WithBindingKeys([]string) ProducerBuilder
    WithAutoDeclare(bool) ProducerBuilder
    WithAutoReconnect(bool) ProducerBuilder
    WithFilterHeader(string) ProducerBuilder
}

funcNewProducerBuilder

go
func NewProducerBuilder() ProducerBuilder

typePublishUncertainError

PublishUncertainError means the broker may have accepted this publication. Replay a durable publication using the original identity, partition and sequence.

go
type PublishUncertainError interface {
    Partition() string
    Sequence() int64
    Unwrap() error
    // contains filtered or unexported methods
}

typeTransportBuilder

TransportBuilder constructs the low-level transport; Dial opens the connection. Build returns this builder as a Transport. Configure before Build only.

go
type TransportBuilder interface {
    WithConfig(*viper.Viper) TransportBuilder
    WithURL(string) TransportBuilder
    WithURLs([]string) TransportBuilder
    Build(context.Context) (types.Transport, error)
}

funcNewTransportBuilder

go
func NewTransportBuilder() TransportBuilder

Generated by gomarkdoc

↑ ↓ move↵ open/ opens search anywhere