rabbitmq

type Action ¶ added in v0.7.0

type Action int

Action is an action that occurs after processed this delivery

const (
	// Ack default ack this msg after you have successfully processed this delivery.
	Ack Action = iota
	// NackDiscard the message will be dropped or delivered to a server configured dead-letter queue.
	NackDiscard
	// NackRequeue deliver this message to a different consumer.
	NackRequeue
	// Message acknowledgement is left to the user using the msg.Ack() method
	Manual
)

type Binding ¶ added in v0.11.0

type Binding struct {
	RoutingKey string
	BindingOptions
}

Binding describes the bhinding of a queue to a routing key on an exchange

type BindingOptions ¶

type BindingOptions struct {
	NoWait  bool
	Args    Table
	Declare bool
}

BindingOptions describes the options a binding can have

type Config ¶ added in v0.8.0

type Config amqp.Config

Config wraps amqp.Config Config is used in DialConfig and Open to specify the desired tuning parameters used during a connection open handshake. The negotiated tuning will be stored in the returned connection's Config field.

type Confirmation ¶ added in v0.7.0

type Confirmation struct {
	amqp.Confirmation
	ReconnectionCount int
}

Confirmation notifies the acknowledgment or negative acknowledgement of a publishing identified by its delivery tag. Use NotifyPublish to consume these events. ReconnectionCount is useful in that each time it increments, the DeliveryTag is reset to 0, meaning you can use ReconnectionCount+DeliveryTag to ensure uniqueness

type Conn ¶ added in v0.11.0

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

Conn manages the connection to a rabbit cluster it is intended to be shared across publishers and consumers

func NewConn(url string, optionFuncs ...func(*ConnectionOptions)) (*Conn, error)

func (conn *Conn) Close() error

type ConnectionOptions ¶ added in v0.11.0

type ConnectionOptions struct {
	ReconnectInterval time.Duration
	Logger            Logger
	Config            Config
}

ConnectionOptions are used to describe how a new consumer will be created.

type Consumer ¶

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

Consumer allows you to create and connect to queues for data consumption.

func NewConsumer(
	conn *Conn,
	queue string,
	optionFuncs ...func(*ConsumerOptions),
) (*Consumer, error)

func (consumer *Consumer) Close()

func (consumer *Consumer) Run(handler Handler) error

type ConsumerOptions ¶

type ConsumerOptions struct {
	RabbitConsumerOptions RabbitConsumerOptions
	QueueOptions          QueueOptions
	ExchangeOptions       []ExchangeOptions
	Concurrency           int
	Logger                logger.Logger
	QOSPrefetch           int
	QOSGlobal             bool
}

ConsumerOptions are used to describe how a new consumer will be created. If QueueOptions is not nil, the options will be used to declare a queue If ExchangeOptions is not nil, it will be used to declare an exchange If there are Bindings, the queue will be bound to them

type Delivery ¶

type Delivery struct {
	amqp.Delivery
}

Delivery captures the fields for a previously delivered message resident in a queue to be delivered by the server to a consumer from Channel.Consume or Channel.Get.

type ExchangeOptions ¶ added in v0.11.0

type ExchangeOptions struct {
	Name       string
	Kind       string // possible values: empty string for default exchange or direct, topic, fanout
	Durable    bool
	AutoDelete bool
	Internal   bool
	NoWait     bool
	Passive    bool // if false, a missing exchange will be created on the server
	Args       Table
	Declare    bool
	Bindings   []Binding
}

ExchangeOptions are used to configure an exchange. If the Passive flag is set the client will only check if the exchange exists on the server and that the settings match, no creation attempt will be made.

type Handler ¶ added in v0.7.0

type Handler func(d Delivery) (action Action)

Handler defines the handler of each Delivery and return Action

type Logger ¶ added in v0.5.0

type Logger logger.Logger

Logger is describes a logging structure. It can be set using WithPublisherOptionsLogger() or WithConsumerOptionsLogger().

type PublishOptions ¶

type PublishOptions struct {
	Exchange string
	// Mandatory fails to publish if there are no queues
	// bound to the routing key
	Mandatory bool
	// Immediate fails to publish if there are no consumers
	// that can ack bound to the queue on the routing key
	Immediate bool
	// MIME content type
	ContentType string
	// Transient (0 or 1) or Persistent (2)
	DeliveryMode uint8
	// Expiration time in ms that a message will expire from a queue.
	// See https://www.rabbitmq.com/ttl.html#per-message-ttl-in-publishers
	Expiration string
	// MIME content encoding
	ContentEncoding string
	// 0 to 9
	Priority uint8
	// correlation identifier
	CorrelationID string
	// address to to reply to (ex: RPC)
	ReplyTo string
	// message identifier
	MessageID string
	// message timestamp
	Timestamp time.Time
	// message type name
	Type string
	// creating user id - ex: "guest"
	UserID string
	// creating application id
	AppID string
	// Application or exchange specific fields,
	// the headers exchange will inspect this field.
	Headers Table
}

PublishOptions are used to control how data is published

type Publisher ¶

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

Publisher allows you to publish messages safely across an open connection

func NewPublisher(conn *Conn, optionFuncs ...func(*PublisherOptions)) (*Publisher, error)

func (publisher *Publisher) Close()

func (publisher *Publisher) NotifyPublish(handler func(p Confirmation))

func (publisher *Publisher) NotifyReturn(handler func(r Return))

func (publisher *Publisher) Publish(
	data []byte,
	routingKeys []string,
	optionFuncs ...func(*PublishOptions),
) error

func (publisher *Publisher) PublishWithContext(
	ctx context.Context,
	data []byte,
	routingKeys []string,
	optionFuncs ...func(*PublishOptions),
) error

func (publisher *Publisher) PublishWithDeferredConfirmWithContext(
	ctx context.Context,
	data []byte,
	routingKeys []string,
	optionFuncs ...func(*PublishOptions),
) (PublisherConfirmation, error)

type PublisherConfirmation ¶ added in v0.12.2

type PublisherConfirmation []*amqp.DeferredConfirmation

type PublisherOptions ¶ added in v0.1.0

type PublisherOptions struct {
	ExchangeOptions ExchangeOptions
	Logger          Logger
	ConfirmMode     bool
}

PublisherOptions are used to describe a publisher's configuration. Logger is a custom logging interface.

type QueueOptions ¶

type QueueOptions struct {
	Name       string
	Durable    bool
	AutoDelete bool
	Exclusive  bool
	NoWait     bool
	Passive    bool // if false, a missing queue will be created on the server
	Args       Table
	Declare    bool
}

QueueOptions are used to configure a queue. A passive queue is assumed by RabbitMQ to already exist, and attempting to connect to a non-existent queue will cause RabbitMQ to throw an exception.

type RabbitConsumerOptions ¶ added in v0.11.0

type RabbitConsumerOptions struct {
	Name      string
	AutoAck   bool
	Exclusive bool
	NoWait    bool
	NoLocal   bool
	Args      Table
}

RabbitConsumerOptions are used to configure the consumer on the rabbit server

type Return ¶

type Return struct {
	amqp.Return
}

Return captures a flattened struct of fields returned by the server when a Publishing is unable to be delivered either due to the `mandatory` flag set and no route found, or `immediate` flag set and no free consumer.

type Table ¶

type Table map[string]interface{}

Table stores user supplied fields of the following types:

bool
byte
float32
float64
int
int16
int32
int64
nil
string
time.Time
amqp.Decimal
amqp.Table
[]byte
[]interface{} - containing above types

Functions taking a table will immediately fail when the table contains a value of an unsupported type.

The caller must be specific in which precision of integer it wishes to encode.

Use a type assertion when reading values from a table for type conversion.

RabbitMQ expects int32 for integer values.