amqp

Documentation

Index

• Variables
• func ConvertDeliveryToPublishing(message amqp.Delivery) amqp.Publishing
• func GetDeathCount(message amqp.Delivery) (int64, error)
• type Client
  • func New(config Config, prometheusRegister prometheus.Registerer) (*Client, error)
  • func (a *Client) Ack(message amqp.Delivery)
  • func (a *Client) Close()
  • func (a *Client) Consumer(queueName, topic, exchangeName string, retryDelay time.Duration) (string, error)
  • func (a *Client) Enabled() bool
  • func (a *Client) Listen(queue string) (string, <-chan amqp.Delivery, error)
  • func (a *Client) Ping() error
  • func (a *Client) Publish(payload amqp.Publishing, exchange string) error
  • func (a *Client) Publisher(exchangeName, exchangeType string, args amqp.Table) error
  • func (a *Client) Reject(message amqp.Delivery, requeue bool)
  • func (c *Client) Retry(message amqp.Delivery, maxRetry int64, delayExchange string) (DeliveryStatus, error)
  • func (a *Client) StopListener(consumer string)
  • func (a *Client) Vhost() string
• type Config
  • func Flags(fs *flag.FlagSet, prefix string) Config
• type Connection
• type DeliveryStatus

Variables

var ErrNoDeathCount = errors.New("no death count")

ErrNoDeathCount occurs when no death count is found in message

Functions

func ConvertDeliveryToPublishing

func ConvertDeliveryToPublishing(message amqp.Delivery) amqp.Publishing

ConvertDeliveryToPublishing convert a delivery to a publishing, for requeuing

func GetDeathCount

func GetDeathCount(message amqp.Delivery) (int64, error)

GetDeathCount of a message

Types

type Client

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

Client wraps all object required for AMQP usage

func New

func New(config Config, prometheusRegister prometheus.Registerer) (*Client, error)

New inits AMQP connection, channel and queue

func (*Client) Ack

func (a *Client) Ack(message amqp.Delivery)

Ack ack a message with error handling

func (*Client) Close

func (a *Client) Close()

Close closes opened ressources

func (*Client) Consumer

func (a *Client) Consumer(queueName, topic, exchangeName string, retryDelay time.Duration) (string, error)

Consumer configures client for consumming from given queue, bind to given exchange, and return delayed Exchange name to publish

func (*Client) Enabled

func (a *Client) Enabled() bool

Enabled checks if connection is setup

func (*Client) Listen

func (a *Client) Listen(queue string) (string, <-chan amqp.Delivery, error)

Listen listens to configured queue

func (*Client) Ping

func (a *Client) Ping() error

Ping checks if connection is live

func (*Client) Publish

func (a *Client) Publish(payload amqp.Publishing, exchange string) error

Publish sends payload to the underlying exchange

func (*Client) Publisher

func (a *Client) Publisher(exchangeName, exchangeType string, args amqp.Table) error

Publisher configures client for publishing to given exchange

func (*Client) Reject

func (a *Client) Reject(message amqp.Delivery, requeue bool)

Reject reject a message with error handling

func (*Client) Retry

func (c *Client) Retry(message amqp.Delivery, maxRetry int64, delayExchange string) (DeliveryStatus, error)

Retry a message if possible on error

func (*Client) StopListener

func (a *Client) StopListener(consumer string)

StopListener cancel consumer listening

func (*Client) Vhost

func (a *Client) Vhost() string

Vhost returns connection Vhost

type Config

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

Config of package

func Flags

func Flags(fs *flag.FlagSet, prefix string) Config

Flags adds flags for configuring package

type Connection

type Connection interface {
	io.Closer
	IsClosed() bool
}

Connection for AMQP

type DeliveryStatus

type DeliveryStatus int

DeliveryStatus on error

const (
	// DeliveryRejected when message is dropped
	DeliveryRejected DeliveryStatus = iota
	// DeliveryDelayed when message is sent in delay queue
	DeliveryDelayed
)