rabbitmq

Documentation

Index

• Constants
• type Channel
• type Consumer
  • func NewConsumer(URI string, options ...Option) (*Consumer, error)
  • func (c *Consumer) Cancel(consumerTag string) error
  • func (c *Consumer) Close() error
  • func (c *Consumer) Consume(queue string, receive func(d amqp.Delivery)) (string, error)
• type Option
  • func WithConnectionName(name string) Option
  • func WithConsumerTagPrefix(prefix string) Option
  • func WithLogger(logger *slog.Logger) Option
  • func WithReconnectWait(wait time.Duration) Option
  • func WithReopenChannelWait(wait time.Duration) Option
• type Producer
  • func NewProducer(logger *slog.Logger, url string) Producer
  • func (p *Producer) Produce(channel Channel, correlationId string, payload any) error

Constants

const (
	DefaultReconnectWait     = 5 * time.Second
	DefaultReopenChannelWait = 2 * time.Second
)

Types

type Channel

type Channel string

type Consumer

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

func NewConsumer

func NewConsumer(URI string, options ...Option) (*Consumer, error)

NewConsumer creates a Consumer that synchronously connects/opens a channel to RabbitMQ at the given URI. If the consumer was able to connect/open a channel it will automatically re-connect and re-open connection and channel if they fail. A consumer holds on to one connection and one channel. A consumer can be used to consume multiple times and from multiple goroutines.

func (*Consumer) Cancel

func (c *Consumer) Cancel(consumerTag string) error

Cancel consuming messages for given consumer. The consumer tag is returned by Consume(). It is safe to call this method multiple times and in multiple goroutines.

func (*Consumer) Close

func (c *Consumer) Close() error

Close connection and channel. A new consumer needs to be created in order to consume again after closing it. It is safe to call this method multiple times and in multiple goroutines.

func (*Consumer) Consume

func (c *Consumer) Consume(queue string, receive func(d amqp.Delivery)) (string, error)

Consume registers the consumer to receive messages from given queue. Consume synchronously declares and registers a consumer to the queue. Once registered it will return the consumer identifier and nil error. receive will be called for every message. Pass the consumer identifier to Cancel() to stop consuming messages. Consume will re-consume if the connection or channel close only if it returned successfully at first. Consume can be called multiple times and from multiple goroutines.

type Option

type Option func(*consumerOptions)

An Option configures consumer options.

func WithConnectionName

func WithConnectionName(name string) Option

WithConnectionName sets the `connection_name` property on the RabbitMQ connection. This can aid in observing/debugging connections (RabbitMQ management/CLI).

func WithConsumerTagPrefix

func WithConsumerTagPrefix(prefix string) Option

WithConsumerTagPrefix sets the prefix to the auto-generated consumer tag. The consumer tag is returned by Consume(). This can aid in observing/debugging consumers on a channel (RabbitMQ management/CLI).

func WithLogger

func WithLogger(logger *slog.Logger) Option

WithLogger sets the logger to be used by the consumer. The consumer will log using slog.Default() by default.

func WithReconnectWait

func WithReconnectWait(wait time.Duration) Option

WithReconnectWait sets the duration to wait after a failed attempt to connect to RabbitMQ.

func WithReopenChannelWait

func WithReopenChannelWait(wait time.Duration) Option

WithReopenChannelWait sets the duration to wait after a failed attempt to open a channel on a RabbitMQ connection.

type Producer

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

func NewProducer

func NewProducer(logger *slog.Logger, url string) Producer

func (*Producer) Produce

func (p *Producer) Produce(channel Channel, correlationId string, payload any) error