amqp

Documentation

Index

• Constants
• func NewSession() (*amqp.Connection, error)
• type Channel
• type Connection
  • func Dial(url string, logger zerolog.Logger, reconnectCount int, ...) (Connection, error)
  • func NewConnection(logger zerolog.Logger, reconnectCount int, minReconnectTimeout time.Duration) (Connection, error)
• type Publisher

Constants

const (
	EnvURL = "CPS_AMQP_URL"
)

Environment variables linked to session parameters

Functions

func NewSession

func NewSession() (*amqp.Connection, error)

NewSession creates a new connection to an AMQP bus, using env var EnvCpsAmqpURL as configuration. Use NewConnection for reconnection feature.

Types

type Channel

type Channel interface {
	Consume(
		queue, consumer string,
		autoAck, exclusive, noLocal, noWait bool,
		args amqp.Table,
	) (<-chan amqp.Delivery, error)
	Ack(tag uint64, multiple bool) error
	Nack(tag uint64, multiple bool, requeue bool) error
	Reject(tag uint64, requeue bool) error
	PublishWithContext(
		ctx context.Context,
		exchange, key string,
		mandatory, immediate bool,
		msg amqp.Publishing,
	) error
	Qos(prefetchCount, prefetchSize int, global bool) error
	Close() error
	ExchangeDeclare(name, kind string, durable, autoDelete, internal, noWait bool, args amqp.Table) error
	QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, args amqp.Table) (amqp.Queue, error)
	QueueBind(name, key, exchange string, noWait bool, args amqp.Table) error
	QueuePurge(name string, noWait bool) (int, error)
	QueueInspect(name string) (amqp.Queue, error)
	QueueDelete(name string, ifUnused, ifEmpty, noWait bool) (int, error)
}

Channel is used to implement amqp channel.

type Connection

type Connection interface {
	Channel() (Channel, error)
	IsClosed() bool
	Close() error
}

Connection is used to implement amqp connection.

func Dial

func Dial(url string, logger zerolog.Logger,
	reconnectCount int, minReconnectTimeout time.Duration) (Connection, error)

Dial accepts a string in the AMQP URI format and returns a new amqp connection. If connection is closed it tries to reconnect.

func NewConnection

func NewConnection(logger zerolog.Logger, reconnectCount int, minReconnectTimeout time.Duration) (Connection, error)

NewSession creates a new connection to an AMQP bus, using env var EnvCpsAmqpURL as configuration. New connection tries to reconnect to AMQP if connection is lost.

type Publisher

type Publisher interface {
	// PublishWithContext sends an amqp.Publishing from the client to an exchange on the server.
	PublishWithContext(ctx context.Context, exchange, key string, mandatory, immediate bool, msg amqp.Publishing) error
}

Publisher is an interface that represents a non-consumable AMQP channel. This interface is implemented by amqp.Channel. It should be used in services that only publish to amqp, in order to be able to test them easily by mocking this interface.