types

package
v0.11.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 9, 2025 License: MIT Imports: 15 Imported by: 0

Documentation

Index

Constants

View Source 
const (

	// In order to prevent the broker from requeuing the message to th end of the queue, we need to set this limit in order for at least the
	// first N requeues to be requeued to the front of the queue.
	// https://www.rabbitmq.com/docs/quorum-queues#repeated-requeues
	DefaultQueueDeliveryLimit = 20
)
View Source 
const (
	/*
		ExchangeKeyDeadLetter can be used in order to create a dead letter exchange (reference: https://www.rabbitmq.com/dlx.html)
		Some queue consumers may be unable to process certain alerts, and the queue itself may reject messages as a result of certain events.
		For instance, a message is dropped if there is no matching queue for it. In that instance, Dead Letter Exchanges must be implemented
		so that those messages can be saved and reprocessed later. The “Dead Letter Exchange” is an AMQP enhancement provided by RabbitMQ.
		This exchange has the capability of capturing messages that are not deliverable.
	*/
	ExchangeKeyDeadLetter = "x-dead-letter-exchange"
)

Variables

View Source 
var (
	ErrInvalidConnectURL = errors.New("invalid connection url")

	// ErrConnectionFailed is just a generic error that is not checked
	// explicitly against in the code.
	ErrConnectionFailed = errors.New("connection failed")

	ErrClosed = errors.New("closed")

	// ErrNotFound is returned by ExchangeDeclarePassive or QueueDeclarePassive in the case that
	// the queue was not found.
	ErrNotFound = errors.New("not found")

	// ErrBlockingFlowControl is returned when the server is under flow control
	// Your HTTP api may return 503 Service Unavailable or 429 Too Many Requests with a Retry-After header (https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Retry-After)
	ErrBlockingFlowControl = errors.New("blocking flow control")

	// ErrReturned is returned when a message is returned by the server when publishing
	ErrReturned = errors.New("returned")

	// ErrReject can be used to reject a specific message
	// This is a special error that negatively acknowledges messages and does not reuque them.
	ErrReject = errors.New("message rejected")
)
View Source 
var (
	// ErrNack is returned in case the broker did not acknowledge a published message
	ErrNack = errors.New("message not acked")

	// returned when a user tries to await confirmations without configuring them for the session
	ErrNoConfirms = errors.New("confirmations are disabled for this session")

	// ErrDeliveryTagMismatch is returne din case we receive a publishing confirmation that
	// contains a delivery tag that doe snot match the one we expect.
	ErrDeliveryTagMismatch = errors.New("delivery tag mismatch")

	ErrDeliveryClosed = errors.New("delivery channel closed")
)
View Source 
var (
	QuorumQueue = Table{
		"x-queue-type": "quorum",
	}
)

Functions

This section is empty.

Types

type BackoffFunc 

type BackoffFunc func(retry int) (sleep time.Duration)

func NewBackoffPolicy 

func NewBackoffPolicy(minDuration, maxDuration time.Duration) BackoffFunc

NewBackoffPolicy creates a new backoff policy with the given min and max duration.

type Connection 

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

Connection is an internal representation of amqp.Connection.

func NewConnection 

func NewConnection(ctx context.Context, connectUrl, name string, options ...ConnectionOption) (*Connection, error)

NewConnection creates a connection wrapper. name: unique connection name

func (*Connection) BlockingFlowControl 

func (ch *Connection) BlockingFlowControl() <-chan amqp.Blocking

func (*Connection) Close 

func (ch *Connection) Close() (err error)

func (*Connection) Connect 

func (ch *Connection) Connect(ctx context.Context) error

Connect tries to connect (or reconnect) Does not block indefinitely, but returns an error upon connection failure.

func (*Connection) Error 

func (ch *Connection) Error() error

Error returns the first error from the errors channel and flushes all other pending errors from the channel In case that there are no errors, nil is returned.

func (*Connection) Flag 

func (ch *Connection) Flag(err error)

Flag flags the connection as broken which must be recovered. A flagged connection implies a closed connection. Flagging of a connectioncan only be undone by Recover-ing the connection.

func (*Connection) IsCached 

func (c *Connection) IsCached() bool

IsCached returns true in case this session is supposed to be returned to a session pool.

func (*Connection) IsClosed 

func (ch *Connection) IsClosed() bool

func (*Connection) IsFlagged 

func (ch *Connection) IsFlagged() bool

func (*Connection) Name 

func (c *Connection) Name() string

Name returns the name of the connection

func (*Connection) Recover 

func (ch *Connection) Recover(ctx context.Context) error

Recover tries to recover the connection until a shutdown occurs via context cancelation or until the passed context is closed.

type ConnectionOption 

type ConnectionOption func(*connectionOption)

func ConnectionWithBackoffPolicy 

func ConnectionWithBackoffPolicy(policy BackoffFunc) ConnectionOption

ConnectionWithBackoffPolicy influences the sleep interval between connection recovery retries.

func ConnectionWithCached 

func ConnectionWithCached(cached bool) ConnectionOption

ConnectionWithCached makes a connection a cached connection This is only necessary for the connection pool, as cached connections are part of a pool and can be returned back to the pool without being closed.

func ConnectionWithHeartbeatInterval 

func ConnectionWithHeartbeatInterval(interval time.Duration) ConnectionOption

ConnectionHeartbeatInterval allows to set a custom heartbeat interval, that MUST be >= 1 * time.Second

func ConnectionWithLogger 

func ConnectionWithLogger(logger *slog.Logger) ConnectionOption

ConnectionWithLogger allows to set a logger. By default no logger is set.

func ConnectionWithRecoverCallback 

func ConnectionWithRecoverCallback(callback ConnectionRecoverCallback) ConnectionOption

ConnectionWithRecoverCallback allows to set a custom recover callback.

func ConnectionWithTLS 

func ConnectionWithTLS(config *tls.Config) ConnectionOption

ConnectionWithTLS allows to configure tls connectivity.

func ConnectionWithTimeout 

func ConnectionWithTimeout(timeout time.Duration) ConnectionOption

ConnectionWithTimeout allows to set a custom connection timeout, that MUST be >= 1 * time.Second

type ConnectionRecoverCallback 

type ConnectionRecoverCallback func(name string, retry int, err error)

ConnectionRecoverCallback is a function that can be called after a connection failed to be established and is about to be recovered.

type ConsumeOptions 

type ConsumeOptions struct {
	// The consumer is identified by a string that is unique and scoped for all consumers on this channel. If you wish to eventually cancel the consumer, use the same non-empty identifier in Channel.Cancel.
	// An empty string will cause the library to generate a unique identity.
	// The consumer identity will be included in every Delivery in the ConsumerTag field
	ConsumerTag string
	// When AutoAck (also known as noAck) is true, the server will acknowledge deliveries to this consumer prior to writing the delivery to the network. When autoAck is true, the consumer should not call Delivery.Ack.
	// Automatically acknowledging deliveries means that some deliveries may get lost if the consumer is unable to process them after the server delivers them. See http://www.rabbitmq.com/confirms.html for more details.
	AutoAck bool
	// When Exclusive is true, the server will ensure that this is the sole consumer from this queue. When exclusive is false, the server will fairly distribute deliveries across multiple consumers.
	Exclusive bool
	// The NoLocal flag is not supported by RabbitMQ.
	// It's advisable to use separate connections for Channel.Publish and Channel.Consume so not to have TCP pushback on publishing affect the ability to consume messages, so this parameter is here mostly for completeness.
	NoLocal bool
	// When NoWait is true, do not wait for the server to confirm the request and immediately begin deliveries. If it is not possible to consume, a channel exception will be raised and the channel will be closed.
	// Optional arguments can be provided that have specific semantics for the queue or server.
	NoWait bool
	// Args are aditional implementation dependent parameters.
	Args Table
}

type Delivery 

type Delivery struct {
	Headers Table // Application or header exchange table

	// Properties
	ContentType     string    // MIME content type
	ContentEncoding string    // MIME content encoding
	DeliveryMode    uint8     // queue implementation use - non-persistent (1) or persistent (2)
	Priority        uint8     // queue implementation use - 0 to 9
	CorrelationId   string    // application use - correlation identifier
	ReplyTo         string    // application use - address to reply to (ex: RPC)
	Expiration      string    // implementation use - message expiration spec
	MessageId       string    // application use - message identifier
	Timestamp       time.Time // application use - message timestamp
	Type            string    // application use - message type name
	UserId          string    // application use - creating user - should be authenticated user
	AppId           string    // application use - creating application id

	// Valid only with Channel.Consume
	ConsumerTag string

	// Valid only with Channel.Get
	MessageCount uint32

	DeliveryTag uint64
	Redelivered bool
	Exchange    string // basic.publish exchange
	RoutingKey  string // basic.publish routing key

	Body []byte
}

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 Delivery struct {
	Headers Table // Application or header exchange table

	// Properties
	ContentType     string    // MIME content type
	ContentEncoding string    // MIME content encoding
	DeliveryMode    uint8     // queue implementation use - non-persistent (1) or persistent (2)
	Priority        uint8     // queue implementation use - 0 to 9
	CorrelationId   string    // application use - correlation identifier
	ReplyTo         string    // application use - address to reply to (ex: RPC)
	Expiration      string    // implementation use - message expiration spec
	MessageId       string    // application use - message identifier
	Timestamp       time.Time // application use - message timestamp
	Type            string    // application use - message type name
	UserId          string    // application use - creating user - should be authenticated user
	AppId           string    // application use - creating application id

	// Valid only with Channel.Consume
	ConsumerTag string

	// Valid only with Channel.Get
	MessageCount uint32

	DeliveryTag uint64
	Redelivered bool
	Exchange    string // basic.publish exchange
	RoutingKey  string // basic.publish routing key

	Body []byte
}

type ExchangeBindOptions 

type ExchangeBindOptions struct {
	// When NoWait is true, do not wait for the server to confirm the binding.  If any
	// error occurs the channel will be closed.  Add a listener to NotifyClose to
	// handle these errors.
	NoWait bool

	// Optional arguments specific to the exchanges bound can also be specified.
	Args Table
}

type ExchangeDeclareOptions 

type ExchangeDeclareOptions struct {
	// Durable and Non-Auto-Deleted exchanges will survive server restarts and remain
	// declared when there are no remaining bindings.  This is the best lifetime for
	// long-lived exchange configurations like stable routes and default exchanges.
	//
	// Non-Durable and Auto-Deleted exchanges will be deleted when there are no
	// remaining bindings and not restored on server restart.  This lifetime is
	// useful for temporary topologies that should not pollute the virtual host on
	// failure or after the consumers have completed.
	//
	// Non-Durable and Non-Auto-deleted exchanges will remain as long as the server is
	// running including when there are no remaining bindings.  This is useful for
	// temporary topologies that may have long delays between bindings.
	//
	// Durable and Auto-Deleted exchanges will survive server restarts and will be
	// removed before and after server restarts when there are no remaining bindings.
	// These exchanges are useful for robust temporary topologies or when you require
	// binding durable queues to auto-deleted exchanges.
	//
	// Note: RabbitMQ declares the default exchange types like 'amq.fanout' as
	// durable, so queues that bind to these pre-declared exchanges must also be
	// durable.
	Durable bool
	// Durable and Non-Auto-Deleted exchanges will survive server restarts and remain
	// declared when there are no remaining bindings.  This is the best lifetime for
	// long-lived exchange configurations like stable routes and default exchanges.
	//
	// Non-Durable and Auto-Deleted exchanges will be deleted when there are no
	// remaining bindings and not restored on server restart.  This lifetime is
	// useful for temporary topologies that should not pollute the virtual host on
	// failure or after the consumers have completed.
	//
	// Non-Durable and Non-Auto-deleted exchanges will remain as long as the server is
	// running including when there are no remaining bindings.  This is useful for
	// temporary topologies that may have long delays between bindings.
	//
	// Durable and Auto-Deleted exchanges will survive server restarts and will be
	// removed before and after server restarts when there are no remaining bindings.
	// These exchanges are useful for robust temporary topologies or when you require
	// binding durable queues to auto-deleted exchanges.
	AutoDelete bool
	// Exchanges declared as `internal` do not accept accept publishings. Internal
	// exchanges are useful when you wish to implement inter-exchange topologies
	// that should not be exposed to users of the broker.
	Internal bool
	// When NoWait is true, declare without waiting for a confirmation from the server.
	// The channel may be closed as a result of an error.  Add a NotifyClose listener
	// to respond to any exceptions.
	NoWait bool
	// Optional Table of arguments that are specific to the server's implementation of
	// the exchange can be sent for exchange types that require extra parameters.
	Args Table
}

type ExchangeDeleteOptions 

type ExchangeDeleteOptions struct {
	// When IfUnused is true, the server will only delete the exchange if it has no queue
	// bindings.  If the exchange has queue bindings the server does not delete it
	// but close the channel with an exception instead.  Set this to true if you are
	// not the sole owner of the exchange.
	IfUnused bool
	// When NoWait is true, do not wait for a server confirmation that the exchange has
	// been deleted.
	NoWait bool
}

type ExchangeKind 

type ExchangeKind string
const (
	/*
		The first RabbitMQ exchange type, the direct exchange, uses a message routing key to transport messages to queues.
		The routing key is a message attribute that the producer adds to the message header. You can consider the routing
		key to be an “address” that the exchange uses to determine how the message should be routed. A message is delivered
		to the queue with the binding key that exactly matches the message’s routing key.

		The direct exchange’s default exchange is “amq. direct“, which AMQP brokers must offer for communication.
		As is shown in the figure, queue A (create_pdf_queue) is tied to a direct exchange (pdf_events) with the binding
		key “pdf_create”. When a new message arrives at the direct exchange with the routing key “pdf_create”, the exchange
		sends it to the queue where the binding key = routing key; which is queue A in this example (create_pdf_queue).
	*/
	ExchangeKindDirect ExchangeKind = "direct"

	/*
		A fanout exchange, like direct and topic exchange, duplicates and routes a received message to any associated queues,
		regardless of routing keys or pattern matching. Here, your provided keys will be entirely ignored.

		Fanout exchanges are useful when the same message needs to be passed to one or perhaps more queues with consumers who may
		process the message differently. As shown in the image, a message received by the fanout exchange is copied and routed to all
		three queues associated with the exchange. When something happens, such as a sporting event or weather forecast, all connected
		mobile devices will be notified. For the fanout RabbitMQ exchange type, “amq.fanout” is the default exchange that must be provided
		by AMQP brokers.
	*/
	ExchangeKindFanOut ExchangeKind = "fanout"

	/*
		Topic RabbitMQ exchange type sends messages to queues depending on wildcard matches between the routing key and the queue binding’s routing pattern.
		Messages are routed to one or more queues based on a pattern that matches a message routing key. A list of words separated by a period must be used
		as the routing key (.).

		The routing patterns may include an asterisk (“*”) to match a word in a specified position of the routing key (for example, a routing pattern of
		“agreements.*.*.b.*” only matches routing keys with “agreements” as the first word and “b” as the fourth word).
		A pound symbol (“#”) denotes a match of zero or more words.

		In topic exchange, consumers indicate which topics are of interest to them. The consumer establishes a queue and binds it to the exchange using a
		certain routing pattern. All messages with a routing key that matches the routing pattern are routed to the queue, where they will remain until
		the consumer consumes them. For the topic RabbitMQ exchange type, “amq.topic” is the default topic exchange that AMQP brokers must provide for
		message exchange.
	*/
	ExchangeKindTopic ExchangeKind = "topic"

	/*
		A headers RabbitMQ exchange type is a message routing system that uses arguments with headers and optional values to route messages.
		Header exchanges are identical to topic exchanges, except that instead of using routing keys, messages are routed based on header values.
		If the value of the header equals the value of supply during binding, the message matches.

		In the binding between exchange and queue, a specific argument termed “x-match” indicates whether all headers must match or only one.
		For the message to match, any common header between the message and the binding should match, or all of the headers referenced in the
		binding must be present in the message.

		The “x-match” property has two possible values: “any” and “all,” with “all” being the default. A value of “all” indicates that all
		header pairs (key, value) must match, whereas “any” indicates that at least one pair must match. Instead of a string, headers can be
		built with a larger range of data types, such as integers or hashes. The headers exchange type (when used with the binding option “any”)
		is useful for steering messages containing a subset of known (unordered) criteria.

		For the header RabbitMQ exchange type, “amq.headers” is the default topic exchange that AMQP brokers must supply.
	*/
	ExchangeKindHeaders ExchangeKind = "headers"
)

type ExchangeUnbindOptions 

type ExchangeUnbindOptions struct {
	// When NoWait is true, do not wait for the server to confirm the deletion of the
	// binding.  If any error occurs the channel will be closed.  Add a listener to
	// NotifyClose to handle these errors.
	NoWait bool

	// Optional arguments that are specific to the type of exchanges bound can also be
	// provided.  These must match the same arguments specified in ExchangeBind to
	// identify the binding.
	Args Table
}

ExchangeUnbindOptions can be used to configure additional unbind options.

type Publishing 

type Publishing struct {
	// Application or exchange specific fields,
	// the headers exchange will inspect this field.
	Headers Table

	// Properties
	ContentType     string    // MIME content type
	ContentEncoding string    // MIME content encoding
	DeliveryMode    uint8     // Persistent (0 or 2) or Transient (1) (different from rabbitmq/amqp091-go library)
	Priority        uint8     // 0 to 9
	CorrelationId   string    // correlation identifier
	ReplyTo         string    // address to to reply to (ex: RPC)
	Expiration      string    // message expiration spec
	MessageId       string    // message identifier
	Timestamp       time.Time // message timestamp
	Type            string    // message type name
	UserId          string    // creating user id - ex: "guest"
	AppId           string    // creating application id

	// The application specific payload of the message
	Body []byte

	// Since publishings are asynchronous, any undeliverable message will get returned by the server.
	// Add a listener with Channel.NotifyReturn to handle any undeliverable message when calling publish with either the mandatory or immediate parameters as true.
	// Publishings can be undeliverable when the mandatory flag is true and no queue is bound that matches the routing key,
	// or when the immediate flag is true and no consumer on the matched queue is ready to accept the delivery.
	// This can return an error when the channel, connection or socket is closed. The error or lack of an error does not indicate whether the server has received this publishing.
	Mandatory bool
	Immediate bool
}

Publishing captures the client message sent to the server. The fields outside of the Headers table included in this struct mirror the underlying fields in the content frame. They use native types for convenience and efficiency.

type Queue 

type Queue amqp091.Queue
type Queue struct {
    Name      string // server confirmed or generated name
    Messages  int    // count of messages not awaiting acknowledgment
    Consumers int    // number of consumers receiving deliveries
}

Queue captures the current server state of the queue on the server returned from Channel.QueueDeclare or Channel.QueueInspect.

type QueueBindOptions 

type QueueBindOptions struct {
	// When NoWait is false and the queue could not be bound, the channel will be
	// closed with an error.
	NoWait bool
	// Additional implementation specific arguments
	Args Table
}

type QueueDeclareOptions 

type QueueDeclareOptions struct {
	// Durable and Non-Auto-Deleted queues will survive server restarts and remain
	// when there are no remaining consumers or bindings.  Persistent publishings will
	// be restored in this queue on server restart.  These queues are only able to be
	// bound to durable exchanges.
	//
	// Non-Durable and Non-Auto-Deleted queues will remain declared as long as the
	// server is running regardless of how many consumers.  This lifetime is useful
	// for temporary topologies that may have long delays between consumer activity.
	// These queues can only be bound to non-durable exchanges.
	//
	// Durable and Auto-Deleted queues will be restored on server restart, but without
	// active consumers will not survive and be removed.  This Lifetime is unlikely
	// to be useful.
	Durable bool
	// Non-Durable and Auto-Deleted queues will not be redeclared on server restart
	// and will be deleted by the server after a short time when the last consumer is
	// canceled or the last consumer's channel is closed.  Queues with this lifetime
	// can also be deleted normally with QueueDelete.  These durable queues can only
	// be bound to non-durable exchanges.
	//
	// Non-Durable and Non-Auto-Deleted queues will remain declared as long as the
	// server is running regardless of how many consumers.  This lifetime is useful
	// for temporary topologies that may have long delays between consumer activity.
	// These queues can only be bound to non-durable exchanges.
	//
	// Durable and Auto-Deleted queues will be restored on server restart, but without
	// active consumers will not survive and be removed.  This Lifetime is unlikely
	// to be useful.
	AutoDelete bool
	// Exclusive queues are only accessible by the connection that declares them and
	// will be deleted when the connection closes.  Channels on other connections
	// will receive an error when attempting  to declare, bind, consume, purge or
	// delete a queue with the same name.
	Exclusive bool
	// When noWait is true, the queue will assume to be declared on the server.  A
	// channel exception will arrive if the conditions are met for existing queues
	// or attempting to modify an existing queue from a different connection.
	//
	NoWait bool
	// Args are additional properties you can set, like the queue type.
	Args Table
}

QueueDeclareOptions can be passed to the queue declaration If you want to change your default queue behavior.

type QueueDeleteOptions 

type QueueDeleteOptions struct {
	// When IfUnused is true, the queue will not be deleted if there are any
	// consumers on the queue.  If there are consumers, an error will be returned and
	// the channel will be closed.
	IfUnused bool
	// When IfEmpty is true, the queue will not be deleted if there are any messages
	// remaining on the queue.  If there are messages, an error will be returned and
	// the channel will be closed.
	IfEmpty bool
	// When NoWait is true, the queue will be deleted without waiting for a response
	// from the server.  The purged message count will not be meaningful. If the queue
	// could not be deleted, a channel exception will be raised and the channel will
	// be closed.
	NoWait bool
}

QueueDeleteOptions are options for deleting a queue.

type QueuePurgeOptions 

type QueuePurgeOptions struct {
	// If NoWait is true, do not wait for the server response and the number of messages purged will not be meaningful.
	NoWait bool
}

type Session 

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

Session is a wrapper for an amqp channel. It MUST not be used in a multithreaded context, but only in a single goroutine.

func NewSession 

func NewSession(conn *Connection, name string, options ...SessionOption) (*Session, error)

NewSession wraps a connection and a channel in order tointeract with the message broker. By default the context of the parent connection is used for cancellation.

func (*Session) Ack 

func (s *Session) Ack(deliveryTag uint64, multiple bool) (err error)

Ack confirms the processing of the message. In case the underlying channel dies, you cannot send a nack for the processed message. You might receive the message again from the broker, as it expects a n/ack

func (*Session) AwaitConfirm 

func (s *Session) AwaitConfirm(ctx context.Context, expectedTag uint64) (err error)

AwaitConfirm tries to await a confirmation from the broker for a published message You may check for ErrNack in order to see whether the broker rejected the message temporatily. WARNING: AwaitConfirm cannot be retried in case the channel dies or errors. You must resend your message and attempt to await it again.

func (*Session) Close 

func (s *Session) Close() (err error)

Close closes the session completely. Do not use this method in case you have acquired the session from a connection pool. Use the ConnectionPool.ResurnSession method in order to return the session.

func (*Session) Connect 

func (s *Session) Connect() (err error)

Connect tries to create (or re-create) the channel from the Connection it is derived from.

func (*Session) ConsumeWithContext 

func (s *Session) ConsumeWithContext(ctx context.Context, queue string, option ...ConsumeOptions) (<-chan amqp091.Delivery, error)

Consume immediately starts delivering queued messages.

Begin receiving on the returned chan Delivery before any other operation on the Connection or Channel. Continues deliveries to the returned chan Delivery until Channel.Cancel, Connection.Close, Channel.Close, or an AMQP exception occurs. Consumers must range over the chan to ensure all deliveries are received.

Unreceived deliveries will block all methods on the same connection. All deliveries in AMQP must be acknowledged. It is expected of the consumer to call Delivery.Ack after it has successfully processed the delivery.

If the consumer is cancelled or the channel or connection is closed any unacknowledged deliveries will be requeued at the end of the same queue.

Inflight messages, limited by Channel.Qos will be buffered until received from the returned chan. When the Channel or Connection is closed, all buffered and inflight messages will be dropped. When the consumer identifier tag is cancelled, all inflight messages will be delivered until the returned chan is closed.

func (*Session) Error 

func (s *Session) Error() error

Error returns all errors from the errors channel and flushes all other pending errors from the channel In case that there are no errors, nil is returned.

func (*Session) ExchangeBind 

func (s *Session) ExchangeBind(ctx context.Context, destination string, routingKey string, source string, option ...ExchangeBindOptions) error

ExchangeBind binds an exchange to another exchange to create inter-exchange routing topologies on the server. This can decouple the private topology and routing exchanges from exchanges intended solely for publishing endpoints.

Binding two exchanges with identical arguments will not create duplicate bindings.

Binding one exchange to another with multiple bindings will only deliver a message once. For example if you bind your exchange to `amq.fanout` with two different binding keys, only a single message will be delivered to your exchange even though multiple bindings will match.

Given a message delivered to the source exchange, the message will be forwarded to the destination exchange when the routing key is matched. 

ExchangeBind("sell", "MSFT", "trade", false, nil)
ExchangeBind("buy", "AAPL", "trade", false, nil)

Delivery       Source      Key      Destination
example        exchange             exchange
-----------------------------------------------
key: AAPL  --> trade ----> MSFT     sell
                     \---> AAPL --> buy

func (*Session) ExchangeDeclare 

func (s *Session) ExchangeDeclare(ctx context.Context, name string, kind ExchangeKind, option ...ExchangeDeclareOptions) error

ExchangeDeclare declares an exchange on the server. If the exchange does not already exist, the server will create it. If the exchange exists, the server verifies that it is of the provided type, durability and auto-delete flags.

Errors returned from this method will close the channel.

Exchange names starting with "amq." are reserved for pre-declared and standardized exchanges. The client MAY declare an exchange starting with "amq." if the passive option is set, or the exchange already exists. Names can consist of a non-empty sequence of letters, digits, hyphen, underscore, period, or colon.

Each exchange belongs to one of a set of exchange kinds/types implemented by the server. The exchange types define the functionality of the exchange - i.e. how messages are routed through it. Once an exchange is declared, its type cannot be changed. The common types are "direct", "fanout", "topic" and "headers".

func (*Session) ExchangeDeclarePassive 

func (s *Session) ExchangeDeclarePassive(ctx context.Context, name string, kind ExchangeKind, option ...ExchangeDeclareOptions) error

ExchangeDeclarePassive is functionally and parametrically equivalent to ExchangeDeclare, except that it sets the "passive" attribute to true. A passive exchange is assumed by RabbitMQ to already exist, and attempting to connect to a non-existent exchange will cause RabbitMQ to throw an exception. This function can be used to detect the existence of an exchange.

func (*Session) ExchangeDelete 

func (s *Session) ExchangeDelete(ctx context.Context, name string, option ...ExchangeDeleteOptions) error

ExchangeDelete removes the named exchange from the server. When an exchange is deleted all queue bindings on the exchange are also deleted. If this exchange does not exist, the channel will be closed with an error.

func (*Session) ExchangeUnbind 

func (s *Session) ExchangeUnbind(ctx context.Context, destination string, routingKey string, source string, option ...ExchangeUnbindOptions) error

ExchangeUnbind unbinds the destination exchange from the source exchange on the server by removing the routing key between them. This is the inverse of ExchangeBind. If the binding does not currently exist, an error will be returned.

func (*Session) Flag 

func (s *Session) Flag(err error)

Flag marks the session as flagged. This is useful in case of a connection pool, where the session is returned to the pool and should be recovered by the next user.

func (*Session) Flow 

func (s *Session) Flow(ctx context.Context, active bool) error

Flow allows to enable or disable flow from the message broker Flow pauses the delivery of messages to consumers on this channel. Channels are opened with flow control active, to open a channel with paused deliveries immediately call this method with `false` after calling Connection.Channel.

When active is `false`, this method asks the server to temporarily pause deliveries until called again with active as `true`.

Channel.Get methods will not be affected by flow control.

This method is not intended to act as window control. Use Channel.Qos to limit the number of unacknowledged messages or bytes in flight instead.

The server may also send us flow methods to throttle our publishings. A well behaving publishing client should add a listener with Channel.NotifyFlow and pause its publishings when `false` is sent on that channel.

Note: RabbitMQ prefers to use TCP push back to control flow for all channels on a connection, so under high volume scenarios, it's wise to open separate Connections for publishings and deliveries.

func (*Session) FlushConfirms 

func (s *Session) FlushConfirms()

Flush confirms channel

func (*Session) FlushReturned 

func (s *Session) FlushReturned()

FlushReturned publish channel

func (*Session) Get 

func (s *Session) Get(ctx context.Context, queue string, autoAck bool) (msg Delivery, ok bool, err error)

Get is only supposed to be used for testing purposes, do not us eit to poll the queue periodically.

func (*Session) IsCached 

func (s *Session) IsCached() bool

IsCached returns true in case this session is supposed to be returned to a session pool.

func (*Session) IsConfirmable 

func (s *Session) IsConfirmable() bool

IsConfirmable returns true in case this session requires that after Publishing a message you also MUST Await its confirmation

func (*Session) IsFlagged 

func (s *Session) IsFlagged() bool

IsFlagged returns whether the session is flagged.

func (*Session) Nack 

func (s *Session) Nack(deliveryTag uint64, multiple bool, requeue bool) (err error)

Nack rejects the message. In case the underlying channel dies, you cannot send a nack for the processed message. You might receive the message again from the broker, as it expects a n/ack

func (*Session) Name 

func (s *Session) Name() string

func (*Session) Publish 

func (s *Session) Publish(ctx context.Context, exchange string, routingKey string, msg Publishing) (deliveryTag uint64, err error)

Publish sends a Publishing from the client to an exchange on the server. When you want a single message to be delivered to a single queue, you can publish to the default exchange with the routingKey of the queue name. This is because every declared queue gets an implicit route to the default exchange. It is possible for publishing to not reach the broker if the underlying socket is shut down without pending publishing packets being flushed from the kernel buffers. The easy way of making it probable that all publishings reach the server is to always call Connection.Close before terminating your publishing application. The way to ensure that all publishings reach the server is to add a listener to Channel.NotifyPublish and put the channel in confirm mode with Channel.Confirm. Publishing delivery tags and their corresponding confirmations start at 1. Exit when all publishings are confirmed. When Publish does not return an error and the channel is in confirm mode, the internal counter for DeliveryTags with the first confirmation starts at 1.

func (*Session) Qos 

func (s *Session) Qos(ctx context.Context, prefetchCount int, prefetchSize int) error

Qos controls how many messages or how many bytes the server will try to keep on the network for consumers before receiving delivery acks. The intent of Qos is to make sure the network buffers stay full between the server and client.

With a prefetch count greater than zero, the server will deliver that many messages to consumers before acknowledgments are received. The server ignores this option when consumers are started with noAck because no acknowledgments are expected or sent.

With a prefetch size greater than zero, the server will try to keep at least that many bytes of deliveries flushed to the network before receiving acknowledgments from the consumers. This option is ignored when consumers are started with noAck.

To get round-robin behavior between consumers consuming from the same queue on different connections, set the prefetch count to 1, and the next available message on the server will be delivered to the next available consumer.

If your consumer work time is reasonably consistent and not much greater than two times your network round trip time, you will see significant throughput improvements starting with a prefetch count of 2 or slightly greater as described by benchmarks on RabbitMQ.

http://www.rabbitmq.com/blog/2012/04/25/rabbitmq-performance-measurements-part-2/

func (*Session) QueueBind 

func (s *Session) QueueBind(ctx context.Context, queueName string, routingKey string, exchange string, option ...QueueBindOptions) error

QueueBind binds an exchange to a queue so that publishings to the exchange will be routed to the queue when the publishing routing key matches the binding routing key. 

QueueBind("pagers", "alert", "log", false, nil)
QueueBind("emails", "info", "log", false, nil)

Delivery       Exchange  Key       Queue
-----------------------------------------------
key: alert --> log ----> alert --> pagers
key: info ---> log ----> info ---> emails
key: debug --> log       (none)    (dropped)

If a binding with the same key and arguments already exists between the exchange and queue, the attempt to rebind will be ignored and the existing binding will be retained.

In the case that multiple bindings may cause the message to be routed to the same queue, the server will only route the publishing once. This is possible with topic exchanges. 

QueueBind("pagers", "alert", "amq.topic", false, nil)
QueueBind("emails", "info", "amq.topic", false, nil)
QueueBind("emails", "#", "amq.topic", false, nil) // match everything

Delivery       Exchange        Key       Queue
-----------------------------------------------
key: alert --> amq.topic ----> alert --> pagers
key: info ---> amq.topic ----> # ------> emails
                         \---> info ---/
key: debug --> amq.topic ----> # ------> emails

It is only possible to bind a durable queue to a durable exchange regardless of whether the queue or exchange is auto-deleted. Bindings between durable queues and exchanges will also be restored on server restart.

If the binding could not complete, an error will be returned and the channel will be closed.

func (*Session) QueueDeclare 

func (s *Session) QueueDeclare(ctx context.Context, name string, option ...QueueDeclareOptions) (Queue, error)

QueueDeclare declares a queue to hold messages and deliver to consumers. Declaring creates a queue if it doesn't already exist, or ensures that an existing queue matches the same parameters.

Every queue declared gets a default binding to the empty exchange "" which has the type "direct" with the routing key matching the queue's name. With this default binding, it is possible to publish messages that route directly to this queue by publishing to "" with the routing key of the queue name. 

QueueDeclare("alerts", true, false, false, false, nil)
Publish("", "alerts", false, false, Publishing{Body: []byte("...")})

Delivery       Exchange  Key       Queue
-----------------------------------------------
key: alerts -> ""     -> alerts -> alerts

The queue name may be empty, in which case the server will generate a unique name which will be returned in the Name field of Queue struct.

When the error return value is not nil, you can assume the queue could not be declared with these parameters, and the channel will be closed.

func (*Session) QueueDeclarePassive 

func (s *Session) QueueDeclarePassive(ctx context.Context, name string, option ...QueueDeclareOptions) (Queue, error)

QueueDeclarePassive is functionally and parametrically equivalent to QueueDeclare, except that it sets the "passive" attribute to true. 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. This function can be used to test for the existence of a queue.

func (*Session) QueueDelete 

func (s *Session) QueueDelete(ctx context.Context, name string, option ...QueueDeleteOptions) (purgedMsgs int, err error)

QueueDelete removes the queue from the server including all bindings then purges the messages based on server configuration, returning the number of messages purged.

func (*Session) QueuePurge 

func (s *Session) QueuePurge(ctx context.Context, name string, options ...QueuePurgeOptions) (int, error)

QueuePurge removes all messages from the named queue which are not waiting to be acknowledged. Messages that have been delivered but have not yet been acknowledged will not be removed. When successful, returns the number of messages purged.

func (*Session) QueueUnbind 

func (s *Session) QueueUnbind(ctx context.Context, name string, routingKey string, exchange string, arg ...Table) error

QueueUnbind removes a binding between an exchange and queue matching the key and arguments. It is possible to send and empty string for the exchange name which means to unbind the queue from the default exchange.

func (*Session) Recover 

func (s *Session) Recover(ctx context.Context) error

func (*Session) Tx 

func (s *Session) Tx() error

Tx puts the channel into transaction mode on the server. All publishings and acknowledgments following this method will be atomically committed or rolled back for a single queue. Call either Channel.TxCommit or Channel.TxRollback to leave this transaction and immediately start a new transaction.

The atomicity across multiple queues is not defined as queue declarations and bindings are not included in the transaction.

The behavior of publishings that are delivered as mandatory or immediate while the channel is in a transaction is not defined.

Once a channel has been put into transaction mode, it cannot be taken out of transaction mode. Use a different channel for non-transactional semantics.

func (*Session) TxCommit 

func (s *Session) TxCommit() error

TxCommit atomically commits all publishings and acknowledgments for a single queue and immediately start a new transaction.

Calling this method without having called Channel.Tx is an error.

func (*Session) TxRollback 

func (s *Session) TxRollback() error

TxRollback atomically rolls back all publishings and acknowledgments for a single queue and immediately start a new transaction.

Calling this method without having called Channel.Tx is an error.

type SessionOption 

type SessionOption func(*sessionOption)

func SessionWithAutoCloseConnection 

func SessionWithAutoCloseConnection(autoClose bool) SessionOption

SessionWithAutoCloseConnection is important for transient sessions which, as they allow to create sessions that close their internal connections automatically upon closing themselves.

func SessionWithBufferCapacity 

func SessionWithBufferCapacity(capacity int) SessionOption

SessionWithBufferSize allows to customize the size of th einternal channel buffers. all buffers/channels are initialized with this size. (e.g. error or confirm channels)

func SessionWithCached 

func SessionWithCached(cached bool) SessionOption

SessionWithCached makes a session a cached session This is only necessary for the session pool, as cached sessions are part of a pool and can be returned back to the pool without being closed.

func SessionWithConfirms 

func SessionWithConfirms(requiresPublishConfirms bool) SessionOption

SessionContext allows enable or explicitly disable message acknowledgements (acks)

func SessionWithConsumeContextRetryCallback 

func SessionWithConsumeContextRetryCallback(callback SessionRetryCallback) SessionOption

SessionWithConsumeContextRetryCallback allows to set a custom consume retry callback. The callback should not interact with anything that may lead to any kind of errors. It should preferrably delegate its work to a separate goroutine.

func SessionWithContext 

func SessionWithContext(ctx context.Context) SessionOption

SessionWithContext allows to set a custom session context that might trigger a shutdown

func SessionWithExchangeBindRetryCallback 

func SessionWithExchangeBindRetryCallback(callback SessionRetryCallback) SessionOption

SessionWithExchangeBindRetryCallback allows to set a custom exchange bind retry callback. The callback should not interact with anything that may lead to any kind of errors. It should preferrably delegate its work to a separate goroutine.

func SessionWithExchangeDeclarePassiveRetryCallback 

func SessionWithExchangeDeclarePassiveRetryCallback(callback SessionRetryCallback) SessionOption

SessionWithExchangeDeclarePassiveRetryCallback allows to set a custom exchange declare passive retry callback. The callback should not interact with anything that may lead to any kind of errors. It should preferrably delegate its work to a separate goroutine.

func SessionWithExchangeDeclareRetryCallback 

func SessionWithExchangeDeclareRetryCallback(callback SessionRetryCallback) SessionOption

SessionWithExchangeDeclareRetryCallback allows to set a custom exchange declare retry callback. The callback should not interact with anything that may lead to any kind of errors. It should preferrably delegate its work to a separate goroutine.

func SessionWithExchangeDeleteRetryCallback 

func SessionWithExchangeDeleteRetryCallback(callback SessionRetryCallback) SessionOption

SessionWithExchangeDeleteRetryCallback allows to set a custom exchange delete retry callback. The callback should not interact with anything that may lead to any kind of errors. It should preferrably delegate its work to a separate goroutine.

func SessionWithExchangeUnbindRetryCallback 

func SessionWithExchangeUnbindRetryCallback(callback SessionRetryCallback) SessionOption

SessionWithExchangeUnbindRetryCallback allows to set a custom exchange unbind retry callback. The callback should not interact with anything that may lead to any kind of errors. It should preferrably delegate its work to a separate goroutine.

func SessionWithFlowRetryCallback 

func SessionWithFlowRetryCallback(callback SessionRetryCallback) SessionOption

SessionWithFlowRetryCallback allows to set a custom flow retry callback. The callback should not interact with anything that may lead to any kind of errors. It should preferrably delegate its work to a separate goroutine.

func SessionWithGetRetryCallback 

func SessionWithGetRetryCallback(callback SessionRetryCallback) SessionOption

SessionWithGetRetryCallback allows to set a custom get retry callback. The callback should not interact with anything that may lead to any kind of errors. It should preferrably delegate its work to a separate goroutine.

func SessionWithLogger 

func SessionWithLogger(logger *slog.Logger) SessionOption

SessionWithLogger allows to set a logger. By default no logger is set.

func SessionWithPublishRetryCallback 

func SessionWithPublishRetryCallback(callback SessionRetryCallback) SessionOption

SessionWithPublishRetryCallback allows to set a custom publish retry callback. The callback should not interact with anything that may lead to any kind of errors. It should preferrably delegate its work to a separate goroutine.

func SessionWithQoSRetryCallback 

func SessionWithQoSRetryCallback(callback SessionRetryCallback) SessionOption

SessionWithQoSRetryCallback allows to set a custom qos retry callback. The callback should not interact with anything that may lead to any kind of errors. It should preferrably delegate its work to a separate goroutine.

func SessionWithQueueBindRetryCallback 

func SessionWithQueueBindRetryCallback(callback SessionRetryCallback) SessionOption

SessionWithQueueBindRetryCallback allows to set a custom queue bind retry callback. The callback should not interact with anything that may lead to any kind of errors. It should preferrably delegate its work to a separate goroutine.

func SessionWithQueueDeclarePassiveRetryCallback 

func SessionWithQueueDeclarePassiveRetryCallback(callback SessionRetryCallback) SessionOption

SessionWithQueueDeclarePassiveRetryCallback allows to set a custom queue declare passive retry callback. The callback should not interact with anything that may lead to any kind of errors. It should preferrably delegate its work to a separate goroutine.

func SessionWithQueueDeclareRetryCallback 

func SessionWithQueueDeclareRetryCallback(callback SessionRetryCallback) SessionOption

SessionWithQueueDeclareRetryCallback allows to set a custom queue declare retry callback. The callback should not interact with anything that may lead to any kind of errors. It should preferrably delegate its work to a separate goroutine.

func SessionWithQueueDeleteRetryCallback 

func SessionWithQueueDeleteRetryCallback(callback SessionRetryCallback) SessionOption

SessionWithQueueDeleteRetryCallback allows to set a custom queue delete retry callback. The callback should not interact with anything that may lead to any kind of errors. It should preferrably delegate its work to a separate goroutine.

func SessionWithQueuePurgeRetryCallback 

func SessionWithQueuePurgeRetryCallback(callback SessionRetryCallback) SessionOption

SessionWithQueuePurgeRetryCallback allows to set a custom queue purge retry callback. The callback should not interact with anything that may lead to any kind of errors. It should preferrably delegate its work to a separate goroutine.

func SessionWithQueueUnbindRetryCallback 

func SessionWithQueueUnbindRetryCallback(callback SessionRetryCallback) SessionOption

SessionWithQueueUnbindRetryCallback allows to set a custom queue unbind retry callback. The callback should not interact with anything that may lead to any kind of errors. It should preferrably delegate its work to a separate goroutine.

func SessionWithRecoverCallback 

func SessionWithRecoverCallback(callback SessionRetryCallback) SessionOption

SessionWithRecoverCallback allows to set a custom recover callback. The callback should not interact with anything that may lead to any kind of errors. It should preferrably delegate its work to a separate goroutine.

func SessionWithRetryCallback 

func SessionWithRetryCallback(callback SessionRetryCallback) SessionOption

SessionWithRetryCallback allows to set a custom retry callback for all operations. The callback should not interact with anything that may lead to any kind of errors. It should preferrably delegate its work to a separate goroutine.

type SessionRetryCallback 

type SessionRetryCallback func(operation, connName, sessionName string, retry int, err error)

RetryCallback is a function that is called when some operation fails.

type Table 

type Table amqp091.Table
Table is a dynamic map of arguments that may be passed additionally to functions.

type Table map[string]interface{}

Table stores user supplied fields of the following types: 

bool
byte
int8
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.

func NewTable 

func NewTable() Table

func (Table) Clone 

func (t Table) Clone() Table

func (Table) Death 

func (t Table) Death() (int64, bool)

func (Table) DeliveryCount 

func (t Table) DeliveryCount() (int64, bool)

Returns the number of deliveries of the message.

func (Table) WithDeadLetterExchange 

func (t Table) WithDeadLetterExchange(exchange string) Table

Rejected messages will be routed to the dead-letter exchange. Which in turn routes them to some specified queue.

func (Table) WithDeadLetterExchangeAndRoutingKey 

func (t Table) WithDeadLetterExchangeAndRoutingKey(exchange, routingKey string) Table

Rejected messages will be routed to the dead-letter exchange. Which in turn routes them to using the specified routing key.

func (Table) WithDeliveryLimit 

func (t Table) WithDeliveryLimit(limit int) Table

This option is interesting for quorum queues. It specifies the number f redeliveries that still presereve the order of the messages. After that the order is not quaranteed anymore.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.