streams

package
v0.65.0 Latest Latest
Warning

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

 Go to latest Published: Sep 16, 2026 License: MIT Imports: 28 Imported by: 0

Documentation

Overview

Package streams publishes to and consumes RabbitMQ streams over the native stream protocol (default port 5552, rabbitmq_stream plugin).

Importing this package (a blank import is enough) registers the lane with app so a configured messaging.streams.uri starts the manager. A process that never imports it carries none of the vendor client (ADR-091).

It complements the AMQP 0.9.1 lane in the parent messaging package: streams declared as AMQP queues (x-queue-type: stream) can be consumed there, but server-side offset tracking and single active consumer need this protocol. Publishing is synchronous and confirmed — see wiki/streams.md.

Index

Constants

View Source 
const (
	// MaxRetryAttempts and MaxRetryWait bound what a declared policy may ask of one
	// partition. The waits happen inside that partition's own delivery callback, so
	// a long policy is a stall every OTHER tenant on the partition pays for; work
	// that needs more patience than this belongs in the hold, which parks one
	// tenant and lets the partition move.
	MaxRetryAttempts = 10
	MaxRetryWait     = time.Minute
)
View Source 
const (

	// The span attributes this lane adds on top of the four the pipeline sets.
	// AttrConsumerName is exported because a hold's gauges carry the same key: an
	// operator joins the backlog to the consumer's own spans and logs by it, so the
	// two must never drift apart silently.
	AttrConsumerName = "messaging.consumer.name"
)
View Source 
const TenantStampProperty = tenantstamp.Header

TenantStampProperty is the application property the framework writes the publishing context's tenant into, and the one a consumer reads it back from. It is the same entry the classic lane uses as an AMQP 0.9.1 header.

Variables

View Source 
var (
	// ErrPayloadUndecodable reports a message body that could not be decoded into
	// the consumer's payload type. Match it with errors.Is.
	ErrPayloadUndecodable = errors.New("streams: payload could not be decoded")

	// ErrPayloadInvalid reports a payload that decoded but failed struct
	// validation. Match it with errors.Is.
	ErrPayloadInvalid = errors.New("streams: payload failed validation")
)
View Source 
var (
	// ErrPublisherNotStarted reports a publish on a publisher Manager.Start has
	// not bound to a client producer yet.
	ErrPublisherNotStarted = errors.New("streams: publisher not started; Manager.Start binds it")
	// ErrPublisherClosed reports a publish on a publisher whose manager has shut
	// down, and is what an in-flight publish is resolved with when that happens.
	ErrPublisherClosed = errors.New("streams: publisher closed")
)
View Source 
var DefaultHoldRetry = RetryOptions{
	MaxAttempts:    3,
	InitialBackoff: 200 * time.Millisecond,
	MaxBackoff:     2 * time.Second,
}

DefaultHoldRetry is the policy a holding consumer gets when it declares none. A consumer that does not hold keeps today's single attempt unless it asks for a policy itself.

View Source 
var ErrTenantStampConflict = tenantstamp.ErrConflict

ErrTenantStampConflict reports a publish whose tenant stamp was supplied by the caller. The framework is the stamp's only writer on both lanes, so this is the same error value messaging.ErrTenantStampConflict names — errors.Is holds whichever lane raised it.

Functions

func DeclareTypedConsumer added in v0.62.0 

func DeclareTypedConsumer[T any](decls *Declarations, opts *ConsumerOptions, fn func(context.Context, T) error)

DeclareTypedConsumer registers a stream consumer whose handler decodes the message body into T and validates it against the struct's `validate` tags before calling fn. It is the streams-lane mirror of messaging.DeclareTypedConsumer, and T is inferred from fn.

There is deliberately no exported NewTypedHandler counterpart on this lane. The declaration carries a poison screen alongside the handler, and a handler handed to DeclareConsumer as a plain Handler could not carry one — a typed consumer assembled that way would park undecodable bodies in the hold, which is exactly what ADR-092 forbids.

Failure semantics: a body that does not decode, or decodes but fails validation, is deterministic poison. It is not retried in place whatever Retry says, it is never parked when Hold is set, and its offset is not committed — the lane skips it, exactly as ADR-059 settles any failure on a consumer that does not hold. Match the two modes with errors.Is against ErrPayloadUndecodable and ErrPayloadInvalid.

It panics on a nil decls, a nil opts, or an opts that already carries a Handler — all three are declaration-time wiring mistakes, which this lane already fails startup on for duplicate registrations.

func DeclareTypedConsumerWithMeta added in v0.62.0 

func DeclareTypedConsumerWithMeta[T any](decls *Declarations, opts *ConsumerOptions, fn func(context.Context, T, *Message) error)

DeclareTypedConsumerWithMeta is DeclareTypedConsumer for an fn that also needs the delivery itself — msg.Offset, msg.Stream and msg.Properties. Same panics, same poison semantics; msg.Data is the body fn's payload was decoded from.

func DeclareTypedSuperStreamConsumer added in v0.62.0 

func DeclareTypedSuperStreamConsumer[T any](decls *Declarations, opts *SuperStreamConsumerOptions, fn func(context.Context, T) error)

DeclareTypedSuperStreamConsumer is DeclareTypedConsumer over every partition of a super stream. fn is called CONCURRENTLY across partitions — see Handler — so it must be safe for concurrent use.

func DeclareTypedSuperStreamConsumerWithMeta added in v0.62.0 

func DeclareTypedSuperStreamConsumerWithMeta[T any](decls *Declarations, opts *SuperStreamConsumerOptions, fn func(context.Context, T, *Message) error)

DeclareTypedSuperStreamConsumerWithMeta is DeclareTypedSuperStreamConsumer for an fn that also needs the delivery. msg.Stream names the PARTITION the message arrived on, not the super stream.

func Permanent added in v0.61.0 

func Permanent(err error) error

Permanent is the handler's claim that retrying is pointless: the delivery ends on the attempt that produced err whatever the policy allows. Permanent(nil) is nil.

Types

type ConsumerOptions 

type ConsumerOptions struct {
	// Stream is the stream to consume; it must be declared in the same Declarations.
	Stream string
	// Name is the consumer/group name. Required: it is the key the broker stores
	// offsets under and, with SAC, the group identity.
	Name string
	// Start is where to begin when the broker holds no stored offset for Name.
	Start OffsetStart
	// SAC enables single active consumer: only one member of the group receives
	// messages at a time (RabbitMQ 3.11+).
	SAC bool
	// Handler processes each message. Required.
	Handler Handler
	// Retry bounds in-place re-invocation of Handler after it returns an error.
	// Nil is one attempt, unless Hold is set — see DefaultHoldRetry. A policy may
	// ask for at most MaxRetryAttempts attempts and MaxRetryWait of total waiting,
	// because the waits run inside the partition's own delivery callback; a failure
	// that needs more patience than that belongs in the hold.
	Retry *RetryOptions
	// Hold parks a failed delivery per tenant so the tenant's later messages wait
	// behind it while the rest of the partition keeps flowing.
	//
	// SECURITY: the tenant is read from the producer-written stamp, which is
	// identification and not authorization (ADR-039's rule for HTTP resolvers,
	// applied here). A producer that can publish to this stream therefore chooses
	// which tenant a message is held under, and a failure it induces holds THAT
	// tenant until the hold drains. Isolating producers — per-tenant credentials
	// or vhosts — stays the deployment's job, and it matters more here than for a
	// delivery that is merely mis-attributed.
	Hold bool
	// TenantOptional lets this consumer run a delivery that carries no tenant stamp
	// (a control-plane consumer). Default false: a consumer that needs a tenant fails
	// the delivery closed rather than running the handler without one. It never
	// admits a stamp that is present but unusable.
	TenantOptional bool
}

ConsumerOptions declares one stream consumer.

type Declarations 

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

Declarations collects the stream infrastructure and consumers a set of modules declares. It is populated once at startup, validated, then replayed by Manager.

func NewDeclarations 

func NewDeclarations() *Declarations

NewDeclarations creates an empty declaration store.

func (*Declarations) DeclareConsumer 

func (d *Declarations) DeclareConsumer(opts *ConsumerOptions)

DeclareConsumer registers a stream consumer. Panics on a nil options pointer, and if the same (stream, name) pair was already declared. Both are programming errors: a nil declaration would consume nothing at all, and a duplicate would otherwise start two members of the same offset group inside one process.

func (*Declarations) DeclarePublisher 

func (d *Declarations) DeclarePublisher(opts *PublisherOptions) *Publisher

DeclarePublisher registers a publisher on a plain stream and returns the handle to publish through. The handle is inert until Manager.Start binds it.

Panics on a nil options pointer, and if the same stream already has a publisher. Both are programming errors: a nil declaration would publish nowhere, and one publisher per target per process is the same contract the consumer side enforces — a second one is a wiring mistake, not a fan-out.

func (*Declarations) DeclareStream 

func (d *Declarations) DeclareStream(name string, spec *StreamSpec)

DeclareStream registers a stream. A nil spec leaves retention to the broker. Re-declaring the same name with an identical spec is a no-op; a conflicting spec is reported by Validate.

func (*Declarations) DeclareSuperStream 

func (d *Declarations) DeclareSuperStream(name string, partitions int, spec *StreamSpec)

DeclareSuperStream registers a super stream of partitions partitions. A nil spec leaves retention to the broker; a non-nil one applies to every partition. Re-declaring the same name identically is a no-op; a conflicting partition count or spec is reported by Validate.

func (*Declarations) DeclareSuperStreamConsumer 

func (d *Declarations) DeclareSuperStreamConsumer(opts *SuperStreamConsumerOptions)

DeclareSuperStreamConsumer registers one consumer over every partition of a super stream. Panics on the same two programming errors as DeclareConsumer.

The declaration is always a single active consumer group — see SuperStreamConsumerOptions for why the choice is not the caller's.

func (*Declarations) DeclareSuperStreamPublisher 

func (d *Declarations) DeclareSuperStreamPublisher(opts *SuperStreamPublisherOptions) *Publisher

DeclareSuperStreamPublisher registers a publisher across every partition of a super stream and returns the handle to publish through. Panics on the same two programming errors as DeclarePublisher.

Every PublishMessage sent through the returned handle must carry a non-empty RoutingKey: it is what picks the partition.

func (*Declarations) IsEmpty 

func (d *Declarations) IsEmpty() bool

IsEmpty reports whether nothing was declared.

func (*Declarations) Stats 

func (d *Declarations) Stats() Stats

Stats returns the declaration counts.

func (*Declarations) Validate 

func (d *Declarations) Validate() error

Validate reports every problem in the store at once. Each declaration kind contributes its own errors, in declaration order, so a caller sees the whole picture rather than the first thing that went wrong.

type Handler 

type Handler func(ctx context.Context, msg *Message) error

Handler processes one stream message. An error is terminal for the MESSAGE, not for the stream: the failure is logged and counted, the offset is NOT committed, and consumption continues with the next message. Streams have no nack or redelivery, so handlers must be idempotent.

Calls are sequential within one stream — and within one partition of a super stream — but CONCURRENT across the partitions of a super stream, because each partition is a separate connection with its own delivery loop. A handler registered with DeclareSuperStreamConsumer must therefore be goroutine-safe.

type HeldMessage added in v0.61.0 

type HeldMessage = streamruntime.HeldMessage

The hold port lives on the streamruntime seam so inbox can implement it without importing this package (ADR-091). The aliases keep the streams spelling for lane callers.

type HoldLedger added in v0.61.0 

type HoldLedger = streamruntime.HoldLedger

The hold port lives on the streamruntime seam so inbox can implement it without importing this package (ADR-091). The aliases keep the streams spelling for lane callers.

type HoldReplayer added in v0.61.0 

type HoldReplayer = streamruntime.HoldReplayer

The hold port lives on the streamruntime seam so inbox can implement it without importing this package (ADR-091). The aliases keep the streams spelling for lane callers.

type Manager 

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

Manager owns the single stream-protocol Environment of a single-tenant service and the consumers and publishers started from its declarations. It is a plain struct on purpose (ADR-045): app/ consumes it concretely, so there is no interface to export and no second implementation to abstract over.

func NewManager 

func NewManager(opts ManagerOptions) *Manager

Panics on a nil Logger. Every consumer lifecycle, handler-failure and shutdown path dereferences it unguarded, so a wiring error would otherwise surface as a nil dereference on the first log line — mid-consumption, in production, from a goroutine the client owns and nothing recovers. Same fail-fast as httpclient.NewBuilder, and it keeps this constructor's single return value. pointer here is an incompatible change (apidiff), and the copy costs one allocation per manager, at startup.

func (*Manager) Close 

func (m *Manager) Close() error

Close stops the consumers and closes the environment. Idempotent.

func (*Manager) HoldConsumers added in v0.61.0 

func (m *Manager) HoldConsumers() []string

HoldConsumers names the running consumers that hold. A consumer that does not hold has nothing parked, so the drain never asks about it.

func (*Manager) Ready 

func (m *Manager) Ready() bool

Ready reports whether every started consumer and publisher is currently connected.

func (*Manager) ReloadHeld added in v0.61.0 

func (m *Manager) ReloadHeld(ctx context.Context, consumer string) error

ReloadHeld refreshes one consumer's held set from the ledger. A consumer this replica does not run is a no-op: the ledger is shared and deployments differ in which consumers they start.

The read happens here rather than in the caller because the generation that makes the replace safe must be taken BEFORE it. A caller that read the ledger first could only hand back a token taken after its own read, which compares equal to itself and erases any park that landed in between.

func (*Manager) Replay added in v0.61.0 

func (m *Manager) Replay(ctx context.Context, consumer string, msg *HeldMessage) error

Replay puts a held message back through the lane. It returns the handler's own error untouched: the drain decides what a failed replay means — defer the tenant — and this call settles nothing, because the row's fate is the drain's to write.

func (*Manager) SetTenantStamps added in v0.61.0 

func (m *Manager) SetTenantStamps(enabled bool)

NewManager creates a Manager. It performs no I/O: the environment is dialed by Start, so a service that declares no streams never opens a connection.

SetTenantStamps tells this manager's consumers to read the tenant stamp off each delivery and seed the handler context with it — true only under multitenant.enabled together with messaging.tenancy: shared.

It is a setter rather than a ManagerOptions field because that struct already sits at gocritic's hugeParam limit; one more field would force NewManager to take a pointer, which is an incompatible change to a shipped exported function. Call it before StartConsumers: a runner reads the flag when it is built.

func (*Manager) Start 

func (m *Manager) Start(ctx context.Context, decls *Declarations) error

Start dials the broker, replays the stream declarations, binds one reliable producer per declared publisher and starts one reliable consumer per declared consumer. Empty declarations are a no-op that dials nothing. Anything that fails to start stops what already came up and returns an error — the caller makes that fatal. A failure leaves nothing to dispose: the connection is closed before the error returns, so a caller that never calls Close does not leak it, and a retried Start cannot orphan the previous environment.

Start is not a resume: once it has dialed, it refuses to run again until Close disposes the environment. StopConsumers deliberately leaves that environment open, so redialing over it would orphan a connection the registered closer still owns.

ctx contributes its VALUES to every handler invocation, never its cancellation: consumers outlive the startup call that created them, and StopConsumers is what ends them. See consumeContext.

func (*Manager) Stats 

func (m *Manager) Stats() map[string]any

Stats reports the manager state for the readiness probe and /ready body.

func (*Manager) StopConsumers 

func (m *Manager) StopConsumers()

StopConsumers stops every consumer, then closes every publisher. Each consumer flushes its pending offset BEFORE closing, so a clean shutdown does not replay successfully handled messages. Publishers close AFTER them — a handler may publish on its way out — and every publish still awaiting a confirmation is resolved with ErrPublisherClosed rather than left to hang. Idempotent.

This is shutdown phase one, not a pause: the environment stays open for Close to dispose, and Start stays refused until then.

type ManagerOptions 

type ManagerOptions struct {
	// URI is the stream-protocol endpoint (rabbitmq-stream:// or rabbitmq-stream+tls://).
	URI string
	// AddressResolverHost and AddressResolverPort pin every connection to one
	// entry point. Required behind a load balancer, NAT, or Docker port mapping,
	// where the address the broker advertises is not reachable by the client.
	AddressResolverHost string
	AddressResolverPort int
	// OffsetStoreCount is how many successfully handled messages accumulate
	// before an offset is committed server-side.
	OffsetStoreCount int
	// OffsetStoreInterval is how long after the last commit a pending offset is
	// committed anyway.
	OffsetStoreInterval time.Duration
	// Logger receives consumer lifecycle and handler-failure events. Required.
	Logger logger.Logger
	// Hold is the ledger consumers declaring Hold park into. Nil means no consumer
	// may declare one — the framework wires it from the inbox's hold, which lives
	// on the control-plane database.
	Hold HoldLedger
}

ManagerOptions configures the stream-protocol Manager. The zero value of every tuning field applies its default.

type Message 

type Message struct {
	// Data is the first data section of the AMQP 1.0 message body.
	Data []byte
	// Offset is this message's offset within Stream.
	Offset int64
	// Stream is the stream (or, for super streams, the partition) it arrived on.
	Stream string
	// Properties carries the AMQP 1.0 application properties, nil when absent.
	Properties map[string]any
}

Message is the framework view of a stream delivery.

type OffsetStart 

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

OffsetStart is where a consumer begins when the broker holds no stored offset for its name. A stored offset always wins over it — see Manager.Start. The zero value is OffsetNext().

func OffsetAt 

func OffsetAt(offset int64) OffsetStart

OffsetAt starts at an absolute stream offset.

func OffsetFirst 

func OffsetFirst() OffsetStart

OffsetFirst starts at the oldest message still retained in the stream.

func OffsetLast 

func OffsetLast() OffsetStart

OffsetLast starts at the beginning of the last chunk written to the stream.

func OffsetNext 

func OffsetNext() OffsetStart

OffsetNext starts at the next message written after the consumer attaches.

func OffsetSince 

func OffsetSince(t time.Time) OffsetStart

OffsetSince starts at the first message stored at or after t.

type PayloadError added in v0.62.0 

type PayloadError struct {
	// Consumer is the declared consumer name the message was routed to. The
	// stream and offset are on the delivery's own log line and span; naming the
	// consumer is what tells two typed consumers of the same stream apart.
	Consumer string

	// Stage is where the failure happened. It is exported to label logs and
	// metrics with "decode" or "validate"; for control flow, match errors.Is
	// against ErrPayloadUndecodable or ErrPayloadInvalid instead.
	Stage PayloadStage
	// contains filtered or unexported fields
}

PayloadError describes why a stream message could not be turned into a typed payload. It is this lane's thin surface over payloaderr.Body, which owns the rendering rules and the SECURITY rationale behind them: Error() and Fields() are safe to log, Unwrap() is not.

It is also the lane's poison marker. A body that does not decode or does not validate fails the same way for every attempt and every replica, so a delivery carrying one is never retried in place and never parked in the hold — see ADR-092 and consumerRunner.parks.

func (*PayloadError) Error added in v0.62.0 

func (e *PayloadError) Error() string

func (*PayloadError) Fields added in v0.62.0 

func (e *PayloadError) Fields() []string

Fields returns the validator field namespaces that failed, e.g. ["OrderPayload.Amount"], with every bracketed span redacted. It is empty for decode failures and for a nil receiver.

func (*PayloadError) Is added in v0.62.0 

func (e *PayloadError) Is(target error) bool

Is maps the stage onto its sentinel so consumers can discriminate the two failure modes without reading Stage.

func (*PayloadError) Unwrap added in v0.62.0 

func (e *PayloadError) Unwrap() error

Unwrap exposes the underlying decode or validation error so errors.As can reach the cause.

SECURITY: the returned error MAY carry payload-derived text. It is the deliberate escape hatch for a caller that needs the raw diagnostic; logging it is opt-in and on the caller.

type PayloadStage added in v0.62.0 

type PayloadStage string

PayloadStage names the half of the typed-payload pipeline that failed. 

const (
	PayloadStageDecode   PayloadStage = PayloadStage(payloaderr.StageDecode)
	PayloadStageValidate PayloadStage = PayloadStage(payloaderr.StageValidate)
)

Stage values carried by PayloadError.Stage. A PayloadError whose Stage is none of these matches neither sentinel.

type PublishMessage 

type PublishMessage struct {
	// Data becomes the AMQP 1.0 data section of the message body.
	Data []byte
	// Properties carries AMQP 1.0 application properties; nil is fine. The map is
	// copied, so the caller's own map is never written to.
	Properties map[string]any
	// RoutingKey selects the partition of a super stream (murmur3 hash — the
	// RabbitMQ cross-client default). Required non-empty on a publisher declared
	// with DeclareSuperStreamPublisher, and must be empty on one declared with
	// DeclarePublisher, which targets a plain stream with no partitions to pick.
	RoutingKey string
}

PublishMessage is the framework view of an outbound stream message.

type Publisher 

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

Publisher publishes to one declared stream or super stream. It is obtained from Declarations.DeclarePublisher or DeclareSuperStreamPublisher at declaration time and is inert until Manager.Start binds it to a client producer. Safe for concurrent use.

func (*Publisher) Closed added in v0.64.0 

func (p *Publisher) Closed() bool

Closed reports whether this publisher's Close has run. A closed publisher is never ready and every Publish through it returns ErrPublisherClosed, so a caller draining a queue can tell a shutdown apart from a delivery failure without publishing to find out. Unlike Ready, this is not a broker-connectivity answer: a Close-then-Start cycle clears it (see bind).

func (*Publisher) Publish 

func (p *Publisher) Publish(ctx context.Context, msg *PublishMessage) error

Publish sends one message and blocks until the broker confirms it, ctx expires, or the publisher closes.

A ctx expiry is NOT proof of failure: the send may still be in flight and may still land. Delivery is at-least-once, so consumers must be idempotent. A nil return from the client's own Send proves nothing either — it swallows write errors — which is why the confirmation is what this waits for.

func (*Publisher) Ready added in v0.64.0 

func (p *Publisher) Ready() bool

Ready reports whether this publisher's producer is connected to the broker, as the client's HA layer reports it: only ha.StatusOpen is ready, so a producer that is reconnecting, closed, or not bound yet is not. A true answer is a snapshot, not a delivery guarantee — a broker failure can invalidate it before the next Publish.

type PublisherOptions 

type PublisherOptions struct {
	// Stream is the stream to publish to; it must be declared in the same Declarations.
	Stream string
}

PublisherOptions declares one plain-stream publisher.

type RetryOptions added in v0.61.0 

type RetryOptions struct {
	MaxAttempts int
	// InitialBackoff is the wait before the second attempt.
	InitialBackoff time.Duration
	// MaxBackoff caps the doubling. Zero means uncapped — the waits keep doubling
	// for the whole bound, which is what MaxRetryWait then has to contain.
	MaxBackoff time.Duration
}

RetryOptions bounds how often a failed delivery's handler is re-invoked in place before the lane settles on the failure. MaxAttempts counts the first attempt, so 1 retries nothing; the wait before attempt n (n >= 2) is InitialBackoff doubled n-2 times, capped at MaxBackoff.

type Stats 

type Stats struct {
	// Streams counts every declared stream, super streams included.
	Streams int
	// SuperStreams counts how many of Streams are partitioned super streams.
	SuperStreams int
	// Consumers counts every declared consumer, of either kind.
	Consumers int
	// Publishers counts every declared publisher.
	Publishers int
}

Stats summarizes a declaration store.

type StreamDeclarer added in v0.61.0 

type StreamDeclarer interface {
	DeclareStreams(decls *Declarations)
}

StreamDeclarer is the optional module interface the framework detects at startup. Modules that implement it have DeclareStreams called automatically. This used to live on app.StreamDeclarer; that name was removed so app does not import this package (ADR-091).

type StreamSpec 

type StreamSpec struct {
	// MaxAge discards segments older than this duration, truncated to whole
	// seconds. A non-zero sub-second value floors to 1s: second granularity is
	// RabbitMQ's, so anything briefer is inexpressible and would otherwise
	// discard the retention the caller asked for. Renders identically to
	// messaging.StreamQueueSpec.MaxAge, so both lanes treat a given value the
	// same way.
	MaxAge time.Duration
	// MaxLengthBytes caps the total retained size of the stream.
	MaxLengthBytes int64
	// MaxSegmentSizeBytes caps the size of one segment file.
	MaxSegmentSizeBytes int64
}

StreamSpec configures retention for a declared stream. Zero-value fields are omitted, leaving the broker's own defaults in place; a negative value is a declaration error that Declarations.Validate reports.

type SuperStreamConsumerOptions 

type SuperStreamConsumerOptions struct {
	// SuperStream is the super stream to consume; it must be declared in the same
	// Declarations.
	SuperStream string
	// Name is the consumer/group name. Required: the broker stores an offset per
	// partition under it, and it is the group identity partitions are distributed by.
	Name string
	// Start is where a partition begins when the broker holds no stored offset for
	// Name on that partition.
	Start OffsetStart
	// Handler processes each message. Required, and called concurrently across
	// partitions — see Handler.
	Handler Handler
	// Retry bounds in-place re-invocation of Handler after it returns an error.
	// Nil is one attempt, unless Hold is set — see DefaultHoldRetry. A policy may
	// ask for at most MaxRetryAttempts attempts and MaxRetryWait of total waiting,
	// because the waits run inside the partition's own delivery callback; a failure
	// that needs more patience than that belongs in the hold.
	Retry *RetryOptions
	// Hold parks a failed delivery per tenant so the tenant's later messages wait
	// behind it while the rest of the partition keeps flowing.
	//
	// SECURITY: the tenant is read from the producer-written stamp, which is
	// identification and not authorization (ADR-039's rule for HTTP resolvers,
	// applied here). A producer that can publish to this stream therefore chooses
	// which tenant a message is held under, and a failure it induces holds THAT
	// tenant until the hold drains. Isolating producers — per-tenant credentials
	// or vhosts — stays the deployment's job, and it matters more here than for a
	// delivery that is merely mis-attributed.
	Hold bool
	// TenantOptional lets this consumer run a delivery that carries no tenant stamp
	// (a control-plane consumer). Default false: fail closed. It never admits a
	// stamp that is present but unusable.
	TenantOptional bool
}

SuperStreamConsumerOptions declares one consumer over every partition of a super stream.

There is deliberately no SAC field: super-stream consumption is ALWAYS a single active consumer group. The client attaches every partition with one shared offset specification, so the SAC promotion callback — which the broker fires once per partition — is the only place a per-partition stored offset can be restored. Without it a restart would replay every partition from Start, which contradicts the stored-offset-wins contract the plain lane documents. A lone member is promoted on every partition, so a single-instance deployment loses nothing. See ADR-059.

type SuperStreamPublisherOptions 

type SuperStreamPublisherOptions struct {
	// SuperStream is the super stream to publish to; it must be declared in the
	// same Declarations.
	SuperStream string
}

SuperStreamPublisherOptions declares one super-stream publisher, which routes each message to a partition by the murmur3 hash of its RoutingKey.

Source Files

View all Source files

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.