rabbitmq

package module
v1.0.0 Latest Latest
Warning

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

 Go to latest Published: Jul 18, 2026 License: MIT Imports: 12 Imported by: 0

README

go-rabbitmq 🐇

Pure-Go RabbitMQ with best-practice auto-reconnecting connections, publishers, and consumers — so you stop re-solving connection resilience in every service. Optional OpenTelemetry tracing + metrics in one line.

📦 Install

go get github.com/Bugs5382/go-rabbitmq

One module, two import paths — the core (github.com/Bugs5382/go-rabbitmq, depends only on amqp091-go) and the optional OTel adapter (github.com/Bugs5382/go-rabbitmq/otel).

🚀 Usage

Connect owns the connection, watches it, and transparently re-dials with bounded exponential backoff — the one thing naive AMQP code always gets wrong is the default here.

conn, err := rabbitmq.Connect(ctx, "amqp://guest:guest@localhost:5672/")
if err != nil {
	log.Fatal(err)
}
defer conn.Close()

📤 Publish

A publisher re-opens its channel and re-declares its exchange after a drop, and retries within the backoff budget. Publish returns an error only once that budget is spent — so an outbox worker can keep the message and try later.

pub := conn.NewPublisher("events")
if err := pub.PublishJSON(ctx, "order.created", order); err != nil {
	// retryable: keep the row, try again later
}

📥 Consume

A consumer re-declares its topology and resumes after any reconnect. Your handler's returned error drives the ack (requeue configurable), so it survives broker restarts.

conn.Consume(ctx, rabbitmq.ConsumerConfig{
	Queue:    rabbitmq.QueueConfig{Name: "orders", Type: rabbitmq.QueueQuorum},
	Exchange: "events",
	Bindings: []string{"order.*"},
}, func(ctx context.Context, d rabbitmq.Delivery) error {
	return handle(d.Body)
})

Ephemeral (server-named / exclusive / auto-delete) queues are forced to classic — quorum queues can't be any of those, and the guard saves you the PRECONDITION_FAILED.

📊 OpenTelemetry

The optional otel subpackage propagates W3C trace context through message headers (producer → consumer spans across the broker) and records publish/consume metrics. The core stays telemetry-free; wire it with one call:

conn, _ := rabbitmq.Connect(ctx, url, rmqotel.Instrument()...)

🛠 Develop

task ci        # build + vet + lint + test
task license   # inject MIT headers (golic)

⚖️ License

MIT © 2026 Shane

Documentation

Overview

Package rabbitmq is a small, dependency-light RabbitMQ client built for resilience. It wraps github.com/rabbitmq/amqp091-go and makes the thing every naive AMQP program gets wrong -- surviving a connection or channel drop -- the default behaviour rather than an afterthought.

Why

A hand-rolled publisher or consumer that opens one connection and one channel works perfectly until the broker restarts, a network blip closes the socket, or the channel is killed by a protocol error. After that the naive program is stuck forever: its channel is dead and nothing re-opens it. This package owns the connection, watches it for closes, and transparently re-dials with bounded exponential backoff. Publishers re-open and re-declare on demand; consumers re-declare their topology and resume consuming after a reconnect.

Pieces

  • Conn is the connection manager. Connect returns one; it re-dials in the background and exposes health via IsClosed, Healthy and Reconnects.
  • Publisher is created from a Conn and an exchange. Publish ensures a live channel and retries within bounded backoff before returning an error, so an outbox worker can simply try again later.
  • Consume runs a consumer that re-declares its queue and bindings and resumes after any reconnect, with manual acknowledgement driven by the handler's returned error.
  • The topology helpers (DeclareExchange, DeclareQueue, BindQueue, DeclareTopology) declare exchanges, queues and bindings idempotently and guard the quorum-vs-classic queue rules.

Pluggable cross-cutting concerns

Logging goes through the minimal Logger interface (default no-op) so callers can plug slog, zerolog or anything else without this package depending on them. Observability goes through the Observer interface (default no-op) so an OpenTelemetry-backed implementation can be supplied without adding an OpenTelemetry dependency here.

Minimal example 

ctx := context.Background()
conn, err := rabbitmq.Connect(ctx, "amqp://guest:guest@localhost:5672/")
if err != nil {
	return err
}
defer conn.Close()

pub := conn.NewPublisher("events")
if err := pub.PublishJSON(ctx, "orders.created", order); err != nil {
	return err // retry later; the message was not accepted
}

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (
	// ErrClosed is returned by operations on a Conn after Close has been called.
	ErrClosed = errors.New("rabbitmq: connection manager is closed")

	// ErrNotReady is returned when a live connection could not be obtained before
	// the supplied context expired.
	ErrNotReady = errors.New("rabbitmq: no live connection available")

	// ErrPublishFailed wraps the last publish error after all bounded retries were
	// exhausted. Callers (for example an outbox worker) should treat it as
	// retryable and try again later.
	ErrPublishFailed = errors.New("rabbitmq: publish failed after retries")

	// ErrInvalidQueue is returned when a QueueConfig violates a broker rule, most
	// commonly a quorum queue that is also exclusive, auto-delete or server-named.
	ErrInvalidQueue = errors.New("rabbitmq: invalid queue configuration")
)

Functions

This section is empty.

Types

type Backoff 

type Backoff struct {
	// Initial is the delay before the first retry. Default 500ms.
	Initial time.Duration
	// Max caps the delay of any single retry. Default 30s.
	Max time.Duration
	// Factor is the exponential growth multiplier per attempt. Default 2.0.
	Factor float64
	// Jitter is the fraction (0..1) of a delay applied as random +/- noise to
	// avoid a thundering herd of reconnects. Default 0.2. A value <=0 disables
	// jitter (useful for deterministic tests).
	Jitter float64
	// MaxRetries bounds the number of retries. 0 means retry indefinitely (the
	// right choice for a long-lived connection). A positive value bounds publish
	// retries and initial-connect attempts.
	MaxRetries int
	// contains filtered or unexported fields
}

Backoff configures the bounded exponential backoff used when re-dialling a dropped connection and when retrying a failed publish. The zero value is not usable on its own; DefaultBackoff supplies sensible defaults and normalize fills any unset field.

func DefaultBackoff 

func DefaultBackoff() Backoff

DefaultBackoff returns the backoff used when none is supplied: 500ms initial, 30s cap, factor 2.0, 20% jitter, retry forever.

type BindingConfig 

type BindingConfig struct {
	Queue      string
	Exchange   string
	RoutingKey string
	NoWait     bool
	Args       amqp.Table
}

BindingConfig describes a queue-to-exchange binding.

type Conn 

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

Conn is a self-healing RabbitMQ connection manager. It owns a single underlying AMQP connection, watches it for closes, and re-dials in the background with bounded exponential backoff. Publishers and consumers created from it survive reconnects transparently.

A Conn is safe for concurrent use by multiple goroutines.

func Connect 

func Connect(ctx context.Context, url string, opts ...Option) (*Conn, error)

Connect establishes the initial connection and starts the background monitor. It retries the first dial with the configured backoff, honouring ctx: if ctx is cancelled before a connection is established (or MaxRetries is exhausted) it returns the last error. The supplied ctx also governs the lifetime of the background reconnect loop; cancel it (or call Close) to shut the manager down.

Example

ExampleConnect shows the minimal path: connect (with auto-reconnect on by default), publish a JSON event, and clean up. 

package main

import (
	"context"
	"log"

	rabbitmq "github.com/Bugs5382/go-rabbitmq"
)

func main() {
	ctx := context.Background()
	conn, err := rabbitmq.Connect(ctx, "amqp://guest:guest@localhost:5672/")
	if err != nil {
		log.Fatal(err)
	}
	defer func() { _ = conn.Close() }()

	pub := conn.NewPublisher("events",
		rabbitmq.WithExchangeDeclare(rabbitmq.ExchangeConfig{Name: "events", Kind: "topic"}),
	)
	order := map[string]any{"id": 42, "total": 19.99}
	if err := pub.PublishJSON(ctx, "orders.created", order); err != nil {
		// The broker did not accept the message after bounded retries; keep it and
		// try again later (for example from an outbox).
		log.Printf("publish failed, will retry: %v", err)
	}
}

func (*Conn) BindQueue 

func (c *Conn) BindQueue(ctx context.Context, cfg BindingConfig) error

BindQueue binds a queue to an exchange with a routing key.

func (*Conn) Close 

func (c *Conn) Close() error

Close shuts down the background monitor and closes the underlying connection. It is idempotent and safe to call concurrently.

func (*Conn) Consume 

func (c *Conn) Consume(ctx context.Context, cfg ConsumerConfig, handler Handler) error

Consume runs a resilient consumer until ctx is cancelled. It (re-)declares the configured topology, sets QoS, and consumes, dispatching each delivery to handler. On any channel or connection drop it waits for the Conn to recover, then re-declares and resumes. It returns ctx.Err() when ctx is cancelled, or ErrClosed if the Conn is closed.

Consume blocks; run it in its own goroutine. Deliveries are dispatched sequentially on the calling goroutine, so a handler that must run concurrently should fan out internally (respecting Prefetch for backpressure).

Example

ExampleConn_Consume shows a resilient consumer that re-declares its topology and resumes after any reconnect. The handler's returned error drives ack/nack. 

package main

import (
	"context"
	"errors"
	"log"
	"time"

	rabbitmq "github.com/Bugs5382/go-rabbitmq"
)

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	conn, err := rabbitmq.Connect(ctx, "amqp://guest:guest@localhost:5672/")
	if err != nil {
		log.Fatal(err)
	}
	defer func() { _ = conn.Close() }()

	cfg := rabbitmq.ConsumerConfig{
		Exchange: rabbitmq.ExchangeConfig{Name: "events", Kind: "topic"},
		Queue:    rabbitmq.QueueConfig{Name: "orders", Type: rabbitmq.QueueQuorum},
		Bindings: []rabbitmq.BindingConfig{{Exchange: "events", RoutingKey: "orders.*"}},
		Prefetch: 20,
	}

	// Consume blocks until ctx is cancelled; run it in its own goroutine.
	go func() {
		err := conn.Consume(ctx, cfg, func(_ context.Context, d rabbitmq.Delivery) error {
			log.Printf("received %s: %s", d.RoutingKey, d.Body)
			return nil // returning nil acks; returning an error nacks
		})
		if err != nil && !errors.Is(err, context.Canceled) {
			log.Printf("consumer stopped: %v", err)
		}
	}()

	time.Sleep(time.Second)
}

func (*Conn) DeclareExchange 

func (c *Conn) DeclareExchange(ctx context.Context, cfg ExchangeConfig) error

DeclareExchange idempotently declares an exchange. Repeated calls with matching arguments are a no-op on the broker.

func (*Conn) DeclareQueue 

func (c *Conn) DeclareQueue(ctx context.Context, cfg QueueConfig) (amqp.Queue, error)

DeclareQueue idempotently declares a queue and returns the broker's view of it. The returned amqp.Queue.Name is authoritative for server-named queues.

func (*Conn) DeclareTopology 

func (c *Conn) DeclareTopology(ctx context.Context, t Topology) error

DeclareTopology declares an entire Topology idempotently on one channel. It is the convenient way to establish the common "topic exchange + durable queue + binding" shape at start-up.

Example

ExampleConn_DeclareTopology declares an exchange, a durable quorum queue and a binding in one call, ready for publishers and consumers to use. 

package main

import (
	"context"
	"log"

	rabbitmq "github.com/Bugs5382/go-rabbitmq"
)

func main() {
	ctx := context.Background()
	conn, err := rabbitmq.Connect(ctx, "amqp://guest:guest@localhost:5672/")
	if err != nil {
		log.Fatal(err)
	}
	defer func() { _ = conn.Close() }()

	err = conn.DeclareTopology(ctx, rabbitmq.Topology{
		Exchanges: []rabbitmq.ExchangeConfig{{Name: "events", Kind: "topic"}},
		Queues:    []rabbitmq.QueueConfig{{Name: "orders", Type: rabbitmq.QueueQuorum}},
		Bindings:  []rabbitmq.BindingConfig{{Queue: "orders", Exchange: "events", RoutingKey: "orders.*"}},
	})
	if err != nil {
		log.Fatal(err)
	}
}

func (*Conn) Healthy 

func (c *Conn) Healthy() bool

Healthy reports whether a live, open connection is currently available.

func (*Conn) NewPublisher 

func (c *Conn) NewPublisher(exchange string, opts ...PublisherOption) *Publisher

NewPublisher creates a Publisher for the given exchange. An empty exchange name targets the default exchange, where the routing key is the destination queue name.

func (*Conn) Reconnects 

func (c *Conn) Reconnects() uint64

Reconnects returns the number of successful reconnections since Connect.

type ConsumeInterceptor 

type ConsumeInterceptor func(ctx context.Context, d Delivery, next Handler) error

ConsumeInterceptor wraps handling of a single delivery. It may extract trace context from d.Headers, derive a new ctx (for example one carrying a consumer span), and pass it to next. Interceptors run outermost-first in the order registered; the innermost next is the user's Handler.

It is the seam the OTel adapter uses to continue a distributed trace across the broker and to time and count message handling.

type ConsumerConfig 

type ConsumerConfig struct {
	// Queue is the queue to consume from. It is declared before consuming.
	Queue QueueConfig
	// Exchange, if Name is non-empty, is declared before the queue.
	Exchange ExchangeConfig
	// Bindings are declared after the queue. A binding with an empty Queue uses
	// the resolved queue name (useful for server-named queues).
	Bindings []BindingConfig
	// Prefetch is the QoS prefetch count (unacked messages in flight). Default 10.
	Prefetch int
	// ConsumerTag is the consumer identifier; empty lets the broker assign one.
	ConsumerTag string
	// AutoAck disables manual acknowledgement. Leave false (the default) for
	// at-least-once delivery driven by the Handler's error.
	AutoAck bool
	// RequeueOnError controls whether a Handler error requeues the message
	// (default true) or drops/dead-letters it (false).
	RequeueOnError bool
	// Exclusive requests exclusive consumer access to the queue.
	Exclusive bool
	// Args are passed to the underlying Consume call.
	Args amqp.Table
	// contains filtered or unexported fields
}

ConsumerConfig describes what and how to consume. Its Exchange, Queue and Bindings are re-declared on every (re)connect so the consumer survives broker restarts.

func (ConsumerConfig) NoRequeue 

func (c ConsumerConfig) NoRequeue() ConsumerConfig

NoRequeue returns a copy of the config that drops (rather than requeues) messages a Handler rejects, so a poison message does not loop forever. Pair it with a dead-letter exchange to capture rejects.

type Delivery 

type Delivery struct {
	Body            []byte
	RoutingKey      string
	Exchange        string
	ContentType     string
	ContentEncoding string
	Headers         amqp.Table
	DeliveryTag     uint64
	Redelivered     bool
	MessageID       string
	CorrelationID   string
	ReplyTo         string
	Type            string
	AppID           string
	Priority        uint8
	Timestamp       time.Time
}

Delivery is a received message handed to a Handler. It is a value copy of the useful fields of an amqp091 delivery; acknowledgement is handled by the consumer based on the Handler's returned error, so the Handler never acks directly.

type ExchangeConfig 

type ExchangeConfig struct {
	Name       string
	Kind       string // "topic" (default), "direct", "fanout", "headers"
	Durable    bool   // defaults to true via normalize
	AutoDelete bool
	Internal   bool
	NoWait     bool
	Args       amqp.Table
	// contains filtered or unexported fields
}

ExchangeConfig describes an exchange to declare. The zero value declares a durable topic exchange, which is the common default.

func (ExchangeConfig) Transient 

func (e ExchangeConfig) Transient() ExchangeConfig

Transient returns a copy of the exchange config marked non-durable, recording the explicit intent so normalize does not re-apply the durable default.

type Handler 

type Handler func(ctx context.Context, d Delivery) error

Handler processes a single Delivery. Returning nil acknowledges the message; returning an error negatively acknowledges it (requeued or dropped per ConsumerConfig.RequeueOnError). A panic in a Handler is recovered and treated as an error.

type Logger 

type Logger interface {
	Debugf(format string, args ...any)
	Infof(format string, args ...any)
	Warnf(format string, args ...any)
	Errorf(format string, args ...any)
}

Logger is the minimal logging surface this package writes to. It is intentionally tiny so callers can adapt slog, zerolog, zap or a house logger without this package importing any of them. The default is a no-op; supply one with WithLogger.

type NopObserver 

type NopObserver struct{}

NopObserver is an Observer that ignores every event. It is the default and a convenient base to embed when implementing only a subset of the interface.

func (NopObserver) OnConnect 

func (NopObserver) OnConnect()

OnConnect implements Observer.

func (NopObserver) OnConsume 

func (NopObserver) OnConsume(string, Delivery, error)

OnConsume implements Observer.

func (NopObserver) OnDisconnect 

func (NopObserver) OnDisconnect(error)

OnDisconnect implements Observer.

func (NopObserver) OnPublish 

func (NopObserver) OnPublish(string, string, error)

OnPublish implements Observer.

func (NopObserver) OnReconnect 

func (NopObserver) OnReconnect(int)

OnReconnect implements Observer.

type Observer 

type Observer interface {
	// OnConnect fires after a connection (or reconnection) is established.
	OnConnect()
	// OnDisconnect fires when the managed connection is observed closed. err is
	// the broker/transport error, or nil for a deliberate Close.
	OnDisconnect(err error)
	// OnReconnect fires before each re-dial attempt; attempt starts at 1.
	OnReconnect(attempt int)
	// OnPublish fires after each publish attempt with its outcome; err is nil on
	// success.
	OnPublish(exchange, routingKey string, err error)
	// OnConsume fires after a delivery is handled; err is the handler's returned
	// error (nil on success), which also decided ack vs nack.
	OnConsume(queue string, d Delivery, err error)
}

Observer receives lifecycle and message events so callers can wire in metrics and tracing (for example an OpenTelemetry-backed implementation) without this package taking a hard dependency on any telemetry library. Every method must be safe for concurrent use and must not block; the library calls them inline on its connection, publish and consume paths. The default is a no-op; supply one with WithObserver.

type Option 

type Option func(*connOptions)

Option configures a Conn at Connect time.

func WithBackoff 

func WithBackoff(b Backoff) Option

WithBackoff sets the reconnect/retry backoff policy. Unset fields are filled with their defaults.

Example

ExampleWithBackoff tunes the reconnect/retry policy. 

package main

import (
	"context"
	"log"
	"time"

	rabbitmq "github.com/Bugs5382/go-rabbitmq"
)

func main() {
	ctx := context.Background()
	conn, err := rabbitmq.Connect(ctx, "amqp://guest:guest@localhost:5672/",
		rabbitmq.WithBackoff(rabbitmq.Backoff{
			Initial:    time.Second,
			Max:        time.Minute,
			Factor:     2.0,
			Jitter:     0.3,
			MaxRetries: 0, // retry forever
		}),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer func() { _ = conn.Close() }()
}

func WithClientProperties 

func WithClientProperties(props amqp.Table) Option

WithClientProperties advertises client properties (for example a connection name shown in the management UI) to the broker.

func WithConsumeInterceptor 

func WithConsumeInterceptor(interceptors ...ConsumeInterceptor) Option

WithConsumeInterceptor registers one or more consume interceptors, applied to every consumer started on the Conn. They run outermost-first in the order given, wrapping the Handler.

func WithHeartbeat 

func WithHeartbeat(d time.Duration) Option

WithHeartbeat sets the AMQP heartbeat interval. The default is 10s. A value under one second lets the server choose.

func WithLogger 

func WithLogger(l Logger) Option

WithLogger plugs in a logger. The default is a no-op.

func WithObserver 

func WithObserver(obs Observer) Option

WithObserver plugs in an Observer for metrics/tracing. The default is a no-op.

func WithPublishInterceptor 

func WithPublishInterceptor(interceptors ...PublishInterceptor) Option

WithPublishInterceptor registers one or more publish interceptors, applied to every Publisher created from the Conn. They run outermost-first in the order given. This is how the OTel adapter injects tracing; see the otel subpackage.

func WithTLS 

func WithTLS(cfg *tls.Config) Option

WithTLS enables TLS by supplying a client configuration. This is applied when the URL scheme is amqps; use it to present a client certificate (mTLS) or to pin a CA bundle. Passing a non-nil config also forces the connection through the TLS handshake path.

func WithVhost 

func WithVhost(vhost string) Option

WithVhost overrides the virtual host. By default the vhost parsed from the URL is used.

type PublishFunc 

type PublishFunc func(ctx context.Context, exchange, routingKey string, msg *amqp.Publishing) error

PublishFunc performs the actual send of a prepared message. It is the "next" link a PublishInterceptor calls.

type PublishInterceptor 

type PublishInterceptor func(ctx context.Context, exchange, routingKey string, msg *amqp.Publishing, next PublishFunc) error

PublishInterceptor wraps a publish. It runs once per Publish call (not per internal retry) and may inspect or mutate msg before calling next -- for example to inject W3C trace-context headers -- and observe the outcome next returns. Interceptors run outermost-first in the order registered.

It is the seam the OTel adapter uses to start a producer span, inject trace context into the message headers, and record publish metrics, without the core package depending on any telemetry library.

type PublishOption 

type PublishOption func(*amqp.Publishing)

PublishOption customises a single Publish call.

func WithAppID 

func WithAppID(id string) PublishOption

WithAppID sets the publishing application id.

func WithContentType 

func WithContentType(ct string) PublishOption

WithContentType overrides the message Content-Type.

func WithCorrelationID 

func WithCorrelationID(id string) PublishOption

WithCorrelationID sets the correlation id.

func WithExpiration 

func WithExpiration(ms string) PublishOption

WithExpiration sets a per-message TTL, expressed as a millisecond string per the AMQP spec (for example "60000").

func WithHeaders 

func WithHeaders(h amqp.Table) PublishOption

WithHeaders sets AMQP headers on the message.

func WithMessageID 

func WithMessageID(id string) PublishOption

WithMessageID sets the message id.

func WithPersistent 

func WithPersistent(persistent bool) PublishOption

WithPersistent overrides the persistence of a single message.

func WithPriority 

func WithPriority(priority uint8) PublishOption

WithPriority sets the message priority (0-9 on a priority queue).

func WithReplyTo 

func WithReplyTo(queue string) PublishOption

WithReplyTo sets the reply-to address.

func WithType 

func WithType(t string) PublishOption

WithType sets the message type name.

type Publisher 

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

Publisher publishes to one exchange over a Conn. It lazily opens a channel, re-opens (and optionally re-declares its exchange) after a drop, and retries a failed publish within bounded backoff. A Publisher is safe for concurrent use.

func (*Publisher) Close 

func (p *Publisher) Close() error

Close releases the publisher's channel. The underlying Conn is left open.

func (*Publisher) Publish 

func (p *Publisher) Publish(ctx context.Context, routingKey string, body []byte, opts ...PublishOption) error

Publish sends body to the publisher's exchange with the given routing key. It ensures a live channel, re-opening and re-declaring on a drop, and retries within the configured bounded backoff. It returns an error (wrapping ErrPublishFailed) only after the retry budget is exhausted, so an outbox worker can keep the message and try again later.

Any publish interceptors registered on the Conn (for example the OTel adapter's tracing interceptor) wrap the whole call: they run once per Publish, may mutate the message headers before it is sent, and observe the final outcome.

func (*Publisher) PublishJSON 

func (p *Publisher) PublishJSON(ctx context.Context, routingKey string, v any, opts ...PublishOption) error

PublishJSON marshals v to JSON and publishes it with Content-Type application/json (regardless of the configured default).

type PublisherOption 

type PublisherOption func(*publisherOptions)

PublisherOption configures a Publisher at construction time.

func WithDefaultContentType 

func WithDefaultContentType(ct string) PublisherOption

WithDefaultContentType sets the Content-Type stamped on messages that do not override it. The default is application/json.

func WithExchangeDeclare 

func WithExchangeDeclare(cfg ExchangeConfig) PublisherOption

WithExchangeDeclare makes the publisher declare its exchange (once per channel, and again after every reconnect) before publishing. Without it the exchange is assumed to already exist.

func WithMandatoryDefault 

func WithMandatoryDefault(mandatory bool) PublisherOption

WithMandatoryDefault sets the default for the AMQP mandatory flag. The default is false.

func WithPersistentDefault 

func WithPersistentDefault(persistent bool) PublisherOption

WithPersistentDefault sets whether messages are persistent by default. The default is true (delivery mode 2).

func WithPublishRetries 

func WithPublishRetries(n int) PublisherOption

WithPublishRetries bounds how many times a single Publish re-opens the channel and retries after a failure before returning ErrPublishFailed. The default is 3. A value <=0 means "one attempt, no retry".

type QueueConfig 

type QueueConfig struct {
	Name       string
	Type       QueueType // QueueClassic (default) or QueueQuorum
	Durable    bool      // defaults to true via normalize
	AutoDelete bool
	Exclusive  bool
	NoWait     bool
	Args       amqp.Table
	// contains filtered or unexported fields
}

QueueConfig describes a queue to declare. The zero value declares a durable classic queue.

func (QueueConfig) Transient 

func (q QueueConfig) Transient() QueueConfig

Transient returns a copy of the queue config marked non-durable.

type QueueType 

type QueueType string

QueueType selects the broker-side queue implementation. 

const (
	// QueueClassic is the traditional (non-replicated) queue. It is the default
	// and the only type that may be exclusive, auto-delete or server-named.
	QueueClassic QueueType = "classic"
	// QueueQuorum is a replicated, Raft-based durable queue. It must be durable
	// and named, and may not be exclusive or auto-delete.
	QueueQuorum QueueType = "quorum"
)

type Topology 

type Topology struct {
	Exchanges []ExchangeConfig
	Queues    []QueueConfig
	Bindings  []BindingConfig
}

Topology is a bundle of exchanges, queues and bindings that should exist together. DeclareTopology declares them all on a single channel, in order: exchanges, then queues, then bindings.

Source Files

View all Source files

Directories

Path Synopsis
Package otel is the OpenTelemetry adapter for go-rabbitmq.
 Package otel is the OpenTelemetry adapter for go-rabbitmq.

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.