Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
- API
  
  Reference documentation for the pkg.go.dev API
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

rabbitstream

package module

Go to main page

Versions in this module

v1

v1.1.1

Sep 9, 2026

v1.1.0

Sep 9, 2026

Changes in this version

type Consumer

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

type Producer

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

v1.0.0

Aug 25, 2026

Changes in this version

+ const FailureAttemptMetadata

+ const FailureCategoryMetadata

+ const FailureSourceOffsetMetadata

+ const FailureSourcePartitionMetadata

+ const FailureSourceStreamMetadata

+ const MaxEndpoints

+ const MaxReconnectAttempts

+ const MaxSuperStreamPartitions

+ const RoutingKeyMetadata

+ var ErrAuthentication = categoryError

+ var ErrAuthorization = categoryError

+ var ErrBrokerRejected = categoryError

+ var ErrCanceled = categoryError

+ var ErrClosed = categoryError

+ var ErrConfirmation = categoryError

+ var ErrConnection = categoryError

+ var ErrFatal = categoryError

+ var ErrHandler = categoryError

+ var ErrInvalidConfiguration = categoryError

+ var ErrMessageTooLarge = categoryError

+ var ErrOffset = categoryError

+ var ErrPartitionUnavailable = categoryError

+ var ErrPublishAmbiguous = categoryError

+ var ErrReplayRange = categoryError

+ var ErrRetentionGap = categoryError

+ var ErrStreamUnavailable = categoryError

+ var ErrTimeout = categoryError

+ var ErrValidation = categoryError

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

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

+ type BatchPolicy struct

+ MaxMessages int

+ MaxWait time.Duration

+ type BatchPublishError struct

+ Cause error

+ Index int

+ func (err *BatchPublishError) Error() string

+ func (err *BatchPublishError) Unwrap() error

+ type ConnectionConfig struct

+ ConnectTimeout time.Duration

+ Credentials CredentialProvider

+ Endpoints []Endpoint

+ Heartbeat time.Duration

+ InitialReconnectDelay time.Duration

+ MaxReconnectAttempts int

+ MaxReconnectBackoff time.Duration

+ Observer Observer

+ RPCTimeout time.Duration

+ Security SecurityConfig

+ VirtualHost string

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

+ func (config ConnectionConfig) Validate() error

+ type Consumer struct

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

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

+ func (consumer *Consumer) Pause() error

+ func (consumer *Consumer) Resume() error

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

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

+ type ConsumerConfig struct

+ ConsumerName string

+ DeadLetterStream string

+ FailurePublisher FailurePublisher

+ Limits Limits

+ Observer Observer

+ Policy ConsumerPolicy

+ RetryStream string

+ Start StartPosition

+ Stream string

+ SuperStream string

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

+ type ConsumerPolicy struct

+ CloseTimeout time.Duration

+ FailureStrategy FailureStrategy

+ HandlerTimeout time.Duration

+ MaxConcurrency int

+ OffsetStoreEveryMessages int

+ Retry RetryPolicy

+ type ConsumerTransport interface

+ Close func() error

+ Next func(context.Context) (Message, error)

+ StoreOffset func(context.Context, string, uint64) error

+ type CredentialProvider interface

+ Credentials func(context.Context) (Credentials, error)

+ type Credentials struct

+ Password []byte

+ Username string

+ type DeduplicationPolicy uint8

+ const DeduplicationNone

+ const DeduplicationPublishingID

+ type DeliveryResult struct

+ Partition string

+ PublishingID uint64

+ State DeliveryState

+ Stream string

+ SuperStream string

+ type DeliveryState uint8

+ const DeliveryAmbiguous

+ const DeliveryConfirmed

+ const DeliveryNotSent

+ const DeliveryRejected

+ type DependencyHealth struct

+ Category ErrorCategory

+ ObservedAt time.Time

+ State DependencyState

+ type DependencyState uint8

+ const DependencyHealthy

+ const DependencyUnavailable

+ type Endpoint struct

+ Host string

+ Port uint16

+ type ErrorCategory string

+ const CategoryAuthentication

+ const CategoryAuthorization

+ const CategoryBrokerRejected

+ const CategoryCanceled

+ const CategoryClosed

+ const CategoryConfirmation

+ const CategoryConnection

+ const CategoryFatal

+ const CategoryHandler

+ const CategoryInvalidConfiguration

+ const CategoryMessageTooLarge

+ const CategoryOffset

+ const CategoryPartitionUnavailable

+ const CategoryPublishAmbiguous

+ const CategoryReplayRange

+ const CategoryRetentionGap

+ const CategoryStreamUnavailable

+ const CategoryTimeout

+ const CategoryValidation

+ type FailurePublisher interface

+ Publish func(context.Context, Message) (DeliveryResult, error)

+ type FailureStrategy uint8

+ const FailureDeadLetter

+ const FailureRetry

+ const FailureRetryStream

+ const FailureStop

+ type InspectionRequest struct

+ ConsumerName string

+ Stream string

+ SuperStream string

+ func (request InspectionRequest) Validate(limits Limits) error

+ type InspectionResult struct

+ ObservedAt time.Time

+ Partitions []StreamInspection

+ SuperStream string

+ type Inspector interface

+ Health func(context.Context) DependencyHealth

+ Inspect func(context.Context, InspectionRequest) (InspectionResult, error)

+ type Limits struct

+ MaxBatchBytes int

+ MaxBatchMessages int

+ MaxBufferedMessages int

+ MaxMetadataBytes int

+ MaxMetadataEntries int

+ MaxMetadataKeyBytes int

+ MaxMetadataValueBytes int

+ MaxPayloadBytes int

+ MaxRoutingKeyBytes int

+ MaxStreamNameBytes int

+ func DefaultLimits() Limits

+ type Message struct

+ BrokerMetadata []MetadataEntry

+ ContentType string

+ CorrelationID string

+ HasOffset bool

+ HasPublishingID bool

+ Headers []MetadataEntry

+ MessageID string

+ Offset uint64

+ Partition string

+ Payload []byte

+ Properties []MetadataEntry

+ PublishingID uint64

+ RoutingKey string

+ Stream string

+ SuperStream string

+ Timestamp time.Time

+ func (message Message) Retain() Message

+ func (message Message) Validate(limits Limits) error

+ func (message Message) ValidateDelivery(limits Limits) error

+ type MessageHandler func(context.Context, Message) error

+ type MetadataEntry struct

+ Key string

+ Value []byte

+ type Observation struct

+ Bytes uint64

+ Category ErrorCategory

+ Count uint64

+ Duration time.Duration

+ Kind ObservationKind

+ Value uint64

+ type ObservationKind string

+ const ObservationAuthenticationError

+ const ObservationConnectionConnecting

+ const ObservationConnectionLost

+ const ObservationConnectionReady

+ const ObservationConsumerLag

+ const ObservationConsumerMessage

+ const ObservationConsumerShutdown

+ const ObservationDeadLetterPublished

+ const ObservationFailurePublishError

+ const ObservationHandlerError

+ const ObservationHandlerRetry

+ const ObservationHandlerSuccess

+ const ObservationOffsetStoreAccepted

+ const ObservationProducerShutdown

+ const ObservationPublishAmbiguous

+ const ObservationPublishAttempt

+ const ObservationPublishConfirmed

+ const ObservationPublishError

+ const ObservationPublishRejected

+ const ObservationReconnectAttempt

+ const ObservationReplayProgress

+ const ObservationRetryStreamPublished

+ const ObservationStreamEndOffset

+ type Observer interface

+ Observe func(Observation)

+ type OffsetStartKind uint8

+ const OffsetStartBeginning

+ const OffsetStartEnd

+ const OffsetStartExplicit

+ const OffsetStartStored

+ const OffsetStartTimestamp

+ type Operation string

+ const OperationClose

+ const OperationConnect

+ const OperationConsume

+ const OperationInspect

+ const OperationPublish

+ const OperationReplay

+ type OperationError struct

+ Category ErrorCategory

+ Cause error

+ Operation Operation

+ func (err *OperationError) Error() string

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

+ func (err *OperationError) Unwrap() error

+ type Producer struct

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

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

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

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

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

+ type ProducerConfig struct

+ ExpectedPartitions int

+ Limits Limits

+ Observer Observer

+ Policy ProducerPolicy

+ RoutingStrategy RoutingStrategy

+ Stream string

+ SuperStream string

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

+ type ProducerPolicy struct

+ CloseTimeout time.Duration

+ ConfirmationTimeout time.Duration

+ Deduplication DeduplicationPolicy

+ MaxOutstanding int

+ ProducerName string

+ type ProducerTransport interface

+ Close func() error

+ Send func(context.Context, Message, func(TransportConfirmation)) error

+ type PublishOutcome struct

+ Err error

+ Result DeliveryResult

+ type ReplayCursor interface

+ Close func() error

+ Next func(context.Context) (Message, error)

+ type ReplayDelivery struct

+ Message Message

+ SideEffectsAllowed bool

+ type ReplayHandler func(context.Context, ReplayDelivery) error

+ type ReplayRequest struct

+ AllowSideEffects bool

+ Checkpoint *uint64

+ EndOffset *uint64

+ ExpectedPartitions []string

+ Partition string

+ Start StartPosition

+ Stream string

+ SuperStream string

+ type ReplaySource interface

+ Open func(context.Context, ReplayRequest) (ReplayCursor, error)

+ RetainedRange func(context.Context, ReplayRequest) (RetainedRange, error)

+ type Replayer struct

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

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

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

+ type RetainedRange struct

+ Empty bool

+ FirstOffset uint64

+ LastOffset uint64

+ type RetryPolicy struct

+ InitialBackoff time.Duration

+ MaxAttempts int

+ MaxBackoff time.Duration

+ type RoutingStrategy uint8

+ const RoutingHash

+ type SecurityConfig struct

+ Mode SecurityMode

+ TLS *tls.Config

+ func DevelopmentPlaintextSecurity() SecurityConfig

+ type SecurityMode uint8

+ const SecurityPlaintext

+ const SecurityTLS

+ type StartPosition struct

+ Kind OffsetStartKind

+ Offset uint64

+ Timestamp time.Time

+ type StreamInspection struct

+ CommittedChunkID *uint64

+ Exists bool

+ FirstOffset *uint64

+ Lag *uint64

+ LastOffset *uint64

+ StoredOffset *uint64

+ Stream string

+ type StringerCredentialProvider interface

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

+ type TransportConfirmation struct

+ Ambiguous bool

+ BrokerRejected bool

+ Cause error

+ Confirmed bool

+ Partition string

+ PublishingID uint64