amqpextra

Documentation

Index

• type Config
• type ConnectionConfig
  • func (c *ConnectionConfig) Raw() string
  • func (c *ConnectionConfig) String() string
• type Worker
  • func New(c Config) (*Worker, error)
  • func (w *Worker) Close() error
  • func (w *Worker) Publish(ctx context.Context, message broker.Publishable) error
  • func (w *Worker) StartConsumer(handlers map[string]func([]byte) error, ...) error

Types

type Config

type Config struct {
	Connection          ConnectionConfig
	Exchange            string
	Queue               string
	RoutingKey          string
	Prefetch            int
	ReconnectionTimeout time.Duration
	ConnectionTimeout   time.Duration
	InitTimeout         time.Duration
	Logger              *log.Logger
}

Config configures a worker.

type ConnectionConfig

type ConnectionConfig struct {
	Host        string
	User        string
	Password    string
	VirtualHost string
	Port        int
}

ConnectionConfig configures an AMQP connection.

func (*ConnectionConfig) Raw

func (c *ConnectionConfig) Raw() string

Raw returns a raw connection string used to dial. Notice that this will reveal the password if the result is logged.

func (*ConnectionConfig) String

func (c *ConnectionConfig) String() string

String returns a masked connection string safe for logging.

type Worker

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

Worker is a RabbitMQ consumer and publisher. It configures an AMQP channel to consume messages from a queue along with a publisher to publish messages to an exchange.

It transparently recovers from connection failures and publishes are blocked during recovery.

func New

func New(c Config) (*Worker, error)

New allocates, intializes and returns a new Worker instance.

func (*Worker) Close

func (w *Worker) Close() error

Close closes the worker stopping any active consumers and publishes.

func (*Worker) Publish

func (w *Worker) Publish(ctx context.Context, message broker.Publishable) error

Publish publishes a Publishable message to AMQP. It will block until the message is confirmed on the server. In case of connection failures the method will block until connection is restored and message has been confirmed on the server.

If ctx is cancelled before completing the publish the operation is cancelled.

func (*Worker) StartConsumer

func (w *Worker) StartConsumer(handlers map[string]func([]byte) error, eventDropped func(msgType string, msgBody []byte, err error)) error

StartConsumer consumes messages from an AMQP queue. The method is blocking and will always return with ErrBrokerClosed after calls to Close. In case an event is dropped after failed handlings and redelivery func eventDropped is called.

The workers configured queue is declared on startup along with a binding to the exchange with routing key.