rabbitmqqueue

package module
v1.1.0 Latest Latest
Warning

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

 Go to latest Published: Sep 5, 2026 License: MIT Imports: 17 Imported by: 0

README

go-rabbitmq-queues

rabbitmqqueue is the RabbitMQ-native AMQP 0-9-1 queue policy package for Go. It keeps exchanges, routing, classic and quorum queue capabilities, publisher outcomes, manual settlement, bounded recovery, and topology ownership visible. It is intentionally separate from retained RabbitMQ Streams and from the backend-neutral go-queue job API.

Status

The module is stable at v1 and requires Go 1.27.0. It provides independent producer and consumer resources with explicit topology, recovery, health, settlement, and observation policies. See the documentation index for operational detail and current evidence boundaries.

go get github.com/faustbrian/go-rabbitmq-queues@v1

For shared package families, selection guidance, ownership, and lifecycle vocabulary, see the versioned v1.4.0 Go library ecosystem index and its Integration and data movement family.

Five-minute producer and worker

The example assumes an operator has provisioned the durable quorum queue orders; passive verification does not mutate production topology. It opens a consumer, publishes through RabbitMQ's default direct exchange, observes one handler invocation, and shuts both owned resources down. It uses the deprecated Close(ctx) compatibility spelling so it compiles across every published v1 minor; prefer Shutdown(ctx) when the selected release provides it.

package main

import (
	"context"
	"time"

	rabbitmqqueue "github.com/faustbrian/go-rabbitmq-queues"
)

func main() {
	config := rabbitmqqueue.ConnectionConfig{
		Endpoints:   []rabbitmqqueue.Endpoint{{Host: "rabbitmq.internal", Port: 5671}},
		VirtualHost: "/orders",
		Credentials: rabbitmqqueue.CredentialProviderFunc(func(context.Context) (rabbitmqqueue.Credentials, error) {
			return rabbitmqqueue.Credentials{Username: "orders", Password: []byte("resolved secret")}, nil
		}),
		TLS:         rabbitmqqueue.TLSConfig{ServerName: "rabbitmq.internal"},
		DialTimeout: 5 * time.Second,
		Heartbeat:   30 * time.Second,
		Recovery: rabbitmqqueue.RecoveryPolicy{
			MaxAttempts: 8, InitialDelay: 100 * time.Millisecond, MaxDelay: 30 * time.Second,
		},
	}
	_, err := rabbitmqqueue.ApplyTopology(context.Background(), config,
		rabbitmqqueue.TopologyPolicy{Mode: rabbitmqqueue.TopologyPassive},
		rabbitmqqueue.Topology{
			Queues: []rabbitmqqueue.Queue{{
				Name: "orders", Type: rabbitmqqueue.QueueQuorum, Durable: true,
			}},
		},
	)
	if err != nil {
		panic(err)
	}

	producer, err := rabbitmqqueue.OpenProducer(context.Background(), config, rabbitmqqueue.ProducerConfig{
		Limits:         rabbitmqqueue.DefaultLimits(),
		MaxOutstanding: 256,
		PublishTimeout: 5 * time.Second,
	})
	if err != nil {
		panic(err)
	}
	defer producer.Close(context.Background())

	handled := make(chan struct{}, 1)
	consumer, err := rabbitmqqueue.OpenConsumer(context.Background(), config, rabbitmqqueue.ConsumerConfig{
		Limits:         rabbitmqqueue.DefaultLimits(),
		Queue:          rabbitmqqueue.QueueReference{Name: "orders", Type: rabbitmqqueue.QueueQuorum},
		Name:           "orders-worker",
		Prefetch:       32,
		Concurrency:    8,
		HandlerTimeout: 30 * time.Second,
		MaxRequeues:    2,
		Failure:        rabbitmqqueue.Reject(false),
	}, func(ctx context.Context, delivery rabbitmqqueue.Delivery) (rabbitmqqueue.Settlement, error) {
		// Persist the application effect before acknowledging the delivery.
		handled <- struct{}{}
		return rabbitmqqueue.Acknowledge(), nil
	})
	if err != nil {
		panic(err)
	}
	defer consumer.Close(context.Background())

	result, err := producer.Publish(context.Background(), rabbitmqqueue.Publication{
		ExchangeKind: rabbitmqqueue.ExchangeDirect,
		RoutingKey:   "orders",
		Mandatory:    true,
		DeliveryMode: rabbitmqqueue.DeliveryPersistent,
		Message: rabbitmqqueue.Message{
			Body: []byte(`{"order_id":"order-1"}`), MessageID: "order-1",
			ContentType: "application/json",
		},
	})
	if err != nil || result.State != rabbitmqqueue.PublishConfirmed {
		panic("publication was not confirmed")
	}
	select {
	case <-handled:
	case <-time.After(30 * time.Second):
		panic("delivery was not handled")
	}
}

Package map

  • github.com/faustbrian/go-rabbitmq-queues owns native AMQP 0-9-1 topology, publishing, consumption, settlement, recovery, health, and observations.
  • github.com/faustbrian/go-queue/adapters/rabbitmq adapts these native contracts to the backend-neutral go-queue worker API; adopt it only through a published, cleanly resolvable module version.
  • github.com/faustbrian/go-queue/rabbitmq is the deprecated compatibility facade for the historical adapter path.

Guarantees and boundaries

  • Publisher confirmation and consumer acknowledgement are separate effects.
  • Cancellation or connection loss after transmission can be ambiguous.
  • Mandatory returns must be reconciled with confirms before acceptance.
  • Connection loss can redeliver a message while its earlier handler invocation is still completing. Applications must tolerate concurrent duplicates.
  • Manual settlement provides at-least-once processing; applications remain responsible for idempotency.
  • Shutdown(ctx) is the preferred producer and consumer lifecycle method. It is repeatable and safe for concurrent use; each caller is bounded by its own context while the one package-owned cleanup continues. The deprecated Close(ctx) methods delegate to the same lifecycle for source compatibility. A producer caller whose context is cancelled or expires accelerates the shared cleanup and can make active publications ambiguous. A consumer caller context bounds only that caller's wait; the consumer's configured handler timeout bounds its shared drain and cleanup.
  • Package-owned connection attempts call credential providers synchronously with a non-nil, bounded context. A shared provider must be concurrency-safe, return on cancellation, avoid re-entering the producer or consumer being opened, and must not panic. The package zeroes its returned password snapshot after each attempt but cannot zero aliases retained by the provider. Direct provider callers own their snapshots and receive callback errors unchanged.
  • The package does not implement RabbitMQ Streams, application schemas, exactly-once processing, an outbox, or a generic messaging interface.

Read the complete guarantees, capability matrix, performance evidence, the specification decision register, and the compatibility policy before production use. Report vulnerabilities through the private process in SECURITY.md; use SUPPORT.md for reproducible defects and adoption questions, and consult the FAQ for common ownership and delivery questions. Run go test ./... for the package test suite and make ci for the complete repository gate. This module is distributed under the MIT license.

Documentation

Overview

Package rabbitmqqueue provides bounded, RabbitMQ-native policy for AMQP 0-9-1 classic and quorum queues.

The package deliberately keeps queue semantics such as exchanges, routing, publisher outcomes, manual settlement, queue types, and recovery visible. It does not provide a backend-neutral queue abstraction and does not claim exactly-once processing.

Index

Examples

Constants

View Source 
const (
	// MaxEndpoints bounds endpoint rotation and diagnostic state.
	MaxEndpoints = 16
	// MaxReconnectAttempts bounds one continuous recovery episode.
	MaxReconnectAttempts = 32
	// MaxRootCAs bounds custom trust-store parsing and retained certificate state.
	MaxRootCAs = 16
	// MaxTLSMaterialBytes bounds aggregate roots, certificate, and private-key bytes.
	MaxTLSMaterialBytes = 1 << 20
)
View Source 
const (
	MaxConsumerPrefetch    = 4096
	MaxConsumerConcurrency = 256
	MaxConsumerRequeues    = 100
	MaxDeathRecords        = 128
	MaxDeathRoutingKeys    = 32
)
View Source 
const (
	MaxOutstandingConfirms = 4096
	MaxPublishBatchSize    = 1024
)
View Source 
const (
	// MaxTopologyExchanges bounds one topology operation's exchange set.
	MaxTopologyExchanges = 128
	// MaxTopologyQueues bounds one topology operation's queue set.
	MaxTopologyQueues = 128
	// MaxTopologyBindings bounds one development declaration's binding set.
	MaxTopologyBindings = 512
)

Variables

View Source 
var (
	// ErrInvalidEndpoint means a connection endpoint is missing or unsafe.
	ErrInvalidEndpoint = errors.New("rabbitmqqueue: invalid endpoint")
	// ErrCredentialsRequired means no rotating credential provider was supplied.
	ErrCredentialsRequired = errors.New("rabbitmqqueue: credentials are required")
	// ErrInvalidTLS means verified TLS configuration is incomplete or unsafe.
	ErrInvalidTLS = errors.New("rabbitmqqueue: invalid TLS configuration")
	// ErrInvalidBounds means a configured resource or time bound is invalid.
	ErrInvalidBounds = errors.New("rabbitmqqueue: invalid resource bounds")
	// ErrInvalidVirtualHost means the AMQP virtual-host identity is invalid.
	ErrInvalidVirtualHost = errors.New("rabbitmqqueue: invalid virtual host")
	// ErrUnsupportedQueuePolicy means a queue option is not supported by its queue type.
	ErrUnsupportedQueuePolicy = errors.New("rabbitmqqueue: unsupported queue policy")
	// ErrUnsupportedExchangeKind means the exchange kind is not an AMQP built-in supported here.
	ErrUnsupportedExchangeKind = errors.New("rabbitmqqueue: unsupported exchange kind")
	// ErrInvalidTopology means a topology identity or property is invalid.
	ErrInvalidTopology = errors.New("rabbitmqqueue: invalid topology")
	// ErrTopologyMutationDenied means active declaration lacks a development-only permit.
	ErrTopologyMutationDenied = errors.New("rabbitmqqueue: topology mutation denied")
	// ErrPassiveBindingVerificationUnsupported means AMQP cannot inspect a binding without mutating it.
	ErrPassiveBindingVerificationUnsupported = errors.New("rabbitmqqueue: passive binding verification unsupported")
	// ErrTopologyUnavailable means required topology is missing or could not be inspected.
	ErrTopologyUnavailable = errors.New("rabbitmqqueue: topology unavailable")
	// ErrTopologyInequivalent means broker topology exists with incompatible declaration properties.
	ErrTopologyInequivalent = errors.New("rabbitmqqueue: topology is inequivalent")
	// ErrTopologyUnauthorized means broker permissions denied topology inspection or declaration.
	ErrTopologyUnauthorized = errors.New("rabbitmqqueue: topology access denied")
	// ErrMessageIDRequired means a publication has no stable application message identity.
	ErrMessageIDRequired = errors.New("rabbitmqqueue: message ID is required")
	// ErrPayloadTooLarge means a payload exceeds the configured byte limit.
	ErrPayloadTooLarge = errors.New("rabbitmqqueue: payload is too large")
	// ErrHeadersTooLarge means header count or bytes exceed configured limits.
	ErrHeadersTooLarge = errors.New("rabbitmqqueue: headers are too large")
	// ErrDuplicateHeader means a message repeats a header key.
	ErrDuplicateHeader = errors.New("rabbitmqqueue: duplicate header")
	// ErrInvalidHeader means a header key or value is outside the stable policy surface.
	ErrInvalidHeader = errors.New("rabbitmqqueue: invalid header")
	// ErrInvalidPriority means a message priority is outside the AMQP octet range.
	ErrInvalidPriority = errors.New("rabbitmqqueue: invalid priority")
	// ErrInvalidExpiration means a message expiration is negative or cannot be encoded safely.
	ErrInvalidExpiration = errors.New("rabbitmqqueue: invalid expiration")
	// ErrInvalidPublication means publication routing or properties are invalid.
	ErrInvalidPublication = errors.New("rabbitmqqueue: invalid publication")
	// ErrOutstandingConfirmLimit means the bounded in-flight publish window is full.
	ErrOutstandingConfirmLimit = errors.New("rabbitmqqueue: outstanding confirm limit reached")
	// ErrInvalidBatch means a publish batch is empty, oversized, or contains invalid work.
	ErrInvalidBatch = errors.New("rabbitmqqueue: invalid publish batch")
	// ErrInvalidPublishCorrelation means a publish sequence or internal token is invalid or reused.
	ErrInvalidPublishCorrelation = errors.New("rabbitmqqueue: invalid publish correlation")
	// ErrContextRequired means an operation received a nil context.
	ErrContextRequired = errors.New("rabbitmqqueue: context is required")
	// ErrProducerClosed means the producer no longer accepts publications.
	ErrProducerClosed = errors.New("rabbitmqqueue: producer is closed")
	// ErrProducerUnavailable means producer setup or its event channel failed.
	ErrProducerUnavailable = errors.New("rabbitmqqueue: producer is unavailable")
	// ErrPublishReturned means mandatory routing returned the publication.
	ErrPublishReturned = errors.New("rabbitmqqueue: publication was returned")
	// ErrPublishRejected means the broker negatively confirmed the publication.
	ErrPublishRejected = errors.New("rabbitmqqueue: publication was rejected")
	// ErrPublishAmbiguous means transmission began but no definitive broker result was observed.
	ErrPublishAmbiguous = errors.New("rabbitmqqueue: publication outcome is ambiguous")
	// ErrReservedHeader means application metadata collides with package- or broker-owned delivery state.
	ErrReservedHeader = errors.New("rabbitmqqueue: reserved header")
	// ErrInvalidConsumer means consumer identity, bounds, or failure policy is invalid.
	ErrInvalidConsumer = errors.New("rabbitmqqueue: invalid consumer")
	// ErrInvalidDelivery means broker delivery data exceeds the safe public policy surface.
	ErrInvalidDelivery = errors.New("rabbitmqqueue: invalid delivery")
	// ErrInvalidSettlement means a handler requested an undefined settlement operation.
	ErrInvalidSettlement = errors.New("rabbitmqqueue: invalid settlement")
	// ErrSettlementResultUnavailable means a delivery has no observable broker settlement result.
	ErrSettlementResultUnavailable = errors.New("rabbitmqqueue: settlement result is unavailable")
	// ErrConsumerClosed means a stopped consumer cannot change admission state.
	ErrConsumerClosed = errors.New("rabbitmqqueue: consumer is closed")
	// ErrConsumerUnavailable means consumer setup, delivery, or settlement reached a terminal state.
	ErrConsumerUnavailable = errors.New("rabbitmqqueue: consumer is unavailable")
)

Functions

This section is empty.

Types

type Binding 

type Binding struct {
	Exchange   string
	Queue      string
	RoutingKey string
	Arguments  []Header
}

Binding identifies one queue binding without exposing a raw AMQP field table. Arguments are supported only for headers exchanges; other built-in exchange kinds use the explicit routing key.

type ConnectionBlockedState 

type ConnectionBlockedState struct {
	Active bool
}

ConnectionBlockedState reports whether RabbitMQ has temporarily blocked publishing on the owned connection. Broker-provided reason text is omitted.

type ConnectionConfig 

type ConnectionConfig struct {
	Endpoints   []Endpoint
	VirtualHost string
	Credentials CredentialProvider
	TLS         TLSConfig
	DialTimeout time.Duration
	Heartbeat   time.Duration
	Recovery    RecoveryPolicy
}

ConnectionConfig owns connection, authentication, TLS, heartbeat, and recovery policy. It contains no formatted URI so credentials cannot leak through ordinary diagnostics.

func (ConnectionConfig) Validate 

func (config ConnectionConfig) Validate() error

Validate rejects unbounded, secret-bearing, or unverifiable connection policy.

type Consumer 

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

Consumer owns one active manual-acknowledgement generation and a bounded worker pool. Runtime recovery replaces the complete connection/channel/ consumer generation before admitting more broker deliveries.

func OpenConsumer 

func OpenConsumer(
	ctx context.Context,
	connection ConnectionConfig,
	config ConsumerConfig,
	handler DeliveryHandler,
) (*Consumer, error)

OpenConsumer establishes an independent consumer-only AMQP connection, applies bounded per-consumer QoS, and starts manual-settlement workers.

func (*Consumer) Close 

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

Close is retained for compatibility. Deprecated: use Shutdown.

func (*Consumer) DependencyHealth 

func (consumer *Consumer) DependencyHealth() DependencyHealth

DependencyHealth reports consumer connection state independently of liveness.

func (*Consumer) Done 

func (consumer *Consumer) Done() <-chan struct{}

Done closes after broker intake stops and all admitted handlers return.

func (*Consumer) Drain 

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

Drain cancels broker intake and waits for every delivery already received from the broker. It leaves the healthy owned connection open after complete settlement; delegated work or a drain deadline closes the connection.

func (*Consumer) Err 

func (consumer *Consumer) Err() error

Err reports a sanitized unexpected terminal consumer failure.

func (*Consumer) Liveness 

func (consumer *Consumer) Liveness() Liveness

Liveness reports consumer supervision state without probing RabbitMQ.

func (*Consumer) Observations 

func (consumer *Consumer) Observations() <-chan Observation

Observations returns the bounded best-effort consumer event stream. It closes after broker intake and all admitted handlers stop.

func (*Consumer) Pause 

func (consumer *Consumer) Pause() error

Pause stops new handler admission without cancelling the active broker consumer. Already admitted handlers continue through settlement. Up to the configured prefetch may be held unsettled until Resume. Pause is idempotent.

func (*Consumer) Readiness 

func (consumer *Consumer) Readiness() Readiness

Readiness reports whether the consumer currently admits broker deliveries.

func (*Consumer) Resume 

func (consumer *Consumer) Resume() error

Resume permits handler admission after Pause. It is idempotent.

func (*Consumer) Shutdown added in v1.1.0 

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

Shutdown stops handler admission, drains admitted handlers, then closes owned resources. Broker deliveries buffered before admission are left unsettled for redelivery and make an overlapping Drain report ErrConsumerUnavailable. It is safe to call repeatedly or concurrently. Each caller waits only for its own context; cleanup continues after a caller returns early. If cancellation or the internal drain deadline fails, resources are still closed for redelivery.

type ConsumerConfig 

type ConsumerConfig struct {
	Limits         Limits
	Queue          QueueReference
	Name           string
	Priority       *int32
	Exclusive      bool
	Prefetch       int
	Concurrency    int
	HandlerTimeout time.Duration
	MaxRequeues    uint32
	Failure        Settlement
}

ConsumerConfig bounds one independent manual-settlement consumer. Priority distinguishes an omitted RabbitMQ default from an explicit signed value, including zero. Exclusive requests classic-queue exclusivity and cannot be combined with single-active-consumer topology. HandlerTimeout also bounds settlement and supplies the shutdown fallback; handlers must observe their context for graceful draining. MaxRequeues uses RabbitMQ 4.3's quorum acquired count when available and otherwise permits at most one redelivery.

func (ConsumerConfig) Validate 

func (config ConsumerConfig) Validate() error

Validate rejects unbounded consumption and unsafe automatic failure outcomes.

type CredentialProvider 

type CredentialProvider interface {
	Credentials(context.Context) (Credentials, error)
}

CredentialProvider resolves credentials for an individual connection attempt. Implementations may be called concurrently by independent producers and consumers. Package-owned connection attempts supply a non-nil, bounded context. Calls are synchronous, are never made while a package lock is held, and panics are not recovered. Implementations must return when the supplied context is cancelled.

type CredentialProviderFunc 

type CredentialProviderFunc func(context.Context) (Credentials, error)

CredentialProviderFunc adapts a function to CredentialProvider. The function must follow CredentialProvider's concurrency, cancellation, and panic rules.

Example 
package main

import (
	"context"
	"fmt"

	rabbitmqqueue "github.com/faustbrian/go-rabbitmq-queues"
)

func main() {
	provider := rabbitmqqueue.CredentialProviderFunc(
		func(context.Context) (rabbitmqqueue.Credentials, error) {
			return rabbitmqqueue.Credentials{
				Username: "orders",
				Password: []byte("resolved-attempt-secret"),
			}, nil
		},
	)

	credentials, err := provider.Credentials(context.Background())
	fmt.Println(credentials.Username, err)
	clear(credentials.Password)

}

Output:
orders <nil>

func (CredentialProviderFunc) Credentials 

func (provider CredentialProviderFunc) Credentials(ctx context.Context) (Credentials, error)

Credentials resolves a fresh caller-owned credential snapshot. OpenProducer, OpenConsumer, and ApplyTopology zero their returned password snapshot after each connection attempt. Direct callers and providers remain responsible for snapshots and aliases they retain.

type Credentials 

type Credentials struct {
	Username string
	Password []byte
}

Credentials are an owned authentication snapshot. Providers should return a fresh password slice on every call so reconnection can observe rotation.

type DeadLetterStrategy 

type DeadLetterStrategy string

DeadLetterStrategy selects quorum queue dead-letter transfer guarantees. The zero value leaves the broker's at-most-once default implicit. 

const (
	DeadLetterAtMostOnce  DeadLetterStrategy = "at-most-once"
	DeadLetterAtLeastOnce DeadLetterStrategy = "at-least-once"
)

type Death 

type Death struct {
	Count              uint64
	Reason             string
	Queue              string
	Exchange           string
	RoutingKeys        []string
	Time               time.Time
	OriginalExpiration *time.Duration
}

Death preserves one bounded RabbitMQ x-death record without exposing a raw field table.

type DelayedRetryType 

type DelayedRetryType string

DelayedRetryType selects which RabbitMQ 4.3 quorum redeliveries receive broker-managed linear backoff. 

const (
	DelayedRetryDisabled DelayedRetryType = "disabled"
	DelayedRetryAll      DelayedRetryType = "all"
	DelayedRetryFailed   DelayedRetryType = "failed"
	DelayedRetryReturned DelayedRetryType = "returned"
)

type Delivery 

type Delivery struct {
	Body            []byte
	Headers         []Header
	MessageID       string
	CorrelationID   string
	ContentType     string
	ContentEncoding string
	ReplyTo         string
	Type            string
	UserID          string
	AppID           string
	Timestamp       time.Time
	// Expiration distinguishes an omitted TTL from RabbitMQ's explicit
	// zero-duration immediate-expiration value.
	Expiration    *time.Duration
	Priority      uint8
	DeliveryMode  DeliveryMode
	Consumer      string
	Exchange      string
	RoutingKey    string
	Redelivered   bool
	AcquiredCount *uint64
	DeliveryCount *uint64
	Deaths        []Death
	// contains filtered or unexported fields
}

Delivery is an owned, bounded AMQP delivery snapshot. Delivery tags and the underlying client delivery never cross the public API boundary.

func (Delivery) AwaitSettlement 

func (delivery Delivery) AwaitSettlement(ctx context.Context) error

AwaitSettlement waits for the broker result of the settlement selected by the delivery handler. The handler must first return its Settlement; waiting synchronously inside that handler cannot complete. A separate goroutine may wait on a copied Delivery while the handler returns. Copies share the same bounded result. The method returns ErrSettlementResultUnavailable for deliveries not created by a Consumer or whose handler delegated settlement.

type DeliveryHandler 

type DeliveryHandler func(context.Context, Delivery) (Settlement, error)

DeliveryHandler processes one owned delivery and returns its explicit manual settlement.

type DeliveryMode 

type DeliveryMode uint8

DeliveryMode selects broker persistence intent. 

const (
	DeliveryTransient DeliveryMode = iota + 1
	DeliveryPersistent
)

type DependencyHealth 

type DependencyHealth string

DependencyHealth reports the owned RabbitMQ dependency state separately from process liveness. 

const (
	DependencyAvailable   DependencyHealth = "available"
	DependencyBlocked     DependencyHealth = "blocked"
	DependencyRecovering  DependencyHealth = "recovering"
	DependencyUnavailable DependencyHealth = "unavailable"
	DependencyUnknown     DependencyHealth = "unknown"
)

type DevelopmentTopologyPermit 

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

DevelopmentTopologyPermit is an explicit capability for test and local topology declaration. Its zero value never permits mutation.

func PermitDevelopmentTopology 

func PermitDevelopmentTopology() DevelopmentTopologyPermit

PermitDevelopmentTopology explicitly opts a development or test process into topology mutation. Production applications must not call this function.

type Endpoint 

type Endpoint struct {
	Host string
	Port uint16
}

Endpoint identifies one AMQP listener without embedding credentials.

type Exchange 

type Exchange struct {
	Name       string
	Kind       ExchangeKind
	Durable    bool
	AutoDelete bool
	Internal   bool
}

Exchange is a stable exchange identity and equivalence policy.

func (Exchange) Validate 

func (exchange Exchange) Validate() error

Validate checks exchange identity and supported kind.

type ExchangeKind 

type ExchangeKind string

ExchangeKind selects a RabbitMQ built-in AMQP exchange algorithm. 

const (
	ExchangeDirect  ExchangeKind = "direct"
	ExchangeTopic   ExchangeKind = "topic"
	ExchangeFanout  ExchangeKind = "fanout"
	ExchangeHeaders ExchangeKind = "headers"
)
type Header struct {
	Key    string
	Kind   HeaderKind
	String string
	Bool   bool
	Int64  int64
	Bytes  []byte
}

Header is one ordered AMQP application header. Nested tables and arrays are intentionally excluded to keep allocation and interoperability bounded.

func BoolHeader 

func BoolHeader(key string, value bool) Header

BoolHeader creates a boolean application header.

func BytesHeader 

func BytesHeader(key string, value []byte) Header

BytesHeader creates a byte-string application header with an owned value copy.

func Int64Header 

func Int64Header(key string, value int64) Header

Int64Header creates a signed 64-bit integer application header.

func StringHeader 

func StringHeader(key, value string) Header

StringHeader creates a string application header.

type HeaderKind 

type HeaderKind uint8

HeaderKind identifies a bounded, language-neutral AMQP field-table value. 

const (
	HeaderString HeaderKind = iota + 1
	HeaderBool
	HeaderInt64
	HeaderBytes
)

type Limits 

type Limits struct {
	MaxPayloadBytes    int
	MaxHeaderEntries   int
	MaxHeaderBytes     int
	MaxNameBytes       int
	MaxRoutingKeyBytes int
}

Limits bounds untrusted message and topology-controlled allocation. Values may be lowered from DefaultLimits but cannot raise the package safety caps.

func DefaultLimits 

func DefaultLimits() Limits

DefaultLimits returns conservative RabbitMQ 4.x policy bounds.

type Liveness 

type Liveness string

Liveness is process-supervision state for one package resource. Temporary dependency outages remain live while bounded recovery is active. 

const (
	LivenessLive    Liveness = "live"
	LivenessFailed  Liveness = "failed"
	LivenessStopped Liveness = "stopped"
)

type Message 

type Message struct {
	Body            []byte
	MessageID       string
	CorrelationID   string
	ReplyTo         string
	ContentType     string
	ContentEncoding string
	Type            string
	AppID           string
	// Timestamp is either zero or a non-negative whole-second AMQP timestamp.
	Timestamp time.Time
	// Expiration distinguishes an omitted TTL from an explicit zero-duration
	// TTL, which RabbitMQ interprets as immediate expiration when the message
	// cannot be delivered directly.
	Expiration *time.Duration
	Priority   *uint16
	Headers    []Header
}

Message contains AMQP message properties and opaque payload bytes.

type Observation 

type Observation struct {
	Resource ObservationResource
	Kind     ObservationKind
	Outcome  ObservationOutcome
	Duration time.Duration
	Dropped  uint64
}

Observation is a payload-free, identifier-free operational event. Duration is populated only for confirmation latency. Dropped reports observations discarded since the previous delivered event because the bounded stream was full. Stream closure emits ObservationStreamClosed and reserves a buffered slot when necessary so undisclosed tail loss remains visible.

type ObservationKind 

type ObservationKind string

ObservationKind is a fixed low-cardinality operational event category. 

const (
	ObservationConnectionState      ObservationKind = "connection_state"
	ObservationConnectionBlocked    ObservationKind = "connection_blocked"
	ObservationReconnect            ObservationKind = "reconnect"
	ObservationPublish              ObservationKind = "publish"
	ObservationReturn               ObservationKind = "return"
	ObservationConfirm              ObservationKind = "confirm"
	ObservationConfirmationLatency  ObservationKind = "confirmation_latency"
	ObservationAmbiguous            ObservationKind = "ambiguous"
	ObservationDelivery             ObservationKind = "delivery"
	ObservationRedelivery           ObservationKind = "redelivery"
	ObservationConsumerCancellation ObservationKind = "consumer_cancellation"
	ObservationAcknowledgement      ObservationKind = "acknowledgement"
	ObservationSettlement           ObservationKind = "settlement"
	ObservationHandlerFailure       ObservationKind = "handler_failure"
	ObservationDeadLetter           ObservationKind = "dead_letter"
	ObservationBacklogPressure      ObservationKind = "backlog_pressure"
	ObservationShutdown             ObservationKind = "shutdown"
	ObservationStreamClosed         ObservationKind = "stream_closed"
)

type ObservationOutcome 

type ObservationOutcome string

ObservationOutcome is a fixed low-cardinality event result or transition. 

const (
	ObservationConnected            ObservationOutcome = "connected"
	ObservationRecovering           ObservationOutcome = "recovering"
	ObservationRecovered            ObservationOutcome = "recovered"
	ObservationUnavailable          ObservationOutcome = "unavailable"
	ObservationBlocked              ObservationOutcome = "blocked"
	ObservationUnblocked            ObservationOutcome = "unblocked"
	ObservationAttempted            ObservationOutcome = "attempted"
	ObservationConfirmed            ObservationOutcome = "confirmed"
	ObservationRejected             ObservationOutcome = "rejected"
	ObservationReturned             ObservationOutcome = "returned"
	ObservationNotSent              ObservationOutcome = "not_sent"
	ObservationAmbiguousOutcome     ObservationOutcome = "ambiguous"
	ObservationDelivered            ObservationOutcome = "delivered"
	ObservationRedelivered          ObservationOutcome = "redelivered"
	ObservationCancelled            ObservationOutcome = "cancelled"
	ObservationAcknowledged         ObservationOutcome = "acknowledged"
	ObservationNegativeAcknowledged ObservationOutcome = "negative_acknowledged"
	ObservationHandlerFailed        ObservationOutcome = "failed"
	ObservationDeadLettered         ObservationOutcome = "dead_lettered"
	ObservationBacklogFull          ObservationOutcome = "full"
	ObservationShutdownStarted      ObservationOutcome = "started"
	ObservationShutdownCompleted    ObservationOutcome = "completed"
	ObservationClosed               ObservationOutcome = "closed"
)

type ObservationResource 

type ObservationResource string

ObservationResource identifies the bounded package resource that emitted an observation without exposing a connection, route, message, or consumer ID. 

const (
	ObservationProducer ObservationResource = "producer"
	ObservationConsumer ObservationResource = "consumer"
)

type Producer 

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

Producer owns one active confirm-enabled AMQP generation and never creates consumers. Publish is safe for concurrent use. Shutdown prevents new work, drains bounded active calls, cancels recovery, and then releases the channel and connection resource.

func OpenProducer 

func OpenProducer(
	ctx context.Context,
	connection ConnectionConfig,
	config ProducerConfig,
) (*Producer, error)

OpenProducer establishes an independent producer-only AMQP connection and confirm-enabled channel. Startup and bounded runtime recovery attempts rotate endpoints and credentials. Exhausted runtime recovery is terminal.

func (*Producer) BlockedNotifications 

func (producer *Producer) BlockedNotifications() <-chan ConnectionBlockedState

BlockedNotifications emits coalesced sanitized state transitions. The channel closes when the producer lifecycle ends.

func (*Producer) Close 

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

Close is retained for compatibility. Deprecated: use Shutdown.

func (*Producer) DependencyHealth 

func (producer *Producer) DependencyHealth() DependencyHealth

DependencyHealth reports producer connection state independently of liveness.

func (*Producer) IsBlocked 

func (producer *Producer) IsBlocked() bool

IsBlocked reports the latest sanitized RabbitMQ connection-blocked state.

func (*Producer) Liveness 

func (producer *Producer) Liveness() Liveness

Liveness reports producer supervision state without probing RabbitMQ.

func (*Producer) Observations 

func (producer *Producer) Observations() <-chan Observation

Observations returns the bounded best-effort producer event stream. The stream closes after Shutdown completes; terminal recovery alone does not release caller-owned observation consumption.

func (*Producer) Publish 

func (producer *Producer) Publish(ctx context.Context, publication Publication) (PublishResult, error)

Publish sends one publication and waits for its exact mandatory-return and confirmation outcome. A timeout after transmission is always ambiguous.

func (*Producer) PublishAsync 

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

PublishAsync admits one bounded publication and returns a channel that emits exactly one terminal outcome. Its caller context and PublishTimeout bound the total interval from admission through transmission and confirmation; expiry before transmission reports PublishNotSent. Admission failures do not create goroutines.

func (*Producer) PublishBatch 

func (producer *Producer) PublishBatch(ctx context.Context, publications []Publication) ([]PublishOutcome, error)

PublishBatch validates the complete bounded batch before publishing each item. Outcomes preserve input order; the batch is not an atomic broker unit.

func (*Producer) Readiness 

func (producer *Producer) Readiness() Readiness

Readiness reports whether the producer currently admits publications.

func (*Producer) Shutdown added in v1.1.0 

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

Shutdown prevents new publications, waits for active bounded calls, and closes owned AMQP resources. It is safe to call repeatedly or concurrently. Each caller waits only for its own context; cleanup continues after a caller returns early. Cancellation or deadline expiry from any caller accelerates the shared cleanup by forcing the owned connection closed, making any still-active publication ambiguous.

type ProducerConfig 

type ProducerConfig struct {
	Limits         Limits
	MaxOutstanding int
	PublishTimeout time.Duration
}

ProducerConfig bounds synchronous producer work and confirmation state.

func (ProducerConfig) Validate 

func (config ProducerConfig) Validate() error

Validate rejects unbounded producer policy.

type Publication 

type Publication struct {
	// Exchange is a named exchange identity, or the empty default-exchange
	// identity when ExchangeKind is explicitly ExchangeDirect.
	Exchange string
	// ExchangeKind records the expected routing semantic for local validation.
	// Omit it only for non-empty direct/topic-compatible routing keys. An
	// explicit direct or topic kind may use RabbitMQ's native empty key; fanout
	// and headers publications must name their kind and use an empty key.
	ExchangeKind ExchangeKind
	RoutingKey   string
	Mandatory    bool
	DeliveryMode DeliveryMode
	Message      Message
}

Publication binds one message to explicit AMQP routing policy.

func (Publication) Validate 

func (publication Publication) Validate(limits Limits) error

Validate bounds and validates publication routing, properties, and headers.

type PublishOutcome 

type PublishOutcome struct {
	Result PublishResult
	Err    error
}

PublishOutcome pairs one terminal result with its sanitized operation error. Batch outcomes preserve input order; asynchronous outcomes are delivered once.

type PublishResult 

type PublishResult struct {
	State  PublishState
	Return *Return
}

PublishResult is the terminal observed state for exactly one publish attempt.

func (PublishResult) Valid 

func (result PublishResult) Valid() bool

Valid reports whether the state and mandatory-return detail form a canonical outcome.

type PublishState 

type PublishState string

PublishState distinguishes broker outcomes without collapsing post-send cancellation or connection loss into a definitive result. 

const (
	PublishNotSent   PublishState = "not_sent"
	PublishRejected  PublishState = "rejected"
	PublishReturned  PublishState = "returned"
	PublishConfirmed PublishState = "confirmed"
	PublishAmbiguous PublishState = "ambiguous"
)

func (PublishState) Valid 

func (state PublishState) Valid() bool

Valid reports whether state is a defined publication outcome.

type Queue 

type Queue struct {
	Name                        string
	Type                        QueueType
	Durable                     bool
	AutoDelete                  bool
	Exclusive                   bool
	SingleActiveConsumer        bool
	DeliveryLimit               *QueueDeliveryLimit
	MaxPriority                 uint8
	MessageTTL                  *time.Duration
	Expires                     *time.Duration
	ConsumerTimeout             *time.Duration
	DisconnectedConsumerTimeout *time.Duration
	DelayedRetry                *QueueDelayedRetry
	MaxLength                   *uint64
	MaxLengthBytes              *uint64
	Overflow                    QueueOverflow
	DeadLetter                  *QueueDeadLetter
}

Queue describes declaration-equivalent queue policy. A zero Name requests a server-generated name and is valid only for an exclusive classic queue. MessageTTL and the length pointers distinguish an explicit zero argument from omission. Expires, when present, must be a positive millisecond value. A nil DeliveryLimit preserves the broker policy or default; a pointer emits an explicit bounded value, including zero. ConsumerTimeout is RabbitMQ 4.3's quorum-only delivery-acknowledgement timeout and accepts non-negative values with millisecond precision. DisconnectedConsumerTimeout is RabbitMQ 4.3's quorum-only wait before held deliveries are returned after a consumer node becomes unreachable. DelayedRetry is RabbitMQ 4.3's quorum-only linear-backoff policy.

func (Queue) Validate 

func (queue Queue) Validate() error

Validate rejects policies that RabbitMQ cannot apply to the selected queue type.

type QueueDeadLetter 

type QueueDeadLetter struct {
	Exchange   string
	RoutingKey *string
	Strategy   DeadLetterStrategy
}

QueueDeadLetter describes declaration-time dead-letter arguments. An empty Exchange explicitly selects the AMQP default exchange. A nil RoutingKey omits the argument and preserves original routing keys; a pointer to an empty string emits an explicit empty routing key. RabbitMQ policies are preferred for production configuration because they remain mutable.

type QueueDelayedRetry 

type QueueDelayedRetry struct {
	Type    DelayedRetryType
	Minimum time.Duration
	Maximum *time.Duration
}

QueueDelayedRetry describes RabbitMQ 4.3 quorum delayed-retry arguments. Enabled retry requires a positive millisecond Minimum. A nil Maximum uses a fixed delay equal to Minimum; otherwise Maximum must not precede Minimum.

type QueueDeliveryLimit 

type QueueDeliveryLimit uint32

QueueDeliveryLimit bounds RabbitMQ quorum failed redeliveries. The unsigned policy intentionally cannot represent RabbitMQ's unsafe unlimited value -1.

type QueueOverflow 

type QueueOverflow string

QueueOverflow selects RabbitMQ's queue-length overflow behavior. 

const (
	QueueOverflowDropHead                QueueOverflow = "drop-head"
	QueueOverflowRejectPublish           QueueOverflow = "reject-publish"
	QueueOverflowRejectPublishDeadLetter QueueOverflow = "reject-publish-dlx"
)

type QueueReference 

type QueueReference struct {
	Name                 string
	Type                 QueueType
	SingleActiveConsumer bool
	Transient            *TransientQueue
}

QueueReference identifies either an existing operator-owned queue or an explicitly client-owned transient queue. SingleActiveConsumer records declaration intent for local policy validation; callers use passive topology verification when they need broker evidence for a named queue.

func (QueueReference) Validate 

func (reference QueueReference) Validate() error

Validate rejects missing queue identities and unsupported queue types.

type QueueType 

type QueueType string

QueueType distinguishes queue implementations whose policies are not interchangeable. 

const (
	QueueClassic QueueType = "classic"
	QueueQuorum  QueueType = "quorum"
)

type Readiness 

type Readiness string

Readiness reports whether one resource can currently accept useful work. 

const (
	ReadinessReady    Readiness = "ready"
	ReadinessNotReady Readiness = "not_ready"
)

type RecoveryPolicy 

type RecoveryPolicy struct {
	MaxAttempts  int
	InitialDelay time.Duration
	MaxDelay     time.Duration
}

RecoveryPolicy bounds reconnection attempts and exponential backoff.

type Return 

type Return struct {
	Code       uint16
	Reason     string
	Exchange   string
	RoutingKey string
}

Return describes a mandatory unroutable outcome without carrying payloads or headers. Exchange and RoutingKey come from the exact registered publication, not untrusted broker return metadata.

type Settlement 

type Settlement struct {
	Method  SettlementMethod
	Requeue bool
}

Settlement is a handler's explicit request for one delivery. Delegate leaves the delivery unsettled until the consumer drains, closes, or loses its connection.

func Acknowledge 

func Acknowledge() Settlement

Acknowledge requests a single-delivery ACK after handler success.

func Delegate 

func Delegate() Settlement

Delegate explicitly leaves settlement to the consumer connection lifecycle.

func NegativeAcknowledge 

func NegativeAcknowledge(requeue bool) Settlement

NegativeAcknowledge requests a single-delivery NACK.

func Reject 

func Reject(requeue bool) Settlement

Reject requests a single-delivery reject.

func (Settlement) Validate 

func (settlement Settlement) Validate() error

Validate rejects unknown methods and impossible requeue flags.

type SettlementMethod 

type SettlementMethod string

SettlementMethod identifies one AMQP manual-settlement operation. 

const (
	SettlementAcknowledge         SettlementMethod = "ack"
	SettlementNegativeAcknowledge SettlementMethod = "nack"
	SettlementReject              SettlementMethod = "reject"
	SettlementDelegate            SettlementMethod = "delegate"
)

type TLSConfig 

type TLSConfig struct {
	ServerName        string
	RootCAs           [][]byte
	ClientCertificate []byte
	ClientPrivateKey  []byte
}

TLSConfig owns verified TLS identity and optional custom trust material. Certificate and key bytes are secrets and must never be observed or logged.

type Topology 

type Topology struct {
	Exchanges []Exchange
	Queues    []Queue
	Bindings  []Binding
}

Topology is one bounded exchange, queue, and binding graph. Passive AMQP verification can compare exchange and queue declarations, but AMQP 0-9-1 has no passive binding method. Bindings therefore require development declaration or separate infrastructure/operator verification.

func (Topology) Validate 

func (topology Topology) Validate(policy TopologyPolicy) error

Validate checks graph bounds, identities, references, exchange-specific binding rules, lifecycle-safe queues, and the passive-binding protocol limitation. Every exclusive queue belongs to its declaring connection; ApplyTopology cannot retain one because it closes that connection on return.

type TopologyMode 

type TopologyMode string

TopologyMode selects passive equivalence verification or active declaration. 

const (
	TopologyPassive TopologyMode = "passive"
	TopologyDeclare TopologyMode = "declare"
)

type TopologyPolicy 

type TopologyPolicy struct {
	Mode        TopologyMode
	Development DevelopmentTopologyPermit
}

TopologyPolicy keeps production verification distinct from development declaration.

func (TopologyPolicy) Validate 

func (policy TopologyPolicy) Validate() error

Validate prevents declaration without an explicit development-only capability.

type TopologyResult 

type TopologyResult struct {
	QueueNames []string
}

TopologyResult returns verified or declared queue names in Topology.Queues order.

func ApplyTopology 

func ApplyTopology(
	ctx context.Context,
	connection ConnectionConfig,
	policy TopologyPolicy,
	topology Topology,
) (TopologyResult, error)

ApplyTopology passively verifies operator-owned exchange and queue equivalence, or performs explicitly permitted development-only declarations. AMQP cannot passively inspect bindings; Topology.Validate rejects passive binding requests rather than mutating production topology. Connection-scoped server-named queues are declared only by a client-owned transient consumer.

type TransientQueue 

type TransientQueue struct {
	Exchange   Exchange
	RoutingKey string
	Arguments  []Header
}

TransientQueue describes an explicitly client-owned, connection-scoped, server-named classic queue bound to an existing exchange. The consumer declares and consumes it on the same connection so RabbitMQ can retain the exclusive queue for exactly that generation.

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.