amqp

Documentation

Index

• Variables
• type Client
  • func New(config Config, prometheusRegister prometheus.Registerer, tracer trace.Tracer) (*Client, error)
  • func NewFromURI(uri string, prefetch int, prometheusRegister prometheus.Registerer, ...) (*Client, error)
  • func (c *Client) Close()
  • func (c *Client) Consumer(queueName, routingKey, exchangeName string, exclusive bool, dlExchange string) (err error)
  • func (c *Client) DelayedExchange(queueName, exchangeName, routingKey string, retryDelay time.Duration) (delayExchange string, err error)
  • func (c *Client) Enabled() bool
  • func (c *Client) Exclusive(ctx context.Context, name string, timeout time.Duration, ...) (acquired bool, err error)
  • func (c *Client) Listen(queueResolver QueueResolver, exchange, routingKey string) (string, <-chan amqp.Delivery, error)
  • func (c *Client) Ping() error
  • func (c *Client) Publish(payload amqp.Publishing, exchange, routingKey string) error
  • func (c *Client) PublishJSON(item any, exchange, routingKey string) error
  • func (c *Client) Publisher(exchangeName, exchangeType string, args amqp.Table) (err error)
  • func (c *Client) SetupExclusive(name string) (err error)
  • func (c *Client) StopListener(consumer string) (err error)
  • func (c *Client) Vhost() string
• type Config
  • func Flags(fs *flag.FlagSet, prefix string) Config
• type Connection
• type QueueResolver

Variables

var ErrNoConfig = errors.New("URI is required")

ErrNoConfig occurs when URI is not provided.

Types

type Client

type Client struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Client wraps all object required for AMQP usage.

func New

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

New inits AMQP connection from Config.

func NewFromURI

func NewFromURI(uri string, prefetch int, prometheusRegister prometheus.Registerer, tracer trace.Tracer) (*Client, error)

NewFromURI inits AMQP connection from given URI.

func (*Client) Close

func (c *Client) Close()

Close closes opened ressources.

func (*Client) Consumer

func (c *Client) Consumer(queueName, routingKey, exchangeName string, exclusive bool, dlExchange string) (err error)

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

func (*Client) DelayedExchange

func (c *Client) DelayedExchange(queueName, exchangeName, routingKey string, retryDelay time.Duration) (delayExchange string, err error)

DelayedExchange configures dead-letter exchange with given ttl.

func (*Client) Enabled

func (c *Client) Enabled() bool

Enabled checks if connection is set up.

func (*Client) Exclusive

func (c *Client) Exclusive(ctx context.Context, name string, timeout time.Duration, action func(context.Context) error) (acquired bool, err error)

Exclusive get an exclusive lock from given queue during duration.

func (*Client) Listen

func (c *Client) Listen(queueResolver QueueResolver, exchange, routingKey string) (string, <-chan amqp.Delivery, error)

Listen listens to configured queue.

func (*Client) Ping

func (c *Client) Ping() error

Ping checks if connection is live.

func (*Client) Publish

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

Publish sends payload to the underlying exchange.

func (*Client) PublishJSON

func (c *Client) PublishJSON(item any, exchange, routingKey string) error

PublishJSON sends JSON payload to the underlying exchange.

func (*Client) Publisher

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

Publisher configures client for publishing to given exchange.

func (*Client) SetupExclusive

func (c *Client) SetupExclusive(name string) (err error)

SetupExclusive configure the exclusive queue.

func (*Client) StopListener

func (c *Client) StopListener(consumer string) (err error)

StopListener cancel consumer listening.

func (*Client) Vhost

func (c *Client) Vhost() string

Vhost returns connection Vhost.

type Config

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

func Flags

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

type Connection

type Connection interface {
	io.Closer
	Channel() (*amqp.Channel, error)
	IsClosed() bool
}

Connection for AMQP.

type QueueResolver

type QueueResolver func() (string, error)

QueueResolver return the name of the queue to listen.