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(ctx context.Context, payload amqp.Publishing, exchange, routingKey string) (err error)
  • func (c *Client) PublishJSON(ctx context.Context, item any, exchange, routingKey string) (err error)
  • func (c *Client) Publisher(exchangeName, exchangeType string, args amqp.Table) (err error)
  • func (c *Client) SetupExclusive(ctx context.Context, 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, overrides ...flags.Override) Config
• type Connection
• type QueueResolver

Variables

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

Types

type Client

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

func New

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

func NewFromURI

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

func (*Client) Close

func (c *Client) Close()

func (*Client) Consumer

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

func (*Client) DelayedExchange

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

func (*Client) Enabled

func (c *Client) Enabled() bool

func (*Client) Exclusive

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

func (*Client) Listen

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

func (*Client) Ping

func (c *Client) Ping() error

func (*Client) Publish

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

func (*Client) PublishJSON

func (c *Client) PublishJSON(ctx context.Context, item any, exchange, routingKey string) (err error)

func (*Client) Publisher

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

func (*Client) SetupExclusive

func (c *Client) SetupExclusive(ctx context.Context, name string) (err error)

func (*Client) StopListener

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

func (*Client) Vhost

func (c *Client) Vhost() string

type Config

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

func Flags

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

type Connection

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

type QueueResolver

type QueueResolver func() (string, error)