rabbitmq

Documentation

Index

• Constants
• type Acknowledger
• type Connection
  • func Connect(ctx context.Context, url string, l Printer) (*Connection, error)
  • func (c *Connection) Close() error
  • func (c *Connection) CreatePublisher() (*Publisher, error)
  • func (c *Connection) QueueClear(ctx context.Context, queue string) (int, error)
  • func (c *Connection) QueueConsume(queue string) (<-chan *Message, func(), error)
  • func (c *Connection) QueueDelete(ctx context.Context, queue string, ifUnused bool, ifEmpty bool) (int, error)
  • func (c *Connection) QueueLen(ctx context.Context, queue string) (int, error)
• type Message
  • func (m *Message) Ack() error
  • func (m *Message) Acknowledger() Acknowledger
  • func (m *Message) Body() []byte
  • func (m *Message) Nack() error
  • func (m *Message) String() string
  • func (m *Message) Timestamp() time.Time
• type Printer
• type Publisher
  • func (p *Publisher) Close() error
  • func (p *Publisher) Publish(ctx context.Context, queue string, msg []byte) error

Constants

const (
	ErrInvalidURL = rabbitmqError(iota)
	ErrUninitialisedConnection
	ErrConnectionClosed
	ErrChannelClosed
	ErrPublisherClosed
	ErrUninitialisedPublisher
	ErrFlowDisabled
	ErrFailedToAckPublishing
	ErrFailedToAcceptPublishing
	ErrEmptyQueueName
	ErrIllegalQos
)

Error codes

const (
	Transient  uint8 = amqp.Transient
	Persistent uint8 = amqp.Persistent
)

DeliveryMode. Transient means higher throughput but messages will not be restored on broker restart. The delivery mode of publishings is unrelated to the durability of the queues they reside on. Transient messages will not be restored to durable queues, persistent messages will be restored to durable queues and lost on non-durable queues during server restart.

This remains typed as uint8 to match Publishing.DeliveryMode. Other delivery modes specific to custom queue implementations are not enumerated here.

Types

type Acknowledger

type Acknowledger interface {
	Ack() error
	Nack() error
}

Acknowledger is the interface satisfied by something that can positively or negatively acknowledge.

type Connection

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

Connection represents a connection to RabbitMQ.

func Connect

func Connect(ctx context.Context, url string, l Printer) (*Connection, error)

Connect returns a connection to the RabbitMQ server at given URL.

func (*Connection) Close

func (c *Connection) Close() error

Close closes the connection.

func (*Connection) CreatePublisher

func (c *Connection) CreatePublisher() (*Publisher, error)

CreatePublisher returns a new publisher, using this connection.

func (*Connection) QueueClear

func (c *Connection) QueueClear(ctx context.Context, queue string) (int, error)

QueueClear deletes all messages from the named queue, returning the number of messages deleted. Messages that have been delivered but not yet acknowledged will not be deleted.

func (*Connection) QueueConsume

func (c *Connection) QueueConsume(queue string) (<-chan *Message, func(), error)

QueueConsume returns a channel containing messages from the given queue using this connection. It also returns a cancel function. When the caller is finished with the channel, the cancel function must be called, otherwise resources may leak.

func (*Connection) QueueDelete

func (c *Connection) QueueDelete(ctx context.Context, queue string, ifUnused bool, ifEmpty bool) (int, error)

QueueDelete deletes the named queue, along with any messages in the queue, returning the number of messages deleted. When ifUnused is true, the queue will not be deleted if there are any consumers on the queue. If there are consumers, an error will be returned. When ifEmpty is true, the queue will not be deleted if there are any messages remaining on the queue. If there are messages, an error will be returned.

func (*Connection) QueueLen

func (c *Connection) QueueLen(ctx context.Context, queue string) (int, error)

QueueLen returns the number of messages in the given queue.

type Message

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

Message represents a message.

func (*Message) Ack

func (m *Message) Ack() error

Ack positively acknowledges the message.

func (*Message) Acknowledger

func (m *Message) Acknowledger() Acknowledger

Acknowledger returns an acknowledger for this message.

func (*Message) Body

func (m *Message) Body() []byte

Body returns the message body.

func (*Message) Nack

func (m *Message) Nack() error

Nack negatively acknowledges the message, and requeues it.

func (*Message) String

func (m *Message) String() string

String returns the message body as a string.

func (*Message) Timestamp

func (m *Message) Timestamp() time.Time

Timestamp returns the timestamp of the message, if set.

type Printer

type Printer interface {
	Printf(format string, v ...interface{})
}

Printer is the interface satisfied by the Printf method.

type Publisher

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

Publisher is used to publish messages to queues.

func (*Publisher) Close

func (p *Publisher) Close() error

Close closes the publisher.

func (*Publisher) Publish

func (p *Publisher) Publish(ctx context.Context, queue string, msg []byte) error

Publish publishes a message to the given queue.