amqp

Documentation

Index

• func ConvertDeliveryToPublishing(message amqp.Delivery) amqp.Publishing
• type Client
  • func New(config Config, prometheusRegister prometheus.Registerer) (*Client, error)
  • func NewFromURI(uri string, prometheusRegister prometheus.Registerer) (*Client, error)
  • func (a *Client) Ack(message amqp.Delivery)
  • func (a *Client) Close()
  • func (a *Client) Consumer(queueName, routingKey, 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 (a *Client) StopListener(consumer string)
  • func (a *Client) Vhost() string
• type Config
  • func Flags(fs *flag.FlagSet, prefix string) Config
• type Connection

Functions

func ConvertDeliveryToPublishing

func ConvertDeliveryToPublishing(message amqp.Delivery) amqp.Publishing

ConvertDeliveryToPublishing convert a delivery to a publishing, for requeuing

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 from Config

func NewFromURI

func NewFromURI(uri string, prometheusRegister prometheus.Registerer) (*Client, error)

NewFromURI inits AMQP connection from given URI

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, routingKey, 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) 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