rabbitmq

Documentation

Overview

Package rabbitmq implements a Celery broker using RabbitMQ and github.com/rabbitmq/amqp091-go.

Index

• Constants
• type Broker
  • func NewBroker(options ...BrokerOption) (*Broker, error)
  • func (br *Broker) Observe(queues []string) error
  • func (br *Broker) Receive() ([]byte, error)
  • func (br *Broker) Send(m []byte, q string) error
• type BrokerOption
  • func WithAmqpUri(amqpUri string) BrokerOption
  • func WithClient(c *amqp.Connection) BrokerOption
  • func WithReceiveTimeout(timeout time.Duration) BrokerOption

Constants

const DefaultAmqpUri = "amqp://guest:guest@localhost:5672/"

DefaultAmqpUri defines the default AMQP URI which is used to connect to RabbitMQ.

const DefaultReceiveTimeout = 5

DefaultReceiveTimeout defines how many seconds the broker's Receive command should block waiting for results from RabbitMQ.

Types

type Broker

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

Broker is a RabbitMQ broker that sends/receives messages from specified queues.

func NewBroker

func NewBroker(options ...BrokerOption) (*Broker, error)

NewBroker creates a broker backed by RabbitMQ. By default, it connects to localhost.

func (*Broker) Observe

func (br *Broker) Observe(queues []string) error

Observe sets the queues from which the tasks should be received. Note, the method is not concurrency safe.

func (*Broker) Receive

func (br *Broker) Receive() ([]byte, error)

Receive fetches a Celery task message from a tail of one of the queues in RabbitMQ. After a timeout it returns nil, nil.

func (*Broker) Send

func (br *Broker) Send(m []byte, q string) error

Send inserts the specified message at the head of the queue. Note, the method is safe to call concurrently.

type BrokerOption

type BrokerOption func(*Broker)

BrokerOption sets up a Broker.

func WithAmqpUri

func WithAmqpUri(amqpUri string) BrokerOption

WithAmqpUri sets the AMQP connection URI to RabbitMQ.

func WithClient

func WithClient(c *amqp.Connection) BrokerOption

WithClient sets RabbitMQ client representing a connection to RabbitMQ.

func WithReceiveTimeout

func WithReceiveTimeout(timeout time.Duration) BrokerOption

WithReceiveTimeout sets a timeout of how long the broker's Receive command should block waiting for results from RabbitMQ. Larger the timeout, longer the client will have to wait for Celery app to exit. Smaller the timeout, more Get commands would have to be sent to RabbitMQ.