Versions in this module Expand all Collapse all v1 v1.0.0 Jul 18, 2026 Changes in this version + var ErrClosed = errors.New("rabbitmq: connection manager is closed") + var ErrInvalidQueue = errors.New("rabbitmq: invalid queue configuration") + var ErrNotReady = errors.New("rabbitmq: no live connection available") + var ErrPublishFailed = errors.New("rabbitmq: publish failed after retries") + type Backoff struct + Factor float64 + Initial time.Duration + Jitter float64 + Max time.Duration + MaxRetries int + func DefaultBackoff() Backoff + type BindingConfig struct + Args amqp.Table + Exchange string + NoWait bool + Queue string + RoutingKey string + type Conn struct + func Connect(ctx context.Context, url string, opts ...Option) (*Conn, error) + func (c *Conn) BindQueue(ctx context.Context, cfg BindingConfig) error + func (c *Conn) Close() error + func (c *Conn) Consume(ctx context.Context, cfg ConsumerConfig, handler Handler) error + func (c *Conn) DeclareExchange(ctx context.Context, cfg ExchangeConfig) error + func (c *Conn) DeclareQueue(ctx context.Context, cfg QueueConfig) (amqp.Queue, error) + func (c *Conn) DeclareTopology(ctx context.Context, t Topology) error + func (c *Conn) Healthy() bool + func (c *Conn) NewPublisher(exchange string, opts ...PublisherOption) *Publisher + func (c *Conn) Reconnects() uint64 + type ConsumeInterceptor func(ctx context.Context, d Delivery, next Handler) error + type ConsumerConfig struct + Args amqp.Table + AutoAck bool + Bindings []BindingConfig + ConsumerTag string + Exchange ExchangeConfig + Exclusive bool + Prefetch int + Queue QueueConfig + RequeueOnError bool + func (c ConsumerConfig) NoRequeue() ConsumerConfig + type Delivery struct + AppID string + Body []byte + ContentEncoding string + ContentType string + CorrelationID string + DeliveryTag uint64 + Exchange string + Headers amqp.Table + MessageID string + Priority uint8 + Redelivered bool + ReplyTo string + RoutingKey string + Timestamp time.Time + Type string + type ExchangeConfig struct + Args amqp.Table + AutoDelete bool + Durable bool + Internal bool + Kind string + Name string + NoWait bool + func (e ExchangeConfig) Transient() ExchangeConfig + type Handler func(ctx context.Context, d Delivery) error + type Logger interface + Debugf func(format string, args ...any) + Errorf func(format string, args ...any) + Infof func(format string, args ...any) + Warnf func(format string, args ...any) + type NopObserver struct + func (NopObserver) OnConnect() + func (NopObserver) OnConsume(string, Delivery, error) + func (NopObserver) OnDisconnect(error) + func (NopObserver) OnPublish(string, string, error) + func (NopObserver) OnReconnect(int) + type Observer interface + OnConnect func() + OnConsume func(queue string, d Delivery, err error) + OnDisconnect func(err error) + OnPublish func(exchange, routingKey string, err error) + OnReconnect func(attempt int) + type Option func(*connOptions) + func WithBackoff(b Backoff) Option + func WithClientProperties(props amqp.Table) Option + func WithConsumeInterceptor(interceptors ...ConsumeInterceptor) Option + func WithHeartbeat(d time.Duration) Option + func WithLogger(l Logger) Option + func WithObserver(obs Observer) Option + func WithPublishInterceptor(interceptors ...PublishInterceptor) Option + func WithTLS(cfg *tls.Config) Option + func WithVhost(vhost string) Option + type PublishFunc func(ctx context.Context, exchange, routingKey string, msg *amqp.Publishing) error + type PublishInterceptor func(ctx context.Context, exchange, routingKey string, msg *amqp.Publishing, ...) error + type PublishOption func(*amqp.Publishing) + func WithAppID(id string) PublishOption + func WithContentType(ct string) PublishOption + func WithCorrelationID(id string) PublishOption + func WithExpiration(ms string) PublishOption + func WithHeaders(h amqp.Table) PublishOption + func WithMessageID(id string) PublishOption + func WithPersistent(persistent bool) PublishOption + func WithPriority(priority uint8) PublishOption + func WithReplyTo(queue string) PublishOption + func WithType(t string) PublishOption + type Publisher struct + func (p *Publisher) Close() error + func (p *Publisher) Publish(ctx context.Context, routingKey string, body []byte, opts ...PublishOption) error + func (p *Publisher) PublishJSON(ctx context.Context, routingKey string, v any, opts ...PublishOption) error + type PublisherOption func(*publisherOptions) + func WithDefaultContentType(ct string) PublisherOption + func WithExchangeDeclare(cfg ExchangeConfig) PublisherOption + func WithMandatoryDefault(mandatory bool) PublisherOption + func WithPersistentDefault(persistent bool) PublisherOption + func WithPublishRetries(n int) PublisherOption + type QueueConfig struct + Args amqp.Table + AutoDelete bool + Durable bool + Exclusive bool + Name string + NoWait bool + Type QueueType + func (q QueueConfig) Transient() QueueConfig + type QueueType string + const QueueClassic + const QueueQuorum + type Topology struct + Bindings []BindingConfig + Exchanges []ExchangeConfig + Queues []QueueConfig
Go to main page