rabbitmq

Documentation

Overview

Package rabbitmq implements the triton-core/amqp module

Index

• Variables
• type Client
  • func NewClient(ctx context.Context, endpoint string) (*Client, error)
  • func (c *Client) Consume(topic string) (msgs <-chan *Delivery, errChan <-chan error, err error)
  • func (c *Client) Done()
  • func (c *Client) Publish(topic string, body []byte) error
  • func (c *Client) SetPrefetch(prefetch int64)
• type Delivery
  • func NewDelivery(ctx context.Context, delivery amqp.Delivery, channel *amqp.Channel) (*Delivery, error)
  • func (d *Delivery) Ack() error
  • func (d *Delivery) Error() error
  • func (d *Delivery) Nack() error
• type DeliveryMetadata
• type MessageQueued

Variables

var (
	// ErrorEnsureExchange is returned when exchanges are unable to be created
	ErrorEnsureExchange = errors.New("failed to ensure exchange")

	// ErrorEnsureConsumerQueues is returned when consumer queues are unable to be created
	ErrorEnsureConsumerQueues = errors.New("failed to ensure consumer queues")

	// ErrorReconnecting is emitted when the channel is reconnecting
	ErrorReconnecting = errors.New("processor is reconnecting")

	// ErrorDied is emitted when a processor dies, with no hope of recovering
	ErrorDied = errors.New("processor died")
)

Types

type Client

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

Client is a RabbitMQ client TODO(jaredallard): all of these maps :(

func NewClient

func NewClient(ctx context.Context, endpoint string) (*Client, error)

NewClient returns a new rabbitmq client

func (*Client) Consume

func (c *Client) Consume(topic string) (msgs <-chan *Delivery, errChan <-chan error, err error)

Consume from a RabbitMQ queue

func (*Client) Done

func (c *Client) Done()

Done waits until this client completely closes

func (*Client) Publish

func (c *Client) Publish(topic string, body []byte) error

Publish a message to an exchange, must be a serialized format

func (*Client) SetPrefetch

func (c *Client) SetPrefetch(prefetch int64)

SetPrefetch updates the prefetch of our channels

type Delivery

type Delivery struct {
	Metadata DeliveryMetadata

	// Delivery is the internal amqp delivery struct
	Delivery amqp.Delivery

	// Channel this message was received on
	Channel *amqp.Channel

	// Context is the context this delivery is running under
	Context context.Context
}

Delivery is an AMQP delivery

func NewDelivery

func NewDelivery(ctx context.Context, delivery amqp.Delivery, channel *amqp.Channel) (*Delivery, error)

NewDelivery creates a delivery object

func (*Delivery) Ack

func (d *Delivery) Ack() error

Ack acks the message

func (*Delivery) Error

func (d *Delivery) Error() error

Error reports an error with a message and reschedules it for a retry TODO(jaredallard): use deadletter queues

func (*Delivery) Nack

func (d *Delivery) Nack() error

Nack dequeues the message

type DeliveryMetadata

type DeliveryMetadata struct {
	Retries int
}

DeliveryMetadata is metadata from AMQP headers

type MessageQueued

type MessageQueued struct {
	// AMQP topic to publish on
	Topic string

	// Message is the AMQP message object
	Message amqp.Publishing

	// Backoff is set when an error has occurred, and is modified
	Backoff uint
}