rabbitmq

Documentation

Overview

Package rabbitmq provides a RabbitMQ backend for the ling-base/mq broker abstraction. It supports automatic reconnection, publisher confirms, QoS prefetch, concurrent consumers, and topology management.

Index

• type Broker
  • func New(cfg Config) (*Broker, error)
  • func (b *Broker) Bind(queue, exchange, routingKey string) error
  • func (b *Broker) Close() error
  • func (b *Broker) Connect() error
  • func (b *Broker) Consumer(queue string, opts mq.ConsumeOptions) (mq.Consumer, error)
  • func (b *Broker) DeclareExchange(name string, opts mq.ExchangeOptions) error
  • func (b *Broker) DeclareQueue(name string, opts mq.QueueOptions) error
  • func (b *Broker) DeleteExchange(name string) error
  • func (b *Broker) DeleteQueue(name string) error
  • func (b *Broker) IsConnected() bool
  • func (b *Broker) Metrics() mq.Metrics
  • func (b *Broker) Producer(exchange string, opts mq.PublishOptions) (mq.Producer, error)
  • func (b *Broker) Unbind(queue, exchange, routingKey string) error
• type Config
  • func DefaultConfig() Config
• type Consumer
  • func (c *Consumer) IsRunning() bool
  • func (c *Consumer) Start(ctx context.Context) error
  • func (c *Consumer) Stop(timeout time.Duration) error
• type Producer
  • func (p *Producer) Close() error
  • func (p *Producer) Publish(ctx context.Context, msg *mq.Message) error

Types

type Broker

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

Broker implements mq.Broker for RabbitMQ.

func New

func New(cfg Config) (*Broker, error)

New creates a new RabbitMQ broker. Call Connect() to establish the connection.

func (*Broker) Bind

func (b *Broker) Bind(queue, exchange, routingKey string) error

Bind binds a queue to an exchange.

func (*Broker) Close

func (b *Broker) Close() error

Close shuts down the broker.

func (*Broker) Connect

func (b *Broker) Connect() error

Connect establishes the connection to RabbitMQ.

func (*Broker) Consumer

func (b *Broker) Consumer(queue string, opts mq.ConsumeOptions) (mq.Consumer, error)

Consumer creates a consumer for the given queue.

func (*Broker) DeclareExchange

func (b *Broker) DeclareExchange(name string, opts mq.ExchangeOptions) error

DeclareExchange declares an exchange.

func (*Broker) DeclareQueue

func (b *Broker) DeclareQueue(name string, opts mq.QueueOptions) error

DeclareQueue declares a queue.

func (*Broker) DeleteExchange

func (b *Broker) DeleteExchange(name string) error

DeleteExchange removes an exchange.

func (*Broker) DeleteQueue

func (b *Broker) DeleteQueue(name string) error

DeleteQueue removes a queue.

func (*Broker) IsConnected

func (b *Broker) IsConnected() bool

IsConnected reports whether the broker is currently connected.

func (*Broker) Metrics

func (b *Broker) Metrics() mq.Metrics

Metrics returns a snapshot of broker metrics.

func (*Broker) Producer

func (b *Broker) Producer(exchange string, opts mq.PublishOptions) (mq.Producer, error)

Producer creates or returns a cached producer for the given exchange.

func (*Broker) Unbind

func (b *Broker) Unbind(queue, exchange, routingKey string) error

Unbind removes a binding.

type Config

type Config struct {
	// URL is the AMQP connection URL.
	// Example: "amqp://guest:guest@localhost:5672/"
	URL string

	// DialerTimeout is the timeout for establishing a connection.
	// Default: 10s.
	DialerTimeout time.Duration

	// ReconnectDelay is the delay between reconnection attempts.
	// Default: 5s.
	ReconnectDelay time.Duration

	// MaxReconnectAttempts is the maximum number of reconnection
	// attempts. 0 = unlimited.
	MaxReconnectAttempts int

	// Heartbeat is the AMQP heartbeat interval. Default: 10s.
	// Set to 0 to disable.
	Heartbeat time.Duration

	// ChannelCacheSize is the maximum number of channels cached per
	// connection. Default: 16.
	ChannelCacheSize int

	// Vhost is the RabbitMQ virtual host. If empty, uses the vhost
	// from the URL or "/".
	Vhost string
}

Config configures a RabbitMQ broker.

func DefaultConfig

func DefaultConfig() Config

DefaultConfig returns a Config with sensible defaults for localhost.

type Consumer

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

Consumer implements mq.Consumer for RabbitMQ.

func (*Consumer) IsRunning

func (c *Consumer) IsRunning() bool

IsRunning reports whether the consumer is actively consuming.

func (*Consumer) Start

func (c *Consumer) Start(ctx context.Context) error

Start begins consuming from the queue.

func (*Consumer) Stop

func (c *Consumer) Stop(timeout time.Duration) error

Stop gracefully stops consuming.

type Producer

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

Producer implements mq.Producer for RabbitMQ.

func (*Producer) Close

func (p *Producer) Close() error

Close releases producer resources.

func (*Producer) Publish

func (p *Producer) Publish(ctx context.Context, msg *mq.Message) error

Publish sends a message to the exchange.