rabbitstream

package module
v1.1.1 Latest Latest
Warning

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

 Go to latest Published: Sep 9, 2026 License: MIT Imports: 10 Imported by: 0

README

rabbitstream

CI CodeQL Coverage Mutation Documentation Go Reference Release Go License

rabbitstream is a policy layer for durable RabbitMQ Streams and Super Streams workloads. It provides bounded publishing, consumption, replay, inspection, failure handling, lifecycle, and observations without implementing the RabbitMQ Streams protocol.

The root policy library, RabbitMQ transport, and OpenTelemetry adapter are stable v1 modules. Their minimum supported Go version is 1.26.6; repository verification currently tests exactly Go 1.26.6.

For the shared construction, ownership, lifecycle, and integration vocabulary, see the versioned v1.4.0 Golib ecosystem index and its Integration and data movement family.

Use rabbitstream Use queue
retained event histories jobs and commands
independent consumer progress competing workers
replay and backlog catch-up process-and-remove delivery
partitioned event ingestion delayed or retried work

Install

go get github.com/faustbrian/go-rabbitmq-streams@v1
go get github.com/faustbrian/go-rabbitmq-streams/adapters/rabbitmq@v1
go get github.com/faustbrian/go-rabbitmq-streams/adapters/otel@v1

Install only the modules an application imports. The root module defines vendor-neutral policy and transport seams. The adapters/rabbitmq module adapts the supported RabbitMQ Go Streams client, and adapters/otel provides optional OpenTelemetry metrics and W3C Trace Context propagation. The released rabbitmq and otel paths remain supported as deprecated, behavior-compatible facades. New code should use the canonical paths; existing code can migrate by changing only its import and module path. See the migration guide.

Package Use
github.com/faustbrian/go-rabbitmq-streams Define bounded messages, delivery policy, producer and consumer contracts, replay, and inspection.
github.com/faustbrian/go-rabbitmq-streams/adapters/rabbitmq Open RabbitMQ Streams protocol resources through the supported Go client.
github.com/faustbrian/go-rabbitmq-streams/adapters/otel Translate observations to caller-owned OpenTelemetry metrics and propagate W3C Trace Context.

Producer

Topology is operator-owned and must exist before the application starts. The zero security value uses verified TLS 1.2 or newer.

connection := rabbitstream.ConnectionConfig{
    Endpoints: []rabbitstream.Endpoint{{
        Host: "rabbitmq.internal",
        Port: 5551,
    }},
    VirtualHost: "/",
    Credentials: rabbitstream.StaticCredentials(
        os.Getenv("RABBITMQ_STREAM_USER"),
        []byte(os.Getenv("RABBITMQ_STREAM_PASSWORD")),
    ),
    Security: rabbitstream.SecurityConfig{
        TLS: &tls.Config{ServerName: "rabbitmq.internal"},
    },
}

producer, err := rabbitmq.OpenProducer(
    ctx,
    connection,
    rabbitstream.ProducerConfig{
        Stream: "tracking.events",
        Policy: rabbitstream.ProducerPolicy{MaxOutstanding: 256},
    },
)
if err != nil {
    return err
}
defer producer.Shutdown(context.Background())

result, err := producer.Publish(ctx, rabbitstream.Message{
    Stream:        "tracking.events",
    MessageID:     eventID,
    CorrelationID: shipmentID,
    Payload:       encoded,
})
if err != nil {
    if result.State == rabbitstream.DeliveryAmbiguous {
        // The broker may have persisted the message. Reconcile before retrying.
    }
    return err
}

Consumer

consumer, err := rabbitmq.OpenConsumer(
    ctx,
    connection,
    rabbitstream.ConsumerConfig{
        Stream:       "tracking.events",
        ConsumerName: "tracking-projector-v1",
        Start: rabbitstream.StartPosition{
            Kind: rabbitstream.OffsetStartStored,
        },
        Policy: rabbitstream.ConsumerPolicy{
            MaxConcurrency:  1,
            HandlerTimeout:  30 * time.Second,
            FailureStrategy: rabbitstream.FailureStop,
        },
    },
)
if err != nil {
    return err
}
defer consumer.Shutdown(context.Background())

return consumer.Run(ctx, func(
    handlerCtx context.Context,
    message rabbitstream.Message,
) error {
    return applyTrackingEvent(handlerCtx, message.Payload)
})

The consumer stores progress only after successful handling. External side effects and RabbitMQ offsets do not share a transaction, so handlers must be idempotent or reconcile duplicates.

Root producer and consumer values own bounded background work after successful construction and require a bounded Shutdown call. Each caller independently bounds its wait while all callers observe one shared terminal cleanup result. The deprecated context-taking Close method remains behavior-compatible. Stop new publications, cancel and join consumer runs, close consumers, then close producers. The RabbitMQ adapter owns the protocol connections and sessions it opens. The OpenTelemetry adapter starts no goroutines, owns no provider or exporter, and exposes no Close or Shutdown; callers flush and shut down their providers after all stream clients stop emitting observations.

Guarantees

  • Publisher confirmations do not prove downstream processing.
  • A lost confirmation is ambiguous, not a definite non-send.
  • Consumption is at least once; crashes can redeliver completed work.
  • Ordering is per stream or Super Stream partition, never global.
  • Replay uses independent cursors and does not advance live consumers.
  • Retry and dead-letter publication complete before the source offset advances.

Read the documentation index before production adoption. It covers API contracts, delivery guarantees, operations, capacity validation, interoperability, and Kafka migration.

The compiler-checked root examples demonstrate producer and consumer construction. The RabbitMQ and OpenTelemetry modules provide their own transport examples and telemetry example.

Compatibility

The public API follows semantic versioning. Pin an exact version for every imported module and upgrade them as one reviewed set. Production capacity and availability depend on the actual RabbitMQ cluster, topology, payloads, handlers, and deployment environment.

License

MIT. See LICENSE.

Documentation

Overview

Package rabbitstream provides bounded RabbitMQ Streams and Super Streams policy for retained event distribution. It is not a work-queue abstraction; applications requiring competing job workers and process-and-remove semantics should use the repository's queue module.

Index

Examples

Constants

View Source 
const (
	// MaxEndpoints bounds connection-rotation state and diagnostic output.
	MaxEndpoints = 16
	// MaxReconnectAttempts is the largest accepted finite reconnect budget.
	MaxReconnectAttempts = 32
)
View Source 
const (

	// FailureSourceStreamMetadata identifies the original stream on a retry or
	// dead-letter publication.
	FailureSourceStreamMetadata = "rabbitstream.source.stream"
	// FailureSourcePartitionMetadata identifies the original backing stream.
	FailureSourcePartitionMetadata = "rabbitstream.source.partition"
	// FailureSourceOffsetMetadata identifies the original numeric offset.
	FailureSourceOffsetMetadata = "rabbitstream.source.offset"
	// FailureAttemptMetadata records the bounded handler attempt count.
	FailureAttemptMetadata = "rabbitstream.failure.attempt"
	// FailureCategoryMetadata carries a safe low-cardinality failure class.
	FailureCategoryMetadata = "rabbitstream.failure.category"
)
View Source 
const (

	// MaxSuperStreamPartitions bounds topology snapshots and producer fan-out.
	MaxSuperStreamPartitions = 1024
)
View Source 
const (

	// RoutingKeyMetadata is the reserved AMQP message-annotation key used by
	// adapters to preserve a Message routing key on the wire.
	RoutingKeyMetadata = "x-rabbitstream-routing-key"
)

Variables

View Source 
var (
	// ErrInvalidConfiguration matches CategoryInvalidConfiguration.
	ErrInvalidConfiguration = categoryError{CategoryInvalidConfiguration}
	// ErrValidation matches CategoryValidation.
	ErrValidation = categoryError{CategoryValidation}
	// ErrClosed matches CategoryClosed.
	ErrClosed = categoryError{CategoryClosed}
	// ErrCanceled matches CategoryCanceled.
	ErrCanceled = categoryError{CategoryCanceled}
	// ErrTimeout matches CategoryTimeout.
	ErrTimeout = categoryError{CategoryTimeout}
	// ErrAuthentication matches CategoryAuthentication.
	ErrAuthentication = categoryError{CategoryAuthentication}
	// ErrAuthorization matches CategoryAuthorization.
	ErrAuthorization = categoryError{CategoryAuthorization}
	// ErrConnection matches CategoryConnection.
	ErrConnection = categoryError{CategoryConnection}
	// ErrStreamUnavailable matches CategoryStreamUnavailable.
	ErrStreamUnavailable = categoryError{CategoryStreamUnavailable}
	// ErrPartitionUnavailable matches CategoryPartitionUnavailable.
	ErrPartitionUnavailable = categoryError{CategoryPartitionUnavailable}
	// ErrBrokerRejected matches CategoryBrokerRejected.
	ErrBrokerRejected = categoryError{CategoryBrokerRejected}
	// ErrMessageTooLarge matches CategoryMessageTooLarge.
	ErrMessageTooLarge = categoryError{CategoryMessageTooLarge}
	// ErrPublishAmbiguous matches CategoryPublishAmbiguous.
	ErrPublishAmbiguous = categoryError{CategoryPublishAmbiguous}
	// ErrConfirmation matches CategoryConfirmation.
	ErrConfirmation = categoryError{CategoryConfirmation}
	// ErrRetentionGap matches CategoryRetentionGap.
	ErrRetentionGap = categoryError{CategoryRetentionGap}
	// ErrReplayRange matches CategoryReplayRange.
	ErrReplayRange = categoryError{CategoryReplayRange}
	// ErrOffset matches CategoryOffset.
	ErrOffset = categoryError{CategoryOffset}
	// ErrHandler matches CategoryHandler.
	ErrHandler = categoryError{CategoryHandler}
	// ErrFatal matches CategoryFatal.
	ErrFatal = categoryError{CategoryFatal}
)

Functions

func ValidateBatch 

func ValidateBatch(messages []Message, limits Limits) error

ValidateBatch validates every message and bounds the count and aggregate payload-plus-metadata bytes. It does not mutate or retain its input.

Types

type BatchMessageHandler 

type BatchMessageHandler func(context.Context, []Message) error

BatchMessageHandler handles one borrowed, single-partition delivery batch. Messages remain ordered by delivery offset and are valid only for the call unless Retain is used on each message that must escape it.

type BatchPolicy 

type BatchPolicy struct {
	// MaxMessages bounds records supplied to one batch invocation.
	MaxMessages int
	// MaxWait bounds how long a partial batch waits for another record.
	MaxWait time.Duration
}

BatchPolicy bounds one consumer handler invocation. Successful batches store only their terminal offset, so the crash-redelivery window is at most MaxMessages records per partition.

type BatchPublishError 

type BatchPublishError struct {
	// Index is the first failed message, or -1 for aggregate validation failure.
	Index int
	// Cause is the stable underlying per-message or validation failure.
	Cause error
}

BatchPublishError identifies the first message that did not complete. Later results remain DeliveryNotSent. Index is -1 when aggregate batch validation failed before an individual message could be selected.

func (*BatchPublishError) Error 

func (err *BatchPublishError) Error() string

Error renders only the failed index and never message contents.

func (*BatchPublishError) Unwrap 

func (err *BatchPublishError) Unwrap() error

Unwrap preserves the stable per-message failure category.

type ConnectionConfig 

type ConnectionConfig struct {
	// Endpoints is the ordered, finite broker rotation set.
	Endpoints []Endpoint
	// VirtualHost selects the RabbitMQ virtual host; empty selects the client default.
	VirtualHost string
	// Credentials is resolved again for each connection attempt.
	Credentials CredentialProvider
	// Security defines verified transport policy.
	Security SecurityConfig
	// ConnectTimeout bounds complete session establishment, including retries.
	ConnectTimeout time.Duration
	// RPCTimeout bounds one RabbitMQ Streams RPC attempt.
	RPCTimeout time.Duration
	// Heartbeat configures the negotiated connection heartbeat.
	Heartbeat time.Duration
	// MaxReconnectAttempts bounds endpoint attempts within one connection budget.
	MaxReconnectAttempts int
	// InitialReconnectDelay is the first delay between connection attempts.
	InitialReconnectDelay time.Duration
	// MaxReconnectBackoff caps exponential connection backoff.
	MaxReconnectBackoff time.Duration
	// Observer receives bounded best-effort lifecycle signals.
	Observer Observer
}

ConnectionConfig defines finite connection, RPC, heartbeat, and reconnect budgets. Endpoint order is the caller's preferred rotation order.

func (ConnectionConfig) Normalized 

func (config ConnectionConfig) Normalized() (ConnectionConfig, error)

Normalized validates the policy and returns an owned configuration with all finite defaults applied. Credential resolution remains deferred.

func (ConnectionConfig) Validate 

func (config ConnectionConfig) Validate() error

Validate rejects unsafe or unbounded connection policy.

type Consumer 

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

Consumer owns bounded workers and broker offset lifecycles. A stable partition-to-worker assignment preserves sequential handling within each backing stream while allowing independent partitions to run concurrently.

func NewConsumer 

func NewConsumer(config ConsumerConfig, transport ConsumerTransport) (*Consumer, error)

NewConsumer constructs a durable policy wrapper and takes ownership of transport after a successful return.

Example 
transport := &singleDeliveryTransport{message: rabbitstream.Message{
	Stream:    "tracking.events",
	Partition: "tracking.events",
	Offset:    42,
	HasOffset: true,
	Payload:   []byte("opaque event bytes"),
}}
consumer, err := rabbitstream.NewConsumer(rabbitstream.ConsumerConfig{
	Stream:       "tracking.events",
	ConsumerName: "tracking-projector-v1",
}, transport)
if err != nil {
	panic(err)
}
defer func() { _ = consumer.Shutdown(context.Background()) }()

ctx, cancel := context.WithCancel(context.Background())
err = consumer.Run(ctx, func(_ context.Context, message rabbitstream.Message) error {
	fmt.Println(message.Stream, message.Offset)
	cancel()
	return nil
})
if err == nil {
	panic("consumer unexpectedly returned nil")
}

Output:
tracking.events 42

func (*Consumer) Close deprecated 

func (consumer *Consumer) Close(ctx context.Context) error

Close preserves the released context-bounded shutdown contract.

Deprecated: use Shutdown.

func (*Consumer) Pause 

func (consumer *Consumer) Pause() error

Pause stops admission from the transport before the next read. Already admitted messages remain bounded and continue processing. Pause is idempotent and may be selected before Run starts.

func (*Consumer) Resume 

func (consumer *Consumer) Resume() error

Resume permits transport admission after Pause. It is idempotent.

func (*Consumer) Run 

func (consumer *Consumer) Run(ctx context.Context, handler MessageHandler) error

Run consumes until cancellation or the first transport, handler, or offset failure. It never stores an offset before successful handler completion.

func (*Consumer) RunBatch 

func (consumer *Consumer) RunBatch(
	ctx context.Context,
	policy BatchPolicy,
	handler BatchMessageHandler,
) error

RunBatch consumes single-partition batches until cancellation or the first transport, handler, publication, or offset failure. A partial batch held at cancellation is left unstored for safe redelivery.

func (*Consumer) Shutdown added in v1.1.0 

func (consumer *Consumer) Shutdown(ctx context.Context) error

Shutdown cancels an active Run, waits within caller and policy bounds, and closes the owned transport exactly once. Each caller's context bounds only that caller's wait; cleanup continues once started and every caller that observes completion receives the same terminal cleanup result.

type ConsumerConfig 

type ConsumerConfig struct {
	// Stream selects one direct stream when SuperStream is empty.
	Stream string
	// SuperStream selects a logical partitioned stream when Stream is empty.
	SuperStream string
	// ConsumerName is the stable broker offset-tracking identity.
	ConsumerName string
	// Start selects initial delivery position before stored progress advances.
	Start StartPosition
	// Limits bounds delivered messages and retained metadata.
	Limits Limits
	// Policy bounds concurrency, handler time, retry, offset storage, and close.
	Policy ConsumerPolicy
	// Observer receives bounded best-effort lifecycle signals.
	Observer Observer

	// FailurePublisher confirms retry or dead-letter publication before offset storage.
	FailurePublisher FailurePublisher
	// RetryStream is required by FailureRetryStream.
	RetryStream string
	// DeadLetterStream is required by FailureDeadLetter.
	DeadLetterStream string
}

ConsumerConfig binds a durable named consumer to one stream or Super Stream. Broker offset storage records the last successfully handled message; it is not transactional with handler side effects.

func (ConsumerConfig) Normalized 

func (config ConsumerConfig) Normalized() (ConsumerConfig, error)

Normalized validates ConsumerConfig and applies finite defaults.

type ConsumerPolicy 

type ConsumerPolicy struct {
	// MaxConcurrency bounds workers across independent partitions.
	MaxConcurrency int
	// HandlerTimeout bounds one handler or batch invocation.
	HandlerTimeout time.Duration
	// CloseTimeout bounds graceful consumer draining.
	CloseTimeout time.Duration
	// OffsetStoreEveryMessages bounds the processed-but-not-stored crash window.
	OffsetStoreEveryMessages int
	// FailureStrategy selects stop, in-process retry, retry stream, or dead letter.
	FailureStrategy FailureStrategy
	// Retry configures bounded in-process handler retries.
	Retry RetryPolicy
}

ConsumerPolicy bounds handler execution, retry, and shutdown.

type ConsumerTransport 

type ConsumerTransport interface {
	// Next returns the next owned delivery or a stable terminal error.
	Next(context.Context) (Message, error)
	// StoreOffset submits the last successfully handled partition offset.
	StoreOffset(context.Context, string, uint64) error
	// Close releases every transport-owned consumer and connection resource.
	Close() error
}

ConsumerTransport is the narrow client-adapter boundary. Next and StoreOffset must honor context cancellation. A nil StoreOffset error means the client accepted the one-way broker command; RabbitMQ does not confirm that command transactionally, so a crash may cause safe redelivery.

type CredentialProvider 

type CredentialProvider interface {
	// Credentials returns a fresh owned snapshot and honors ctx cancellation.
	Credentials(context.Context) (Credentials, error)
}

CredentialProvider supplies a fresh credential snapshot for each connection attempt. Implementations must honor cancellation and must not render secrets.

type Credentials 

type Credentials struct {
	// Username is the RabbitMQ authentication identity.
	Username string
	// Password is an owned secret snapshot that callers must not log.
	Password []byte
}

Credentials are an owned authentication snapshot. A provider must return a fresh password slice on every call so credential rotation can occur between connection attempts.

type DeduplicationPolicy 

type DeduplicationPolicy uint8

DeduplicationPolicy selects whether publishing IDs participate in RabbitMQ broker-side producer deduplication. It is not an end-to-end exactly-once guarantee. 

const (
	// DeduplicationNone leaves broker-side publishing-ID deduplication disabled.
	DeduplicationNone DeduplicationPolicy = iota
	// DeduplicationPublishingID requires a stable producer name and an explicit
	// publishing ID on every message.
	DeduplicationPublishingID
)

type DeliveryResult 

type DeliveryResult struct {
	// State expresses whether the outcome is unsent, confirmed, rejected, or ambiguous.
	State DeliveryState
	// Stream is the direct target when publishing without a Super Stream.
	Stream string
	// SuperStream is the logical target for partitioned publishing.
	SuperStream string
	// Partition is the backing stream selected by routing when known.
	Partition string
	// PublishingID is the broker sequence associated with this result.
	PublishingID uint64
}

DeliveryResult is the per-message publish outcome. Ordering is scoped to Stream or Partition; no global order exists across Super Stream partitions.

type DeliveryState 

type DeliveryState uint8

DeliveryState describes the caller-visible certainty of one publish. 

const (
	// DeliveryNotSent means validation, cancellation, closure, or local send
	// failure happened before the transport accepted the message.
	DeliveryNotSent DeliveryState = iota
	// DeliveryConfirmed means the broker confirmed persistence.
	DeliveryConfirmed
	// DeliveryRejected means the broker definitively rejected the message.
	DeliveryRejected
	// DeliveryAmbiguous means transmission occurred but confirmation was not
	// observed before cancellation, timeout, or connection loss.
	DeliveryAmbiguous
)

type DependencyHealth 

type DependencyHealth struct {
	// State is the bounded dependency result.
	State DependencyState
	// ObservedAt records when the diagnostic completed.
	ObservedAt time.Time
	// Category classifies an unavailable dependency without resource names.
	Category ErrorCategory
}

DependencyHealth is a bounded readiness/diagnostic result. RabbitMQ unavailability alone does not imply that the process should restart.

type DependencyState 

type DependencyState uint8

DependencyState separates dependency health from process liveness. 

const (
	// DependencyHealthy reports a successful bounded broker diagnostic.
	DependencyHealthy DependencyState = iota
	// DependencyUnavailable reports a failed dependency diagnostic without process failure.
	DependencyUnavailable
)

type Endpoint 

type Endpoint struct {
	// Host is a DNS name or IP literal without credentials or a scheme.
	Host string
	// Port is the RabbitMQ Streams listener port.
	Port uint16
}

Endpoint identifies one RabbitMQ Streams listener. Credentials are kept out of endpoint values so they cannot be exposed by URI formatting.

type ErrorCategory 

type ErrorCategory string

ErrorCategory is a stable, low-cardinality failure classification. It is safe for branching and diagnostics; it never contains broker or caller data. 

const (
	// CategoryInvalidConfiguration classifies rejected static policy.
	CategoryInvalidConfiguration ErrorCategory = "invalid_configuration"
	// CategoryValidation classifies rejected caller message or request data.
	CategoryValidation ErrorCategory = "validation"
	// CategoryClosed classifies operations attempted after lifecycle closure.
	CategoryClosed ErrorCategory = "closed"
	// CategoryCanceled classifies caller-requested cancellation.
	CategoryCanceled ErrorCategory = "canceled"
	// CategoryTimeout classifies an exhausted bounded operation deadline.
	CategoryTimeout ErrorCategory = "timeout"
	// CategoryAuthentication classifies rejected or unavailable credentials.
	CategoryAuthentication ErrorCategory = "authentication"
	// CategoryAuthorization classifies insufficient broker permissions.
	CategoryAuthorization ErrorCategory = "authorization"
	// CategoryConnection classifies broker connectivity or session failure.
	CategoryConnection ErrorCategory = "connection"
	// CategoryStreamUnavailable classifies a missing or unavailable stream.
	CategoryStreamUnavailable ErrorCategory = "stream_unavailable"
	// CategoryPartitionUnavailable classifies a missing backing stream.
	CategoryPartitionUnavailable ErrorCategory = "partition_unavailable"
	// CategoryBrokerRejected classifies a definitive publish rejection.
	CategoryBrokerRejected ErrorCategory = "broker_rejected"
	// CategoryMessageTooLarge classifies a message exceeding package or broker limits.
	CategoryMessageTooLarge ErrorCategory = "message_too_large"
	// CategoryPublishAmbiguous classifies transmission without delivery certainty.
	CategoryPublishAmbiguous ErrorCategory = "publish_ambiguous"
	// CategoryConfirmation classifies an invalid or failed confirmation.
	CategoryConfirmation ErrorCategory = "confirmation"
	// CategoryRetentionGap classifies requested history no longer retained.
	CategoryRetentionGap ErrorCategory = "retention_gap"
	// CategoryReplayRange classifies an invalid or incomplete replay range.
	CategoryReplayRange ErrorCategory = "replay_range"
	// CategoryOffset classifies broker offset tracking failure.
	CategoryOffset ErrorCategory = "offset"
	// CategoryHandler classifies application handler failure.
	CategoryHandler ErrorCategory = "handler"
	// CategoryFatal classifies a permanent client failure requiring intervention.
	CategoryFatal ErrorCategory = "fatal"
)

type FailurePublisher 

type FailurePublisher interface {
	// Publish must return a broker-confirmed outcome before source progress advances.
	Publish(context.Context, Message) (DeliveryResult, error)
}

FailurePublisher is satisfied by Producer and deliberately exposes only the confirmed publish operation needed before source-offset advancement.

type FailureStrategy 

type FailureStrategy uint8

FailureStrategy controls handler failure without implying queue-style NACK or broker redelivery behavior. 

const (
	// FailureStop stops the ordering scope without advancing its offset.
	FailureStop FailureStrategy = iota
	// FailureRetry retries the handler in process within RetryPolicy bounds.
	FailureRetry
	// FailureRetryStream publishes a new record to an explicit retry stream.
	FailureRetryStream
	// FailureDeadLetter publishes a new record to an explicit dead-letter stream.
	FailureDeadLetter
)

type InspectionRequest 

type InspectionRequest struct {
	// Stream selects one direct stream when SuperStream is empty.
	Stream string
	// SuperStream selects a logical partitioned stream when Stream is empty.
	SuperStream string
	// ConsumerName optionally requests its broker-stored offset.
	ConsumerName string
}

InspectionRequest selects exactly one read-only stream or Super Stream target. ConsumerName optionally requests its broker-stored offset.

func (InspectionRequest) Validate 

func (request InspectionRequest) Validate(limits Limits) error

Validate checks target and diagnostic identity bounds.

type InspectionResult 

type InspectionResult struct {
	// SuperStream is the logical target, empty for a direct stream request.
	SuperStream string
	// Partitions contains one bounded snapshot per direct or backing stream.
	Partitions []StreamInspection
	// ObservedAt records when the snapshot was assembled.
	ObservedAt time.Time
}

InspectionResult is a bounded snapshot. Super Stream partition ordering is the broker routing topology ordering observed for this request.

type Inspector 

type Inspector interface {
	// Inspect returns a bounded read-only topology and offset snapshot.
	Inspect(context.Context, InspectionRequest) (InspectionResult, error)
	// Health reports dependency state without changing broker topology.
	Health(context.Context) DependencyHealth
}

Inspector exposes read-only broker topology and offset diagnostics.

type Limits 

type Limits struct {
	// MaxStreamNameBytes bounds stream, Super Stream, partition, and consumer names.
	MaxStreamNameBytes int
	// MaxRoutingKeyBytes bounds routing and partition keys.
	MaxRoutingKeyBytes int
	// MaxPayloadBytes bounds one message payload before transport allocation.
	MaxPayloadBytes int
	// MaxMetadataEntries bounds all header, property, and broker metadata entries.
	MaxMetadataEntries int
	// MaxMetadataKeyBytes bounds one metadata key.
	MaxMetadataKeyBytes int
	// MaxMetadataValueBytes bounds one metadata value or standard property.
	MaxMetadataValueBytes int
	// MaxMetadataBytes bounds aggregate message metadata.
	MaxMetadataBytes int
	// MaxBatchMessages bounds one synchronous batch.
	MaxBatchMessages int
	// MaxBatchBytes bounds aggregate payload and metadata bytes in a batch.
	MaxBatchBytes int
	// MaxBufferedMessages bounds asynchronous admission.
	MaxBufferedMessages int
}

Limits bounds all caller-controlled material retained by core operations. Values are bytes unless their name states otherwise.

func DefaultLimits 

func DefaultLimits() Limits

DefaultLimits returns conservative finite defaults. Callers may lower them; higher values must remain compatible with broker frame and resource policy.

type Message 

type Message struct {
	// Stream is the direct stream target or consumed backing stream.
	Stream string
	// SuperStream is the logical partitioned target when direct Stream is empty.
	SuperStream string
	// Partition identifies the selected backing stream after routing or delivery.
	Partition string
	// RoutingKey selects a Super Stream partition and scopes keyed ordering.
	RoutingKey string
	// PublishingID is the broker deduplication sequence when HasPublishingID is true.
	PublishingID uint64
	// HasPublishingID distinguishes an explicit zero ID from no publishing ID.
	HasPublishingID bool
	// Offset is the delivered stream offset when HasOffset is true.
	Offset uint64
	// HasOffset distinguishes offset zero from an unset producer message offset.
	HasOffset bool
	// Timestamp is application message time, not broker receipt time.
	Timestamp time.Time
	// ContentType is the language-neutral payload media type.
	ContentType string
	// MessageID is the stable application message identity.
	MessageID string
	// CorrelationID links related application messages without affecting routing.
	CorrelationID string
	// Payload is borrowed during synchronous calls and copied by Retain.
	Payload []byte
	// Headers preserves ordered message annotations other than RoutingKey.
	Headers []MetadataEntry
	// Properties preserves ordered application properties.
	Properties []MetadataEntry
	// BrokerMetadata contains bounded delivery diagnostics supplied by an adapter.
	BrokerMetadata []MetadataEntry
}

Message is the language-neutral byte and ordered-metadata model used for publishing. Exactly one of Stream and SuperStream must be set. Payload, Headers, and Properties are borrowed for the duration of a synchronous call; asynchronous retention must use Retain.

func (Message) Retain 

func (message Message) Retain() Message

Retain returns a deep owned copy of byte slices and ordered metadata.

func (Message) Validate 

func (message Message) Validate(limits Limits) error

Validate checks target identity and every retained byte dimension.

func (Message) ValidateDelivery 

func (message Message) ValidateDelivery(limits Limits) error

ValidateDelivery checks the owned broker-delivery shape used by consumers, replay, and telemetry. A Super Stream delivery carries both its logical SuperStream and its backing Stream/Partition identity.

type MessageHandler 

type MessageHandler func(context.Context, Message) error

MessageHandler handles one borrowed delivery. The message is valid only for the call unless Retain is used.

type MetadataEntry 

type MetadataEntry struct {
	// Key is a bounded application or broker metadata name.
	Key string
	// Value is borrowed unless the enclosing Message is retained.
	Value []byte
}

MetadataEntry preserves application-property and header order, including duplicate keys. Key and Value are borrowed unless the enclosing Message is retained by calling Retain.

type Observation 

type Observation struct {
	// Kind identifies the stable signal.
	Kind ObservationKind
	// Count is a bounded event or message count.
	Count uint64
	// Bytes is a bounded payload-byte total.
	Bytes uint64
	// Value is a kind-specific bounded scalar such as offset or lag.
	Value uint64
	// Duration is a kind-specific bounded elapsed time.
	Duration time.Duration
	// Category classifies a failure without high-cardinality details.
	Category ErrorCategory
}

Observation contains only bounded scalar data. It deliberately excludes stream names, routing keys, message IDs, payloads, headers, and credentials.

type ObservationKind 

type ObservationKind string

ObservationKind is a stable low-cardinality lifecycle signal. 

const (
	// ObservationConnectionConnecting reports a bounded connection attempt.
	ObservationConnectionConnecting ObservationKind = "connection_connecting"
	// ObservationConnectionReady reports an established usable session.
	ObservationConnectionReady ObservationKind = "connection_ready"
	// ObservationConnectionLost reports loss of an established session.
	ObservationConnectionLost ObservationKind = "connection_lost"
	// ObservationReconnectAttempt reports an attempt to restore a lost session.
	ObservationReconnectAttempt ObservationKind = "reconnect_attempt"
	// ObservationAuthenticationError reports credential rejection or resolution failure.
	ObservationAuthenticationError ObservationKind = "authentication_error"
	// ObservationPublishAttempt reports one caller publish attempt.
	ObservationPublishAttempt ObservationKind = "publish_attempt"
	// ObservationPublishConfirmed reports one definitive broker confirmation.
	ObservationPublishConfirmed ObservationKind = "publish_confirmed"
	// ObservationPublishRejected reports one definitive broker rejection.
	ObservationPublishRejected ObservationKind = "publish_rejected"
	// ObservationPublishAmbiguous reports a sent message without observed certainty.
	ObservationPublishAmbiguous ObservationKind = "publish_ambiguous"
	// ObservationPublishError reports a publish failure before confirmation.
	ObservationPublishError ObservationKind = "publish_error"
	// ObservationConsumerMessage reports one delivered message.
	ObservationConsumerMessage ObservationKind = "consumer_message"
	// ObservationHandlerSuccess reports a successful handler invocation.
	ObservationHandlerSuccess ObservationKind = "handler_success"
	// ObservationHandlerError reports a failed handler invocation.
	ObservationHandlerError ObservationKind = "handler_error"
	// ObservationHandlerRetry reports a bounded in-process retry.
	ObservationHandlerRetry ObservationKind = "handler_retry"
	// ObservationRetryStreamPublished reports confirmed retry-stream publication.
	ObservationRetryStreamPublished ObservationKind = "retry_stream_published"
	// ObservationDeadLetterPublished reports confirmed dead-letter publication.
	ObservationDeadLetterPublished ObservationKind = "dead_letter_published"
	// ObservationFailurePublishError reports failed retry or dead-letter publication.
	ObservationFailurePublishError ObservationKind = "failure_publish_error"
	// ObservationOffsetStoreAccepted reports client acceptance of an offset-store command.
	ObservationOffsetStoreAccepted ObservationKind = "offset_store_accepted"
	// ObservationStreamEndOffset reports an observed retained end offset.
	ObservationStreamEndOffset ObservationKind = "stream_end_offset"
	// ObservationConsumerLag reports bounded numeric backlog.
	ObservationConsumerLag ObservationKind = "consumer_lag"
	// ObservationReplayProgress reports one replay progress point.
	ObservationReplayProgress ObservationKind = "replay_progress"
	// ObservationProducerShutdown reports bounded producer close duration.
	ObservationProducerShutdown ObservationKind = "producer_shutdown"
	// ObservationConsumerShutdown reports bounded consumer close duration.
	ObservationConsumerShutdown ObservationKind = "consumer_shutdown"
)

type Observer 

type Observer interface {
	// Observe receives one bounded signal and must return promptly.
	Observe(Observation)
}

Observer receives best-effort lifecycle signals. Implementations must not block indefinitely; their panics are contained and never affect delivery.

type OffsetStartKind 

type OffsetStartKind uint8

OffsetStartKind selects the first delivery requested from RabbitMQ Streams. 

const (
	// OffsetStartStored resumes at the named consumer's broker-stored offset, so
	// the last stored delivery may be observed again under at-least-once policy.
	OffsetStartStored OffsetStartKind = iota
	// OffsetStartBeginning starts at the first retained message.
	OffsetStartBeginning
	// OffsetStartEnd starts with messages appended after the consumer attaches.
	OffsetStartEnd
	// OffsetStartExplicit starts at an exact numeric offset.
	OffsetStartExplicit
	// OffsetStartTimestamp starts at the first message at or after Timestamp.
	OffsetStartTimestamp
)

type Operation 

type Operation string

Operation identifies a stable package operation without including resource names or other high-cardinality data. 

const (
	// OperationConnect establishes or restores a broker session.
	OperationConnect Operation = "connect"
	// OperationPublish sends and confirms a message.
	OperationPublish Operation = "publish"
	// OperationConsume receives, handles, or stores consumer progress.
	OperationConsume Operation = "consume"
	// OperationReplay reads an isolated retained range.
	OperationReplay Operation = "replay"
	// OperationInspect reads broker topology or offsets.
	OperationInspect Operation = "inspect"
	// OperationClose drains and releases owned resources.
	OperationClose Operation = "close"
)

type OperationError 

type OperationError struct {
	// Operation is the stable operation that failed.
	Operation Operation
	// Category is the stable low-cardinality failure class.
	Category ErrorCategory
	// Cause preserves programmatic detail and may require redaction before logging.
	Cause error
}

OperationError preserves a programmatically inspectable cause while its rendered form exposes only a stable operation and category. Callers must not log the unwrapped cause unless they have independently established it is safe.

func (*OperationError) Error 

func (err *OperationError) Error() string

Error renders a bounded operation and category without the underlying cause.

func (*OperationError) Is 

func (err *OperationError) Is(target error) bool

Is matches the sentinel corresponding to Category.

func (*OperationError) Unwrap 

func (err *OperationError) Unwrap() error

Unwrap preserves the original cause for errors.Is and errors.As.

type Producer 

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

Producer owns one bounded publishing lifecycle. It is safe for concurrent use. Message confirmation order can differ from caller goroutine completion order; RabbitMQ ordering remains scoped to the selected backing stream.

func NewProducer 

func NewProducer(config ProducerConfig, transport ProducerTransport) (*Producer, error)

NewProducer constructs the policy wrapper around a transport. The wrapper owns transport after a successful return and must be closed.

Example 
producer, err := rabbitstream.NewProducer(rabbitstream.ProducerConfig{
	Stream: "tracking.events",
}, confirmedTransport{})
if err != nil {
	panic(err)
}
defer func() { _ = producer.Shutdown(context.Background()) }()

result, err := producer.Publish(context.Background(), rabbitstream.Message{
	Stream:    "tracking.events",
	MessageID: "event-123",
	Payload:   []byte("opaque event bytes"),
})
if err != nil {
	panic(err)
}
fmt.Println(result.State == rabbitstream.DeliveryConfirmed)

Output:
true

func (*Producer) Close deprecated 

func (producer *Producer) Close(ctx context.Context) error

Close preserves the released context-bounded shutdown contract.

Deprecated: use Shutdown.

func (*Producer) Publish 

func (producer *Producer) Publish(ctx context.Context, message Message) (result DeliveryResult, err error)

Publish validates and owns message bytes until a definitive confirmation or an explicitly ambiguous outcome. Cancellation before transport admission is definite; cancellation after Send succeeds is ambiguous.

func (*Producer) PublishAsync 

func (producer *Producer) PublishAsync(
	ctx context.Context,
	message Message,
) (<-chan PublishOutcome, error)

PublishAsync retains message before returning and admits at most Limits.MaxBufferedMessages asynchronous operations. Cancellation before admission is definite; cancellation after transport send remains ambiguous.

func (*Producer) PublishBatch 

func (producer *Producer) PublishBatch(
	ctx context.Context,
	messages []Message,
) ([]DeliveryResult, error)

PublishBatch validates the entire bounded batch before allocating per-message results or sending anything, then publishes in input order. It stops at the first per-message failure so partial delivery is explicit.

func (*Producer) Shutdown added in v1.1.0 

func (producer *Producer) Shutdown(ctx context.Context) error

Shutdown stops admission, waits for admitted publishes within their finite confirmation bounds, then closes the transport. It is idempotent and safe for concurrent use. Each caller's context bounds only that caller's wait; cleanup continues once started and every caller that observes completion receives the same terminal cleanup result.

type ProducerConfig 

type ProducerConfig struct {
	// Stream selects one direct stream when SuperStream is empty.
	Stream string
	// SuperStream selects a logical partitioned stream when Stream is empty.
	SuperStream string
	// RoutingStrategy selects the reviewed key-to-partition algorithm.
	RoutingStrategy RoutingStrategy
	// ExpectedPartitions rejects an unexpected non-zero Super Stream partition count.
	ExpectedPartitions int
	// Limits bounds messages, batches, metadata, and asynchronous buffering.
	Limits Limits
	// Policy bounds confirmations, outstanding sends, deduplication, and close.
	Policy ProducerPolicy
	// Observer receives bounded best-effort lifecycle signals.
	Observer Observer
}

ProducerConfig binds one producer to exactly one stream or Super Stream. Super Stream publishing always requires a non-empty routing key so ordering decisions cannot silently fall back to an unstable default.

func (ProducerConfig) Normalized 

func (config ProducerConfig) Normalized() (ProducerConfig, error)

Normalized validates ProducerConfig and returns finite defaults.

type ProducerPolicy 

type ProducerPolicy struct {
	// MaxOutstanding bounds sent messages awaiting confirmation.
	MaxOutstanding int
	// ConfirmationTimeout bounds how long a sent message awaits broker certainty.
	ConfirmationTimeout time.Duration
	// CloseTimeout bounds confirmation draining during shutdown.
	CloseTimeout time.Duration
	// Deduplication selects explicit broker publishing-ID deduplication policy.
	Deduplication DeduplicationPolicy
	// ProducerName is the stable broker deduplication identity when enabled.
	ProducerName string
}

ProducerPolicy bounds confirmations, in-flight memory, and shutdown. A producer name activates broker-side deduplication and must therefore remain stable and unique for the target stream.

type ProducerTransport 

type ProducerTransport interface {
	// Send admits one owned message and invokes confirm at most once.
	Send(context.Context, Message, func(TransportConfirmation)) error
	// Close releases all transport resources and is idempotent.
	Close() error
}

ProducerTransport is the narrow adapter boundary implemented by the nested RabbitMQ client module. Send receives an owned Message and must invoke confirm at most once. It may use ctx only for work known to precede transmission, such as bounded connection admission. A nil Send error means transmission may have occurred and later cancellation must be represented as ambiguous.

type PublishOutcome 

type PublishOutcome struct {
	// Result is the caller-visible delivery certainty.
	Result DeliveryResult
	// Err is the stable operation failure associated with Result.
	Err error
}

PublishOutcome is the terminal result of one accepted asynchronous publish. The result channel returned by PublishAsync always receives exactly one outcome and is then closed.

type ReplayCursor 

type ReplayCursor interface {
	// Next returns the next retained message or io.EOF after the requested range.
	Next(context.Context) (Message, error)
	// Close cancels cursor work and releases all owned resources.
	Close() error
}

ReplayCursor is an isolated, non-offset-storing retained-message cursor.

type ReplayDelivery 

type ReplayDelivery struct {
	// Message is the retained delivery in partition offset order.
	Message Message
	// SideEffectsAllowed repeats the caller's explicit replay authority.
	SideEffectsAllowed bool
}

ReplayDelivery makes the caller's explicit side-effect policy visible to the handler. Replayers never mutate a live consumer's stored offset.

type ReplayHandler 

type ReplayHandler func(context.Context, ReplayDelivery) error

ReplayHandler processes one retained message in partition offset order.

type ReplayRequest 

type ReplayRequest struct {
	// Stream selects one direct stream when SuperStream is empty.
	Stream string
	// SuperStream selects a logical partitioned stream when Stream is empty.
	SuperStream string
	// Partition is the single backing stream replayed for a Super Stream.
	Partition string
	// ExpectedPartitions is the ordered Super Stream topology the caller
	// approved for this replay. Super Stream replay rejects a different live
	// topology because routing and ordering assumptions may have changed.
	ExpectedPartitions []string
	// Start selects the first requested retained message.
	Start StartPosition
	// EndOffset optionally requires an inclusive exact terminal offset.
	EndOffset *uint64
	// Checkpoint optionally selects a caller-owned exact replay start. Callers
	// storing the last completed offset must advance it before reuse.
	Checkpoint *uint64
	// AllowSideEffects makes application side-effect authority explicit to handlers.
	AllowSideEffects bool
}

ReplayRequest identifies an isolated replay. Super Stream replay is always partition-specific because no global cross-partition order exists.

type ReplaySource 

type ReplaySource interface {
	// RetainedRange returns exact currently available offsets for the request.
	RetainedRange(context.Context, ReplayRequest) (RetainedRange, error)
	// Open creates an isolated cursor that never stores live-consumer progress.
	Open(context.Context, ReplayRequest) (ReplayCursor, error)
}

ReplaySource supplies exact retained-range inspection and isolated cursors.

type Replayer 

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

Replayer validates exact retention boundaries before invoking application code and never owns or advances normal consumer progress.

func NewReplayer 

func NewReplayer(limits Limits, source ReplaySource, observer Observer) (*Replayer, error)

NewReplayer validates finite message bounds and a replay source.

func (*Replayer) Inspect 

func (replayer *Replayer) Inspect(ctx context.Context, request ReplayRequest) (RetainedRange, error)

Inspect returns the exact currently retained range without opening a cursor.

func (*Replayer) Run 

func (replayer *Replayer) Run(
	ctx context.Context,
	request ReplayRequest,
	handler ReplayHandler,
) (runErr error)

Run replays an exact retained range. Explicit missing starts are retention gaps; missing requested ends are incomplete replay ranges.

type RetainedRange 

type RetainedRange struct {
	// FirstOffset is the earliest retained message offset when Empty is false.
	FirstOffset uint64
	// LastOffset is the latest retained message offset when Empty is false.
	LastOffset uint64
	// Empty distinguishes no retained messages from a range beginning at zero.
	Empty bool
}

RetainedRange is an exact snapshot of offsets available when replay opens. Implementations must not substitute committed chunk IDs or estimates for LastOffset. Empty distinguishes an empty stream from offset zero.

type RetryPolicy 

type RetryPolicy struct {
	// MaxAttempts includes the initial handler invocation.
	MaxAttempts int
	// InitialBackoff is the delay before the first retry.
	InitialBackoff time.Duration
	// MaxBackoff caps exponential retry delay.
	MaxBackoff time.Duration
}

RetryPolicy bounds in-process handler attempts and backoff. MaxAttempts includes the initial attempt.

type RoutingStrategy 

type RoutingStrategy uint8

RoutingStrategy selects a reviewed Super Stream routing contract. 

const (
	// RoutingHash uses the RabbitMQ client-compatible Murmur3 strategy. Ordering
	// is stable only while the ordered backing-partition topology is unchanged.
	RoutingHash RoutingStrategy = iota
)

type SecurityConfig 

type SecurityConfig struct {
	// Mode selects verified TLS or explicit development-only plaintext.
	Mode SecurityMode
	// TLS is cloned during normalization and must retain peer verification.
	TLS *tls.Config
	// contains filtered or unexported fields
}

SecurityConfig owns transport security policy. The zero value is verified TLS with a TLS 1.2 minimum. InsecureSkipVerify is always rejected.

func DevelopmentPlaintextSecurity 

func DevelopmentPlaintextSecurity() SecurityConfig

DevelopmentPlaintextSecurity opts into unencrypted local development. It must never be used for production credentials or traffic.

type SecurityMode 

type SecurityMode uint8

SecurityMode selects verified TLS or an explicit local-development-only plaintext connection. 

const (
	// SecurityTLS requires certificate-verified TLS and is the zero-value mode.
	SecurityTLS SecurityMode = iota
	// SecurityPlaintext is accepted only through DevelopmentPlaintextSecurity.
	SecurityPlaintext
)

type StartPosition 

type StartPosition struct {
	// Kind selects stored, retained beginning, live end, exact offset, or timestamp.
	Kind OffsetStartKind
	// Offset is used only with OffsetStartExplicit.
	Offset uint64
	// Timestamp is used only with OffsetStartTimestamp.
	Timestamp time.Time
}

StartPosition models RabbitMQ Streams offsets directly. Offset and Timestamp are used only by their corresponding kinds.

type StreamInspection 

type StreamInspection struct {
	// Stream identifies the direct or backing stream inspected.
	Stream string
	// Exists reports whether the broker found the stream.
	Exists bool
	// FirstOffset is the earliest retained offset when the broker exposes it.
	FirstOffset *uint64
	// LastOffset is the exact retained end offset when available.
	LastOffset *uint64
	// CommittedChunkID is broker chunk metadata and is not an exact end offset.
	CommittedChunkID *uint64
	// StoredOffset is the named consumer's broker-stored progress when requested.
	StoredOffset *uint64
	// Lag is the exact non-negative distance to LastOffset when both are known.
	Lag *uint64
}

StreamInspection reports broker facts without interpreting committed chunk IDs as exact stream end offsets.

type StringerCredentialProvider 

type StringerCredentialProvider interface {
	CredentialProvider
	fmt.Stringer
}

StringerCredentialProvider combines the provider contract with a safe diagnostic identity. It exists to make the safety of built-in providers explicit without requiring custom providers to implement String.

func StaticCredentials 

func StaticCredentials(username string, password []byte) StringerCredentialProvider

StaticCredentials copies password immediately and returns a provider that returns owned copies. Applications requiring rotation should implement CredentialProvider instead.

type TransportConfirmation 

type TransportConfirmation struct {
	// Confirmed reports a definitive broker confirmation.
	Confirmed bool
	// BrokerRejected reports a definitive broker rejection.
	BrokerRejected bool
	// Ambiguous reports transmission without observed confirmation or rejection.
	Ambiguous bool
	// PublishingID is the broker sequence returned by the adapter.
	PublishingID uint64
	// Partition is the backing stream that handled the publish.
	Partition string
	// Cause is a safe programmatic transport failure.
	Cause error
}

TransportConfirmation is the stable result supplied by a ProducerTransport. Cause must already be safe for programmatic exposure and diagnostics.

Source Files

View all Source files

Directories

Path Synopsis
adapters
otel module
rabbitmq module
otel module
rabbitmq module

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.