amqp

Documentation

Index

• Variables
• type Client
  • func New(config Config, prometheusRegister prometheus.Registerer) (*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) (*Client, error)

New inits AMQP connection from Config

func NewFromURI

func NewFromURI(uri string, prefetch int, prometheusRegister prometheus.Registerer) (*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 setup

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
}

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