amqp091

README

Go RabbitMQ Client Library

This is a Go AMQP 0.9.1 client maintained by the RabbitMQ core team. It was originally developed by Sean Treadway.

Differences from streadway/amqp

Some things are different compared to the original client, others haven't changed.

Package Name

This library uses a different package name. If moving from streadway/amqp, using an alias may reduce the number of changes needed:

amqp "github.com/rabbitmq/amqp091-go"

License

This client uses the same 2-clause BSD license as the original project.

Public API Evolution

This client retains key API elements as practically possible. It is, however, open to reasonable breaking public API changes suggested by the community. We don't have the "no breaking public API changes ever" rule and fully recognize that a good client API evolves over time.

Project Maturity

This project is based on a mature Go client that's been around for over a decade.

Supported Go Versions

This client supports two most recent Go release series.

Supported RabbitMQ Versions

This project supports RabbitMQ versions starting with 2.0 but primarily tested against currently supported RabbitMQ release series.

Some features and behaviours may be server version-specific.

Goals

Provide a functional interface that closely represents the AMQP 0.9.1 model targeted to RabbitMQ as a server. This includes the minimum necessary to interact the semantics of the protocol.

Non-goals

Things not intended to be supported.

• Auto reconnect and re-synchronization of client and server topologies.
  • Reconnection would require understanding the error paths when the topology cannot be declared on reconnect. This would require a new set of types and code paths that are best suited at the call-site of this package. AMQP has a dynamic topology that needs all peers to agree. If this doesn't happen, the behavior is undefined. Instead of producing a possible interface with undefined behavior, this package is designed to be simple for the caller to implement the necessary connection-time topology declaration so that reconnection is trivial and encapsulated in the caller's application code.
• AMQP Protocol negotiation for forward or backward compatibility.
  • 0.9.1 is stable and widely deployed. AMQP 1.0 is a divergent specification (a different protocol) and belongs to a different library.
• Anything other than PLAIN and EXTERNAL authentication mechanisms.
  • Keeping the mechanisms interface modular makes it possible to extend outside of this package. If other mechanisms prove to be popular, then we would accept patches to include them in this package.
• Support for basic.return and basic.ack frame ordering. This client uses Go channels for certain protocol events and ordering between events sent to two different channels generally cannot be guaranteed.

Usage

See the _examples subdirectory for simple producers and consumers executables. If you have a use-case in mind which isn't well-represented by the examples, please file an issue.

Documentation

• Godoc API reference
• RabbitMQ tutorials in Go

Contributing

Pull requests are very much welcomed. Create your pull request on a non-main branch, make sure a test or example is included that covers your change, and your commits represent coherent changes that include a reason for the change.

See CONTRIBUTING.md for more information.

License

BSD 2 clause, see LICENSE for more details.

Documentation

Overview

Package amqp091 is an AMQP 0.9.1 client with RabbitMQ extensions

Understand the AMQP 0.9.1 messaging model by reviewing these links first. Much of the terminology in this library directly relates to AMQP concepts.

Resources

http://www.rabbitmq.com/tutorials/amqp-concepts.html
http://www.rabbitmq.com/getstarted.html
http://www.rabbitmq.com/amqp-0-9-1-reference.html

Design

Most other broker clients publish to queues, but in AMQP, clients publish Exchanges instead. AMQP is programmable, meaning that both the producers and consumers agree on the configuration of the broker, instead of requiring an operator or system configuration that declares the logical topology in the broker. The routing between producers and consumer queues is via Bindings. These bindings form the logical topology of the broker.

In this library, a message sent from publisher is called a "Publishing" and a message received to a consumer is called a "Delivery". The fields of Publishings and Deliveries are close but not exact mappings to the underlying wire format to maintain stronger types. Many other libraries will combine message properties with message headers. In this library, the message well known properties are strongly typed fields on the Publishings and Deliveries, whereas the user defined headers are in the Headers field.

The method naming closely matches the protocol's method name with positional parameters mapping to named protocol message fields. The motivation here is to present a comprehensive view over all possible interactions with the server.

Generally, methods that map to protocol methods of the "basic" class will be elided in this interface, and "select" methods of various channel mode selectors will be elided for example Channel.Confirm and Channel.Tx.

The library is intentionally designed to be synchronous, where responses for each protocol message are required to be received in an RPC manner. Some methods have a noWait parameter like Channel.QueueDeclare, and some methods are asynchronous like Channel.Publish. The error values should still be checked for these methods as they will indicate IO failures like when the underlying connection closes.

Asynchronous Events

Clients of this library may be interested in receiving some of the protocol messages other than Deliveries like basic.ack methods while a channel is in confirm mode.

The Notify* methods with Connection and Channel receivers model the pattern of asynchronous events like closes due to exceptions, or messages that are sent out of band from an RPC call like basic.ack or basic.flow.

Any asynchronous events, including Deliveries and Publishings must always have a receiver until the corresponding chans are closed. Without asynchronous receivers, the synchronous methods will block.

Use Case

It's important as a client to an AMQP topology to ensure the state of the broker matches your expectations. For both publish and consume use cases, make sure you declare the queues, exchanges and bindings you expect to exist prior to calling Channel.PublishWithContext or Channel.Consume.

// Connections start with amqp.Dial() typically from a command line argument
// or environment variable.
connection, err := amqp.Dial(os.Getenv("AMQP_URL"))

// To cleanly shutdown by flushing kernel buffers, make sure to close and
// wait for the response.
defer connection.Close()

// Most operations happen on a channel.  If any error is returned on a
// channel, the channel will no longer be valid, throw it away and try with
// a different channel.  If you use many channels, it's useful for the
// server to
channel, err := connection.Channel()

// Declare your topology here, if it doesn't exist, it will be created, if
// it existed already and is not what you expect, then that's considered an
// error.

// Use your connection on this topology with either Publish or Consume, or
// inspect your queues with QueueInspect.  It's unwise to mix Publish and
// Consume to let TCP do its job well.

SSL/TLS - Secure connections

When Dial encounters an amqps:// scheme, it will use the zero value of a tls.Config. This will only perform server certificate and host verification.

Use DialTLS when you wish to provide a client certificate (recommended), include a private certificate authority's certificate in the cert chain for server validity, or run insecure by not verifying the server certificate. DialTLS will use the provided tls.Config when it encounters an amqps:// scheme and will dial a plain connection when it encounters an amqp:// scheme.

SSL/TLS in RabbitMQ is documented here: http://www.rabbitmq.com/ssl.html

Best practises for Connection and Channel notifications:

In order to be notified when a connection or channel gets closed, both structures offer the possibility to register channels using Channel.NotifyClose and Connection.NotifyClose functions:

notifyConnCloseCh := conn.NotifyClose(make(chan *amqp.Error, 1))

No errors will be sent in case of a graceful connection close. In case of a non-graceful closure due to e.g. network issue, or forced connection closure from the Management UI, the error will be notified synchronously by the library.

The library sends to notification channels just once. After sending a notification to all channels, the library closes all registered notification channels. After receiving a notification, the application should create and register a new channel. To avoid deadlocks in the library, it is necessary to consume from the channels. This could be done inside a different goroutine with a select listening on the two channels inside a for loop like:

go func() {
  for notifyConnClose != nil || notifyChanClose != nil {
    select {
      case err, ok := <-notifyConnClose:
        if !ok {
          notifyConnClose = nil
        } else {
          fmt.Printf("connection closed, error %s", err)
        }
      case err, ok := <-notifyChanClose:
        if !ok {
          notifyChanClose = nil
        } else {
          fmt.Printf("channel closed, error %s", err)
        }
      }
  }
}()

It is strongly recommended to use buffered channels to avoid deadlocks inside the library.

Best practises for NotifyPublish notifications:

Using Channel.NotifyPublish allows the caller of the library to be notified, through a go channel, when a message has been received and confirmed by the broker. It's advisable to wait for all Confirmations to arrive before calling Channel.Close or Connection.Close. It is also necessary to consume from this channel until it gets closed. The library sends synchronously to the registered channel. It is advisable to use a buffered channel, with capacity set to the maximum acceptable number of unconfirmed messages.

It is important to consume from the confirmation channel at all times, in order to avoid deadlocks in the library.

Example (Consume)

package main

import (
	"context"
	"errors"
	"log"
	"os"
	"sync"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
	queueName := "job_queue"
	addr := "amqp://guest:guest@localhost:5672/"
	queue := New(queueName, addr)

	// Give the connection sometime to set up
	<-time.After(time.Second)

	ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
	defer cancel()

	deliveries, err := queue.Consume()
	if err != nil {
		log.Printf("Could not start consuming: %s\n", err)
		return
	}

	// This channel will receive a notification when a channel closed event
	// happens. This must be different from Client.notifyChanClose because the
	// library sends only one notification and Client.notifyChanClose already has
	// a receiver in handleReconnect().
	// Recommended to make it buffered to avoid deadlocks
	chClosedCh := make(chan *amqp.Error, 1)
	queue.channel.NotifyClose(chClosedCh)

	for {
		select {
		case <-ctx.Done():
			err := queue.Close()
			if err != nil {
				log.Printf("Close failed: %s\n", err)
			}
			return

		case amqErr := <-chClosedCh:
			// This case handles the event of closed channel e.g. abnormal shutdown
			log.Printf("AMQP Channel closed due to: %s\n", amqErr)

			deliveries, err = queue.Consume()
			if err != nil {
				// If the AMQP channel is not ready, it will continue the loop. Next
				// iteration will enter this case because chClosedCh is closed by the
				// library
				log.Println("Error trying to consume, will try again")
				continue
			}

			// Re-set channel to receive notifications
			// The library closes this channel after abnormal shutdown
			chClosedCh = make(chan *amqp.Error, 1)
			queue.channel.NotifyClose(chClosedCh)

		case delivery := <-deliveries:
			// Ack a message every 2 seconds
			log.Printf("Received message: %s\n", delivery.Body)
			if err := delivery.Ack(false); err != nil {
				log.Printf("Error acknowledging message: %s\n", err)
			}
			<-time.After(time.Second * 2)
		}
	}
}

// Client is the base struct for handling connection recovery, consumption and
// publishing. Note that this struct has an internal mutex to safeguard against
// data races. As you develop and iterate over this example, you may need to add
// further locks, or safeguards, to keep your application safe from data races
type Client struct {
	m               *sync.Mutex
	queueName       string
	logger          *log.Logger
	connection      *amqp.Connection
	channel         *amqp.Channel
	done            chan bool
	notifyConnClose chan *amqp.Error
	notifyChanClose chan *amqp.Error
	notifyConfirm   chan amqp.Confirmation
	isReady         bool
}

const (
	reconnectDelay = 5 * time.Second

	reInitDelay = 2 * time.Second

	resendDelay = 5 * time.Second
)

var (
	errNotConnected  = errors.New("not connected to a server")
	errAlreadyClosed = errors.New("already closed: not connected to the server")
	errShutdown      = errors.New("client is shutting down")
)

// New creates a new consumer state instance, and automatically
// attempts to connect to the server.
func New(queueName, addr string) *Client {
	client := Client{
		m:         &sync.Mutex{},
		logger:    log.New(os.Stdout, "", log.LstdFlags),
		queueName: queueName,
		done:      make(chan bool),
	}
	go client.handleReconnect(addr)
	return &client
}

// handleReconnect will wait for a connection error on
// notifyConnClose, and then continuously attempt to reconnect.
func (client *Client) handleReconnect(addr string) {
	for {
		client.m.Lock()
		client.isReady = false
		client.m.Unlock()

		client.logger.Println("Attempting to connect")

		conn, err := client.connect(addr)

		if err != nil {
			client.logger.Println("Failed to connect. Retrying...")

			select {
			case <-client.done:
				return
			case <-time.After(reconnectDelay):
			}
			continue
		}

		if done := client.handleReInit(conn); done {
			break
		}
	}
}

// connect will create a new AMQP connection
func (client *Client) connect(addr string) (*amqp.Connection, error) {
	conn, err := amqp.Dial(addr)

	if err != nil {
		return nil, err
	}

	client.changeConnection(conn)
	client.logger.Println("Connected!")
	return conn, nil
}

// handleReInit will wait for a channel error
// and then continuously attempt to re-initialize both channels
func (client *Client) handleReInit(conn *amqp.Connection) bool {
	for {
		client.m.Lock()
		client.isReady = false
		client.m.Unlock()

		err := client.init(conn)

		if err != nil {
			client.logger.Println("Failed to initialize channel. Retrying...")

			select {
			case <-client.done:
				return true
			case <-client.notifyConnClose:
				client.logger.Println("Connection closed. Reconnecting...")
				return false
			case <-time.After(reInitDelay):
			}
			continue
		}

		select {
		case <-client.done:
			return true
		case <-client.notifyConnClose:
			client.logger.Println("Connection closed. Reconnecting...")
			return false
		case <-client.notifyChanClose:
			client.logger.Println("Channel closed. Re-running init...")
		}
	}
}

// init will initialize channel & declare queue
func (client *Client) init(conn *amqp.Connection) error {
	ch, err := conn.Channel()

	if err != nil {
		return err
	}

	err = ch.Confirm(false)

	if err != nil {
		return err
	}
	_, err = ch.QueueDeclare(
		client.queueName,
		false,
		false,
		false,
		false,
		nil,
	)

	if err != nil {
		return err
	}

	client.changeChannel(ch)
	client.m.Lock()
	client.isReady = true
	client.m.Unlock()
	client.logger.Println("Setup!")

	return nil
}

// changeConnection takes a new connection to the queue,
// and updates the close listener to reflect this.
func (client *Client) changeConnection(connection *amqp.Connection) {
	client.connection = connection
	client.notifyConnClose = make(chan *amqp.Error, 1)
	client.connection.NotifyClose(client.notifyConnClose)
}

// changeChannel takes a new channel to the queue,
// and updates the channel listeners to reflect this.
func (client *Client) changeChannel(channel *amqp.Channel) {
	client.channel = channel
	client.notifyChanClose = make(chan *amqp.Error, 1)
	client.notifyConfirm = make(chan amqp.Confirmation, 1)
	client.channel.NotifyClose(client.notifyChanClose)
	client.channel.NotifyPublish(client.notifyConfirm)
}

// Push will push data onto the queue, and wait for a confirmation.
// This will block until the server sends a confirmation. Errors are
// only returned if the push action itself fails, see UnsafePush.
func (client *Client) Push(data []byte) error {
	client.m.Lock()
	if !client.isReady {
		client.m.Unlock()
		return errors.New("failed to push: not connected")
	}
	client.m.Unlock()
	for {
		err := client.UnsafePush(data)
		if err != nil {
			client.logger.Println("Push failed. Retrying...")
			select {
			case <-client.done:
				return errShutdown
			case <-time.After(resendDelay):
			}
			continue
		}
		confirm := <-client.notifyConfirm
		if confirm.Ack {
			client.logger.Printf("Push confirmed [%d]!", confirm.DeliveryTag)
			return nil
		}
	}
}

// UnsafePush will push to the queue without checking for
// confirmation. It returns an error if it fails to connect.
// No guarantees are provided for whether the server will
// receive the message.
func (client *Client) UnsafePush(data []byte) error {
	client.m.Lock()
	if !client.isReady {
		client.m.Unlock()
		return errNotConnected
	}
	client.m.Unlock()

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	return client.channel.PublishWithContext(
		ctx,
		"",
		client.queueName,
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        data,
		},
	)
}

// Consume will continuously put queue items on the channel.
// It is required to call delivery.Ack when it has been
// successfully processed, or delivery.Nack when it fails.
// Ignoring this will cause data to build up on the server.
func (client *Client) Consume() (<-chan amqp.Delivery, error) {
	client.m.Lock()
	if !client.isReady {
		client.m.Unlock()
		return nil, errNotConnected
	}
	client.m.Unlock()

	if err := client.channel.Qos(
		1,
		0,
		false,
	); err != nil {
		return nil, err
	}

	return client.channel.Consume(
		client.queueName,
		"",
		false,
		false,
		false,
		false,
		nil,
	)
}

// Close will cleanly shut down the channel and connection.
func (client *Client) Close() error {
	client.m.Lock()

	defer client.m.Unlock()

	if !client.isReady {
		return errAlreadyClosed
	}
	close(client.done)
	err := client.channel.Close()
	if err != nil {
		return err
	}
	err = client.connection.Close()
	if err != nil {
		return err
	}

	client.isReady = false
	return nil
}

Example (Publish)

This exports a Client object that wraps this library. It automatically reconnects when the connection fails, and blocks all pushes until the connection succeeds. It also confirms every outgoing message, so none are lost. It doesn't automatically ack each message, but leaves that to the parent process, since it is usage-dependent.

Try running this in one terminal, and rabbitmq-server in another.

Stop & restart RabbitMQ to see how the queue reacts.

package main

import (
	"context"
	"errors"
	"log"
	"os"
	"sync"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
	queueName := "job_queue"
	addr := "amqp://guest:guest@localhost:5672/"
	queue := New(queueName, addr)
	message := []byte("message")

	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Second*20))
	defer cancel()
loop:
	for {
		select {
		// Attempt to push a message every 2 seconds
		case <-time.After(time.Second * 2):
			if err := queue.Push(message); err != nil {
				log.Printf("Push failed: %s\n", err)
			} else {
				log.Println("Push succeeded!")
			}
		case <-ctx.Done():
			if err := queue.Close(); err != nil {
				log.Printf("Close failed: %s\n", err)
			}
			break loop
		}
	}
}

// Client is the base struct for handling connection recovery, consumption and
// publishing. Note that this struct has an internal mutex to safeguard against
// data races. As you develop and iterate over this example, you may need to add
// further locks, or safeguards, to keep your application safe from data races
type Client struct {
	m               *sync.Mutex
	queueName       string
	logger          *log.Logger
	connection      *amqp.Connection
	channel         *amqp.Channel
	done            chan bool
	notifyConnClose chan *amqp.Error
	notifyChanClose chan *amqp.Error
	notifyConfirm   chan amqp.Confirmation
	isReady         bool
}

const (
	reconnectDelay = 5 * time.Second

	reInitDelay = 2 * time.Second

	resendDelay = 5 * time.Second
)

var (
	errNotConnected  = errors.New("not connected to a server")
	errAlreadyClosed = errors.New("already closed: not connected to the server")
	errShutdown      = errors.New("client is shutting down")
)

// New creates a new consumer state instance, and automatically
// attempts to connect to the server.
func New(queueName, addr string) *Client {
	client := Client{
		m:         &sync.Mutex{},
		logger:    log.New(os.Stdout, "", log.LstdFlags),
		queueName: queueName,
		done:      make(chan bool),
	}
	go client.handleReconnect(addr)
	return &client
}

// handleReconnect will wait for a connection error on
// notifyConnClose, and then continuously attempt to reconnect.
func (client *Client) handleReconnect(addr string) {
	for {
		client.m.Lock()
		client.isReady = false
		client.m.Unlock()

		client.logger.Println("Attempting to connect")

		conn, err := client.connect(addr)

		if err != nil {
			client.logger.Println("Failed to connect. Retrying...")

			select {
			case <-client.done:
				return
			case <-time.After(reconnectDelay):
			}
			continue
		}

		if done := client.handleReInit(conn); done {
			break
		}
	}
}

// connect will create a new AMQP connection
func (client *Client) connect(addr string) (*amqp.Connection, error) {
	conn, err := amqp.Dial(addr)

	if err != nil {
		return nil, err
	}

	client.changeConnection(conn)
	client.logger.Println("Connected!")
	return conn, nil
}

// handleReInit will wait for a channel error
// and then continuously attempt to re-initialize both channels
func (client *Client) handleReInit(conn *amqp.Connection) bool {
	for {
		client.m.Lock()
		client.isReady = false
		client.m.Unlock()

		err := client.init(conn)

		if err != nil {
			client.logger.Println("Failed to initialize channel. Retrying...")

			select {
			case <-client.done:
				return true
			case <-client.notifyConnClose:
				client.logger.Println("Connection closed. Reconnecting...")
				return false
			case <-time.After(reInitDelay):
			}
			continue
		}

		select {
		case <-client.done:
			return true
		case <-client.notifyConnClose:
			client.logger.Println("Connection closed. Reconnecting...")
			return false
		case <-client.notifyChanClose:
			client.logger.Println("Channel closed. Re-running init...")
		}
	}
}

// init will initialize channel & declare queue
func (client *Client) init(conn *amqp.Connection) error {
	ch, err := conn.Channel()

	if err != nil {
		return err
	}

	err = ch.Confirm(false)

	if err != nil {
		return err
	}
	_, err = ch.QueueDeclare(
		client.queueName,
		false,
		false,
		false,
		false,
		nil,
	)

	if err != nil {
		return err
	}

	client.changeChannel(ch)
	client.m.Lock()
	client.isReady = true
	client.m.Unlock()
	client.logger.Println("Setup!")

	return nil
}

// changeConnection takes a new connection to the queue,
// and updates the close listener to reflect this.
func (client *Client) changeConnection(connection *amqp.Connection) {
	client.connection = connection
	client.notifyConnClose = make(chan *amqp.Error, 1)
	client.connection.NotifyClose(client.notifyConnClose)
}

// changeChannel takes a new channel to the queue,
// and updates the channel listeners to reflect this.
func (client *Client) changeChannel(channel *amqp.Channel) {
	client.channel = channel
	client.notifyChanClose = make(chan *amqp.Error, 1)
	client.notifyConfirm = make(chan amqp.Confirmation, 1)
	client.channel.NotifyClose(client.notifyChanClose)
	client.channel.NotifyPublish(client.notifyConfirm)
}

// Push will push data onto the queue, and wait for a confirmation.
// This will block until the server sends a confirmation. Errors are
// only returned if the push action itself fails, see UnsafePush.
func (client *Client) Push(data []byte) error {
	client.m.Lock()
	if !client.isReady {
		client.m.Unlock()
		return errors.New("failed to push: not connected")
	}
	client.m.Unlock()
	for {
		err := client.UnsafePush(data)
		if err != nil {
			client.logger.Println("Push failed. Retrying...")
			select {
			case <-client.done:
				return errShutdown
			case <-time.After(resendDelay):
			}
			continue
		}
		confirm := <-client.notifyConfirm
		if confirm.Ack {
			client.logger.Printf("Push confirmed [%d]!", confirm.DeliveryTag)
			return nil
		}
	}
}

// UnsafePush will push to the queue without checking for
// confirmation. It returns an error if it fails to connect.
// No guarantees are provided for whether the server will
// receive the message.
func (client *Client) UnsafePush(data []byte) error {
	client.m.Lock()
	if !client.isReady {
		client.m.Unlock()
		return errNotConnected
	}
	client.m.Unlock()

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	return client.channel.PublishWithContext(
		ctx,
		"",
		client.queueName,
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        data,
		},
	)
}

// Consume will continuously put queue items on the channel.
// It is required to call delivery.Ack when it has been
// successfully processed, or delivery.Nack when it fails.
// Ignoring this will cause data to build up on the server.
func (client *Client) Consume() (<-chan amqp.Delivery, error) {
	client.m.Lock()
	if !client.isReady {
		client.m.Unlock()
		return nil, errNotConnected
	}
	client.m.Unlock()

	if err := client.channel.Qos(
		1,
		0,
		false,
	); err != nil {
		return nil, err
	}

	return client.channel.Consume(
		client.queueName,
		"",
		false,
		false,
		false,
		false,
		nil,
	)
}

// Close will cleanly shut down the channel and connection.
func (client *Client) Close() error {
	client.m.Lock()

	defer client.m.Unlock()

	if !client.isReady {
		return errAlreadyClosed
	}
	close(client.done)
	err := client.channel.Close()
	if err != nil {
		return err
	}
	err = client.connection.Close()
	if err != nil {
		return err
	}

	client.isReady = false
	return nil
}

Index

• Constants
• Variables
• func DefaultDial(connectionTimeout time.Duration) func(network, addr string) (net.Conn, error)
• func SetLogger(logger Logging)
• type AMQPlainAuth
  • func (auth *AMQPlainAuth) Mechanism() string
  • func (auth *AMQPlainAuth) Response() string
• type Acknowledger
• type Authentication
• type Blocking
• type Channel
  • func (ch *Channel) Ack(tag uint64, multiple bool) error
  • func (ch *Channel) Cancel(consumer string, noWait bool) error
  • func (ch *Channel) Close() error
  • func (ch *Channel) Confirm(noWait bool) error
  • func (ch *Channel) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args Table) (<-chan Delivery, error)
  • func (ch *Channel) ConsumeWithContext(ctx context.Context, queue, consumer string, ...) (<-chan Delivery, error)
  • func (ch *Channel) ExchangeBind(destination, key, source string, noWait bool, args Table) error
  • func (ch *Channel) ExchangeDeclare(name, kind string, durable, autoDelete, internal, noWait bool, args Table) error
  • func (ch *Channel) ExchangeDeclarePassive(name, kind string, durable, autoDelete, internal, noWait bool, args Table) error
  • func (ch *Channel) ExchangeDelete(name string, ifUnused, noWait bool) error
  • func (ch *Channel) ExchangeUnbind(destination, key, source string, noWait bool, args Table) error
  • func (ch *Channel) Flow(active bool) error
  • func (ch *Channel) Get(queue string, autoAck bool) (msg Delivery, ok bool, err error)
  • func (ch *Channel) GetNextPublishSeqNo() uint64
  • func (ch *Channel) IsClosed() bool
  • func (ch *Channel) Nack(tag uint64, multiple bool, requeue bool) error
  • func (ch *Channel) NotifyCancel(c chan string) chan string
  • func (ch *Channel) NotifyClose(c chan *Error) chan *Error
  • func (ch *Channel) NotifyConfirm(ack, nack chan uint64) (chan uint64, chan uint64)
  • func (ch *Channel) NotifyFlow(c chan bool) chan bool
  • func (ch *Channel) NotifyPublish(confirm chan Confirmation) chan Confirmation
  • func (ch *Channel) NotifyReturn(c chan Return) chan Return
  • func (ch *Channel) Publish(exchange, key string, mandatory, immediate bool, msg Publishing) error
  • func (ch *Channel) PublishWithContext(_ context.Context, exchange, key string, mandatory, immediate bool, ...) error
  • func (ch *Channel) PublishWithDeferredConfirm(exchange, key string, mandatory, immediate bool, msg Publishing) (*DeferredConfirmation, error)
  • func (ch *Channel) PublishWithDeferredConfirmWithContext(_ context.Context, exchange, key string, mandatory, immediate bool, ...) (*DeferredConfirmation, error)
  • func (ch *Channel) Qos(prefetchCount, prefetchSize int, global bool) error
  • func (ch *Channel) QueueBind(name, key, exchange string, noWait bool, args Table) error
  • func (ch *Channel) QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, args Table) (Queue, error)
  • func (ch *Channel) QueueDeclarePassive(name string, durable, autoDelete, exclusive, noWait bool, args Table) (Queue, error)
  • func (ch *Channel) QueueDelete(name string, ifUnused, ifEmpty, noWait bool) (int, error)
  • func (ch *Channel) QueueInspect(name string) (Queue, error)
  • func (ch *Channel) QueuePurge(name string, noWait bool) (int, error)
  • func (ch *Channel) QueueUnbind(name, key, exchange string, args Table) error
  • func (ch *Channel) Recover(requeue bool) error
  • func (ch *Channel) Reject(tag uint64, requeue bool) error
  • func (ch *Channel) Tx() error
  • func (ch *Channel) TxCommit() error
  • func (ch *Channel) TxRollback() error
• type Config
• type Confirmation
• type Connection
  • func Dial(url string) (*Connection, error)
  • func DialConfig(url string, config Config) (*Connection, error)
  • func DialTLS(url string, amqps *tls.Config) (*Connection, error)
  • func DialTLS_ExternalAuth(url string, amqps *tls.Config) (*Connection, error)
  • func Open(conn io.ReadWriteCloser, config Config) (*Connection, error)
  • func (c *Connection) Channel() (*Channel, error)
  • func (c *Connection) Close() error
  • func (c *Connection) CloseDeadline(deadline time.Time) error
  • func (c *Connection) ConnectionState() tls.ConnectionState
  • func (c *Connection) IsClosed() bool
  • func (c *Connection) LocalAddr() net.Addr
  • func (c *Connection) NotifyBlocked(receiver chan Blocking) chan Blocking
  • func (c *Connection) NotifyClose(receiver chan *Error) chan *Error
  • func (c *Connection) RemoteAddr() net.Addr
  • func (c *Connection) UpdateSecret(newSecret, reason string) error
• type Decimal
• type DeferredConfirmation
  • func (d *DeferredConfirmation) Acked() bool
  • func (d *DeferredConfirmation) Done() <-chan struct{}
  • func (d *DeferredConfirmation) Wait() bool
  • func (d *DeferredConfirmation) WaitContext(ctx context.Context) (bool, error)
• type Delivery
  • func (d Delivery) Ack(multiple bool) error
  • func (d Delivery) Nack(multiple, requeue bool) error
  • func (d Delivery) Reject(requeue bool) error
• type Error
  • func (e Error) Error() string
• type ExternalAuth
  • func (*ExternalAuth) Mechanism() string
  • func (*ExternalAuth) Response() string
• type Logging
• type NullLogger
  • func (l NullLogger) Printf(format string, v ...interface{})
• type PlainAuth
  • func (auth *PlainAuth) Mechanism() string
  • func (auth *PlainAuth) Response() string
• type Publishing
• type Queue
• type Return
• type Table
  • func NewConnectionProperties() Table
  • func (t Table) SetClientConnectionName(connName string)
  • func (t Table) Validate() error
• type URI
  • func ParseURI(uri string) (URI, error)
  • func (uri URI) AMQPlainAuth() *AMQPlainAuth
  • func (uri URI) PlainAuth() *PlainAuth
  • func (uri URI) String() string

Examples

• Package (Consume)
• Package (Publish)
• Channel.Confirm (Bridge)
• Channel.Consume
• Channel.PublishWithContext
• Channel.QueueDeclare (ClassicQueueV2)
• Channel.QueueDeclare (ConsumerTimeout)
• Channel.QueueDeclare (Quorum)
• Channel.QueueDeclare (Stream)
• Config (Timeout)
• Connection (Reconnect)
• Connection.NotifyBlocked
• Connection.UpdateSecret
• DialTLS
• Table.SetClientConnectionName

Constants

const (
	ContentTooLarge    = 311
	NoRoute            = 312
	NoConsumers        = 313
	ConnectionForced   = 320
	InvalidPath        = 402
	AccessRefused      = 403
	NotFound           = 404
	ResourceLocked     = 405
	PreconditionFailed = 406
	FrameError         = 501
	SyntaxError        = 502
	CommandInvalid     = 503
	ChannelError       = 504
	UnexpectedFrame    = 505
	ResourceError      = 506
	NotAllowed         = 530
	NotImplemented     = 540
	InternalError      = 541
)

Error codes that can be sent from the server during a connection or channel exception or used by the client to indicate a class of error like ErrCredentials. The text of the error is likely more interesting than these constants.

const (
	ExchangeDirect  = "direct"
	ExchangeFanout  = "fanout"
	ExchangeTopic   = "topic"
	ExchangeHeaders = "headers"
)

Constants for standard AMQP 0-9-1 exchange types.

const (
	Transient  uint8 = 1
	Persistent uint8 = 2
)

DeliveryMode. Transient means higher throughput but messages will not be restored on broker restart. The delivery mode of publishings is unrelated to the durability of the queues they reside on. Transient messages will not be restored to durable queues, persistent messages will be restored to durable queues and lost on non-durable queues during server restart.

This remains typed as uint8 to match Publishing.DeliveryMode. Other delivery modes specific to custom queue implementations are not enumerated here.

const (
	NeverExpire       string = ""  // empty value means never expire
	ImmediatelyExpire string = "0" // 0 means immediately expire
)

Expiration. These constants can be used to set a messages expiration TTL. They should be viewed as a clarification of the expiration functionality in messages and their usage is not enforced by this pkg.

The server requires a string value that is interpreted by the server as milliseconds. If no value is set, which translates to the nil value of string, the message will never expire by itself. This does not influence queue configured TTL configurations.

const (
	QueueTypeArg                 = "x-queue-type"
	QueueMaxLenArg               = "x-max-length"
	QueueMaxLenBytesArg          = "x-max-length-bytes"
	StreamMaxLenBytesArg         = "x-max-length-bytes"
	QueueOverflowArg             = "x-overflow"
	QueueMessageTTLArg           = "x-message-ttl"
	QueueTTLArg                  = "x-expires"
	StreamMaxAgeArg              = "x-max-age"
	StreamMaxSegmentSizeBytesArg = "x-stream-max-segment-size-bytes"
	// QueueVersionArg declares the Classic Queue version to use. Expects an integer, either 1 or 2.
	QueueVersionArg = "x-queue-version"
	// ConsumerTimeoutArg is available in RabbitMQ 3.12+ as a queue argument.
	ConsumerTimeoutArg      = "x-consumer-timeout"
	SingleActiveConsumerArg = "x-single-active-consumer"
)

Most common queue argument keys in queue declaration. For a comprehensive list of queue arguments, visit RabbitMQ Queue docs.

QueueTypeArg queue argument is used to declare quorum and stream queues. Accepted values are QueueTypeClassic (default), QueueTypeQuorum and QueueTypeStream. Quorum Queues accept (almost) all queue arguments as their Classic Queues counterparts. Check feature comparison docs for more information.

Queues can define their max length using QueueMaxLenArg and QueueMaxLenBytesArg queue arguments. Overflow behaviour is set using QueueOverflowArg. Accepted values are QueueOverflowDropHead (default), QueueOverflowRejectPublish and QueueOverflowRejectPublishDLX.

Queue TTL can be defined using QueueTTLArg. That is, the time-to-live for an unused queue. Queue Message TTL can be defined using QueueMessageTTLArg. This will set a time-to-live for messages in the queue.

Stream retention can be configured using StreamMaxLenBytesArg, to set the maximum size of the stream. Please note that stream queues always keep, at least, one segment. Stream retention can also be set using StreamMaxAgeArg, to set time-based retention. Values are string with unit suffix. Valid suffixes are Y, M, D, h, m, s. E.g. "7D" for one week. The maximum segment size can be set using StreamMaxSegmentSizeBytesArg. The default value is 500_000_000 bytes ~= 500 megabytes

Starting with RabbitMQ 3.12, consumer timeout can be configured as a queue argument. This is the timeout for a consumer to acknowledge a message. The value is the time in milliseconds. The timeout is evaluated periodically, at one minute intervals. Values lower than one minute are not supported. See the consumer timeout guide for more information.

Single Active Consumer on quorum and classic queues can be configured using SingleActiveConsumerArg. This argument expects a boolean value. It is false by default.

const (
	QueueTypeClassic              = "classic"
	QueueTypeQuorum               = "quorum"
	QueueTypeStream               = "stream"
	QueueOverflowDropHead         = "drop-head"
	QueueOverflowRejectPublish    = "reject-publish"
	QueueOverflowRejectPublishDLX = "reject-publish-dlx"
)

Values for queue arguments. Use as values for queue arguments during queue declaration. The following argument table will create a classic queue, with max length set to 100 messages, and a queue TTL of 30 minutes.

args := amqp.Table{
	amqp.QueueTypeArg: QueueTypeClassic,
	amqp.QueueMaxLenArg: 100,
	amqp.QueueTTLArg: 1800000,
}

Refer to Channel.QueueDeclare for more examples.

const DefaultExchange = ""

DefaultExchange is the default direct exchange that binds every queue by its name. Applications can route to a queue using the queue name as routing key.

Variables

var (
	// ErrClosed is returned when the channel or connection is not open
	ErrClosed = &Error{Code: ChannelError, Reason: "channel/connection is not open"}

	// ErrChannelMax is returned when Connection.Channel has been called enough
	// times that all channel IDs have been exhausted in the client or the
	// server.
	ErrChannelMax = &Error{Code: ChannelError, Reason: "channel id space exhausted"}

	// ErrSASL is returned from Dial when the authentication mechanism could not
	// be negotiated.
	ErrSASL = &Error{Code: AccessRefused, Reason: "SASL could not negotiate a shared mechanism"}

	// ErrCredentials is returned when the authenticated client is not authorized
	// to any vhost.
	ErrCredentials = &Error{Code: AccessRefused, Reason: "username or password not allowed"}

	// ErrVhost is returned when the authenticated user is not permitted to
	// access the requested Vhost.
	ErrVhost = &Error{Code: AccessRefused, Reason: "no access to this vhost"}

	// ErrSyntax is hard protocol error, indicating an unsupported protocol,
	// implementation or encoding.
	ErrSyntax = &Error{Code: SyntaxError, Reason: "invalid field or value inside of a frame"}

	// ErrFrame is returned when the protocol frame cannot be read from the
	// server, indicating an unsupported protocol or unsupported frame type.
	ErrFrame = &Error{Code: FrameError, Reason: "frame could not be parsed"}

	// ErrCommandInvalid is returned when the server sends an unexpected response
	// to this requested message type. This indicates a bug in this client.
	ErrCommandInvalid = &Error{Code: CommandInvalid, Reason: "unexpected command received"}

	// ErrUnexpectedFrame is returned when something other than a method or
	// heartbeat frame is delivered to the Connection, indicating a bug in the
	// client.
	ErrUnexpectedFrame = &Error{Code: UnexpectedFrame, Reason: "unexpected frame received"}

	// ErrFieldType is returned when writing a message containing a Go type unsupported by AMQP.
	ErrFieldType = &Error{Code: SyntaxError, Reason: "unsupported table field type"}
)

Functions

func DefaultDial

func DefaultDial(connectionTimeout time.Duration) func(network, addr string) (net.Conn, error)

DefaultDial establishes a connection when config.Dial is not provided

func SetLogger

func SetLogger(logger Logging)

Enables logging using a custom Logging instance. Note that this is not thread safe and should be called at application start

Types

type AMQPlainAuth

type AMQPlainAuth struct {
	Username string
	Password string
}

AMQPlainAuth is similar to PlainAuth

func (*AMQPlainAuth) Mechanism

func (auth *AMQPlainAuth) Mechanism() string

Mechanism returns "AMQPLAIN"

func (*AMQPlainAuth) Response

func (auth *AMQPlainAuth) Response() string

Response returns an AMQP encoded credentials table, without the field table size.

type Acknowledger

type Acknowledger interface {
	Ack(tag uint64, multiple bool) error
	Nack(tag uint64, multiple bool, requeue bool) error
	Reject(tag uint64, requeue bool) error
}

Acknowledger notifies the server of successful or failed consumption of deliveries via identifier found in the Delivery.DeliveryTag field.

Applications can provide mock implementations in tests of Delivery handlers.

type Authentication

type Authentication interface {
	Mechanism() string
	Response() string
}

Authentication interface provides a means for different SASL authentication mechanisms to be used during connection tuning.

type Blocking

type Blocking struct {
	Active bool   // TCP pushback active/inactive on server
	Reason string // Server reason for activation
}

Blocking notifies the server's TCP flow control of the Connection. When a server hits a memory or disk alarm it will block all connections until the resources are reclaimed. Use NotifyBlock on the Connection to receive these events.

type Channel

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

Channel represents an AMQP channel. Used as a context for valid message exchange. Errors on methods with this Channel as a receiver means this channel should be discarded and a new channel established.

func (*Channel) Ack

func (ch *Channel) Ack(tag uint64, multiple bool) error

Ack acknowledges a delivery by its delivery tag when having been consumed with Channel.Consume or Channel.Get.

Ack acknowledges all message received prior to the delivery tag when multiple is true.

See also Delivery.Ack

func (*Channel) Cancel

func (ch *Channel) Cancel(consumer string, noWait bool) error

Cancel stops deliveries to the consumer chan established in Channel.Consume and identified by consumer.

Only use this method to cleanly stop receiving deliveries from the server and cleanly shut down the consumer chan identified by this tag. Using this method and waiting for remaining messages to flush from the consumer chan will ensure all messages received on the network will be delivered to the receiver of your consumer chan.

Continue consuming from the chan Delivery provided by Channel.Consume until the chan closes.

When noWait is true, do not wait for the server to acknowledge the cancel. Only use this when you are certain there are no deliveries in flight that require an acknowledgment, otherwise they will arrive and be dropped in the client without an ack, and will not be redelivered to other consumers.

func (*Channel) Close

func (ch *Channel) Close() error

Close initiate a clean channel closure by sending a close message with the error code set to '200'.

It is safe to call this method multiple times.

func (*Channel) Confirm

func (ch *Channel) Confirm(noWait bool) error

Confirm puts this channel into confirm mode so that the client can ensure all publishings have successfully been received by the server. After entering this mode, the server will send a basic.ack or basic.nack message with the deliver tag set to a 1 based incremental index corresponding to every publishing received after the this method returns.

Add a listener to Channel.NotifyPublish to respond to the Confirmations. If Channel.NotifyPublish is not called, the Confirmations will be silently ignored.

The order of acknowledgments is not bound to the order of deliveries.

Ack and Nack confirmations will arrive at some point in the future.

Unroutable mandatory or immediate messages are acknowledged immediately after any Channel.NotifyReturn listeners have been notified. Other messages are acknowledged when all queues that should have the message routed to them have either received acknowledgment of delivery or have enqueued the message, persisting the message if necessary.

When noWait is true, the client will not wait for a response. A channel exception could occur if the server does not support this method.

Example (Bridge)

package main

import (
	"context"
	"log"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
	// This example acts as a bridge, shoveling all messages sent from the source
	// exchange "log" to destination exchange "log".

	// Confirming publishes can help from overproduction and ensure every message
	// is delivered.

	// Setup the source of the store and forward
	source, err := amqp.Dial("amqp://source/")
	if err != nil {
		log.Fatalf("connection.open source: %s", err)
	}
	defer source.Close()

	chs, err := source.Channel()
	if err != nil {
		log.Fatalf("channel.open source: %s", err)
	}

	if err := chs.ExchangeDeclare("log", "topic", true, false, false, false, nil); err != nil {
		log.Fatalf("exchange.declare destination: %s", err)
	}

	if _, err := chs.QueueDeclare("remote-tee", true, true, false, false, nil); err != nil {
		log.Fatalf("queue.declare source: %s", err)
	}

	if err := chs.QueueBind("remote-tee", "#", "logs", false, nil); err != nil {
		log.Fatalf("queue.bind source: %s", err)
	}

	shovel, err := chs.Consume("remote-tee", "shovel", false, false, false, false, nil)
	if err != nil {
		log.Fatalf("basic.consume source: %s", err)
	}

	// Setup the destination of the store and forward
	destination, err := amqp.Dial("amqp://destination/")
	if err != nil {
		log.Fatalf("connection.open destination: %s", err)
	}
	defer destination.Close()

	chd, err := destination.Channel()
	if err != nil {
		log.Fatalf("channel.open destination: %s", err)
	}

	if err := chd.ExchangeDeclare("log", "topic", true, false, false, false, nil); err != nil {
		log.Fatalf("exchange.declare destination: %s", err)
	}

	// Buffer of 1 for our single outstanding publishing
	confirms := chd.NotifyPublish(make(chan amqp.Confirmation, 1))

	if err := chd.Confirm(false); err != nil {
		log.Fatalf("confirm.select destination: %s", err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	// Now pump the messages, one by one, a smarter implementation
	// would batch the deliveries and use multiple ack/nacks
	for {
		msg, ok := <-shovel
		if !ok {
			log.Fatalf("source channel closed, see the reconnect example for handling this")
		}

		err = chd.PublishWithContext(ctx, "logs", msg.RoutingKey, false, false, amqp.Publishing{
			// Copy all the properties
			ContentType:     msg.ContentType,
			ContentEncoding: msg.ContentEncoding,
			DeliveryMode:    msg.DeliveryMode,
			Priority:        msg.Priority,
			CorrelationId:   msg.CorrelationId,
			ReplyTo:         msg.ReplyTo,
			Expiration:      msg.Expiration,
			MessageId:       msg.MessageId,
			Timestamp:       msg.Timestamp,
			Type:            msg.Type,
			UserId:          msg.UserId,
			AppId:           msg.AppId,

			// Custom headers
			Headers: msg.Headers,

			// And the body
			Body: msg.Body,
		})

		if err != nil {
			if e := msg.Nack(false, false); e != nil {
				log.Printf("nack error: %+v", e)
			}
			log.Fatalf("basic.publish destination: %+v", msg)
		}

		// only ack the source delivery when the destination acks the publishing
		if confirmed := <-confirms; confirmed.Ack {
			if e := msg.Ack(false); e != nil {
				log.Printf("ack error: %+v", e)
			}
		} else {
			if e := msg.Nack(false, false); e != nil {
				log.Printf("nack error: %+v", e)
			}
		}
	}
}

func (*Channel) Consume

func (ch *Channel) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args Table) (<-chan 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.

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

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.

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.

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.

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.

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. RabbitMQ will requeue messages not acknowledged. In other words, dropped messages in this way won't be lost.

When the consumer tag is cancelled, all inflight messages will be delivered until the returned chan is closed.

Example

package main

import (
	"log"
	"runtime"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
	// Connects opens an AMQP connection from the credentials in the URL.
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	if err != nil {
		log.Fatalf("connection.open: %s", err)
	}
	defer conn.Close()

	c, err := conn.Channel()
	if err != nil {
		log.Fatalf("channel.open: %s", err)
	}

	// We declare our topology on both the publisher and consumer to ensure they
	// are the same.  This is part of AMQP being a programmable messaging model.
	//
	// See the Channel.Publish example for the complimentary declare.
	err = c.ExchangeDeclare("logs", "topic", true, false, false, false, nil)
	if err != nil {
		log.Fatalf("exchange.declare: %s", err)
	}

	// Establish our queue topologies that we are responsible for
	type bind struct {
		queue string
		key   string
	}

	bindings := []bind{
		{"page", "alert"},
		{"email", "info"},
		{"firehose", "#"},
	}

	for _, b := range bindings {
		_, err = c.QueueDeclare(b.queue, true, false, false, false, nil)
		if err != nil {
			log.Fatalf("queue.declare: %v", err)
		}

		err = c.QueueBind(b.queue, b.key, "logs", false, nil)
		if err != nil {
			log.Fatalf("queue.bind: %v", err)
		}
	}

	// Set our quality of service.  Since we're sharing 3 consumers on the same
	// channel, we want at least 3 messages in flight.
	err = c.Qos(3, 0, false)
	if err != nil {
		log.Fatalf("basic.qos: %v", err)
	}

	// Establish our consumers that have different responsibilities.  Our first
	// two queues do not ack the messages on the server, so require to be acked
	// on the client.

	pages, err := c.Consume("page", "pager", false, false, false, false, nil)
	if err != nil {
		log.Fatalf("basic.consume: %v", err)
	}

	go func() {
		for page := range pages {
			// ... this consumer is responsible for sending pages per log
			if e := page.Ack(false); e != nil {
				log.Printf("ack error: %+v", e)
			}
		}
	}()

	// Notice how the concern for which messages arrive here are in the AMQP
	// topology and not in the queue.  We let the server pick a consumer tag this
	// time.

	emails, err := c.Consume("email", "", false, false, false, false, nil)
	if err != nil {
		log.Fatalf("basic.consume: %v", err)
	}

	go func() {
		for email := range emails {
			// ... this consumer is responsible for sending emails per log
			if e := email.Ack(false); e != nil {
				log.Printf("ack error: %+v", e)
			}
		}
	}()

	// This consumer requests that every message is acknowledged as soon as it's
	// delivered.

	firehose, err := c.Consume("firehose", "", true, false, false, false, nil)
	if err != nil {
		log.Fatalf("basic.consume: %v", err)
	}

	// To show how to process the items in parallel, we'll use a work pool.
	for i := 0; i < runtime.NumCPU(); i++ {
		go func(work <-chan amqp.Delivery) {
			for range work {
				// ... this consumer pulls from the firehose and doesn't need to acknowledge
			}
		}(firehose)
	}

	// Wait until you're ready to finish, could be a signal handler here.
	time.Sleep(10 * time.Second)

	// Cancelling a consumer by name will finish the range and gracefully end the
	// goroutine
	err = c.Cancel("pager", false)
	if err != nil {
		log.Fatalf("basic.cancel: %v", err)
	}

	// deferred closing the Connection will also finish the consumer's ranges of
	// their delivery chans.  If you need every delivery to be processed, make
	// sure to wait for all consumers goroutines to finish before exiting your
	// process.
}

func (*Channel) ConsumeWithContext

func (ch *Channel) ConsumeWithContext(ctx context.Context, queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args Table) (<-chan Delivery, error)

ConsumeWithContext immediately starts delivering queued messages.

This function is similar to Channel.Consume, and accepts a context to control consumer lifecycle. When the context passed to this function is canceled, the consumer associated with the deliveries channel will be canceled too. When the context passed to this function is cancelled, the deliveries channel will be closed.

An application is advised to keep on receiving messages from the delivery channel until the channel is empty. This is specially important to avoid memory leaks from unconsumed messages from the delivery channel.

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, context is cancelled, 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.

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

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.

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.

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.

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.

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. RabbitMQ will requeue messages not acknowledged. In other words, dropped messages in this way won't be lost.

func (*Channel) ExchangeBind

func (ch *Channel) ExchangeBind(destination, key, source string, noWait bool, args Table) 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

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.

Optional arguments specific to the exchanges bound can also be specified.

func (*Channel) ExchangeDeclare

func (ch *Channel) ExchangeDeclare(name, kind string, durable, autoDelete, internal, noWait bool, args Table) 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".

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.

Exchanges declared as `internal` do not accept publishings. Internal exchanges are useful when you wish to implement inter-exchange topologies that should not be exposed to users of the broker.

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.

Optional amqp.Table of arguments that are specific to the server's implementation of the exchange can be sent for exchange types that require extra parameters.

func (*Channel) ExchangeDeclarePassive

func (ch *Channel) ExchangeDeclarePassive(name, kind string, durable, autoDelete, internal, noWait bool, args Table) 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 (*Channel) ExchangeDelete

func (ch *Channel) ExchangeDelete(name string, ifUnused, noWait bool) 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.

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.

When noWait is true, do not wait for a server confirmation that the exchange has been deleted. Failing to delete the channel could close the channel. Add a NotifyClose listener to respond to these channel exceptions.

func (*Channel) ExchangeUnbind

func (ch *Channel) ExchangeUnbind(destination, key, source string, noWait bool, args Table) 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.

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.

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.

func (*Channel) Flow

func (ch *Channel) Flow(active bool) error

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 (*Channel) Get

func (ch *Channel) Get(queue string, autoAck bool) (msg Delivery, ok bool, err error)

Get synchronously receives a single Delivery from the head of a queue from the server to the client. In almost all cases, using Channel.Consume will be preferred.

If there was a delivery waiting on the queue and that delivery was received, the second return value will be true. If there was no delivery waiting or an error occurred, the ok bool will be false.

All deliveries must be acknowledged including those from Channel.Get. Call Delivery.Ack on the returned delivery when you have fully processed this delivery.

When autoAck is true, the server will automatically acknowledge this message so you don't have to. But if you are unable to fully process this message before the channel or connection is closed, the message will not get requeued.

func (*Channel) GetNextPublishSeqNo

func (ch *Channel) GetNextPublishSeqNo() uint64

GetNextPublishSeqNo returns the sequence number of the next message to be published, when in confirm mode.

func (*Channel) IsClosed

func (ch *Channel) IsClosed() bool

IsClosed returns true if the channel is marked as closed, otherwise false is returned.

func (*Channel) Nack

func (ch *Channel) Nack(tag uint64, multiple bool, requeue bool) error

Nack negatively acknowledges a delivery by its delivery tag. Prefer this method to notify the server that you were not able to process this delivery and it must be redelivered or dropped.

See also Delivery.Nack

func (*Channel) NotifyCancel

func (ch *Channel) NotifyCancel(c chan string) chan string

NotifyCancel registers a listener for basic.cancel methods. These can be sent from the server when a queue is deleted or when consuming from a mirrored queue where the master has just failed (and was moved to another node).

The subscription tag is returned to the listener.

func (*Channel) NotifyClose

func (ch *Channel) NotifyClose(c chan *Error) chan *Error

NotifyClose registers a listener for when the server sends a channel or connection exception in the form of a Connection.Close or Channel.Close method. Connection exceptions will be broadcast to all open channels and all channels will be closed, where channel exceptions will only be broadcast to listeners to this channel.

The chan provided will be closed when the Channel is closed and on a graceful close, no error will be sent.

In case of a non graceful close the error will be notified synchronously by the library so that it will be necessary to consume the Channel from the caller in order to avoid deadlocks

func (*Channel) NotifyConfirm

func (ch *Channel) NotifyConfirm(ack, nack chan uint64) (chan uint64, chan uint64)

NotifyConfirm calls NotifyPublish and starts a goroutine sending ordered Ack and Nack DeliveryTag to the respective channels.

For strict ordering, use NotifyPublish instead.

func (*Channel) NotifyFlow

func (ch *Channel) NotifyFlow(c chan bool) chan bool

NotifyFlow registers a listener for basic.flow methods sent by the server. When `false` is sent on one of the listener channels, all publishers should pause until a `true` is sent.

The server may ask the producer to pause or restart the flow of Publishings sent by on a channel. This is a simple flow-control mechanism that a server can use to avoid overflowing its queues or otherwise finding itself receiving more messages than it can process. Note that this method is not intended for window control. It does not affect contents returned by basic.get-ok methods.

When a new channel is opened, it is active (flow is active). Some applications assume that channels are inactive until started. To emulate this behavior a client MAY open the channel, then pause it.

Publishers should respond to a flow messages as rapidly as possible and the server may disconnect over producing channels that do not respect these messages.

basic.flow-ok methods will always be returned to the server regardless of the number of listeners there are.

To control the flow of deliveries from the server, use the Channel.Flow() method instead.

Note: RabbitMQ will rather use TCP pushback on the network connection instead of sending basic.flow. This means that if a single channel is producing too much on the same connection, all channels using that connection will suffer, including acknowledgments from deliveries. Use different Connections if you desire to interleave consumers and producers in the same process to avoid your basic.ack messages from getting rate limited with your basic.publish messages.

func (*Channel) NotifyPublish

func (ch *Channel) NotifyPublish(confirm chan Confirmation) chan Confirmation

NotifyPublish registers a listener for reliable publishing. Receives from this chan for every publish after Channel.Confirm will be in order starting with DeliveryTag 1.

There will be one and only one Confirmation Publishing starting with the delivery tag of 1 and progressing sequentially until the total number of Publishings have been seen by the server.

Acknowledgments will be received in the order of delivery from the NotifyPublish channels even if the server acknowledges them out of order.

The listener chan will be closed when the Channel is closed.

The capacity of the chan Confirmation must be at least as large as the number of outstanding publishings. Not having enough buffered chans will create a deadlock if you attempt to perform other operations on the Connection or Channel while confirms are in-flight.

It's advisable to wait for all Confirmations to arrive before calling Channel.Close() or Connection.Close().

It is also advisable for the caller to consume from the channel returned till it is closed to avoid possible deadlocks

func (*Channel) NotifyReturn

func (ch *Channel) NotifyReturn(c chan Return) chan Return

NotifyReturn registers a listener for basic.return methods. These can be sent from the server when a publish is undeliverable either from the mandatory or immediate flags.

A return struct has a copy of the Publishing along with some error information about why the publishing failed.

func (*Channel) Publish

func (ch *Channel) Publish(exchange, key string, mandatory, immediate bool, msg Publishing) 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.

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.

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 (*Channel) PublishWithContext

func (ch *Channel) PublishWithContext(_ context.Context, exchange, key string, mandatory, immediate bool, msg Publishing) error

PublishWithContext sends a Publishing from the client to an exchange on the server.

NOTE: this function is equivalent to Channel.Publish. Context is not honoured.

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.

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.

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.

Example

package main

import (
	"context"
	"log"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
	// Connects opens an AMQP connection from the credentials in the URL.
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	if err != nil {
		log.Fatalf("connection.open: %s", err)
	}

	// This waits for a server acknowledgment which means the sockets will have
	// flushed all outbound publishings prior to returning.  It's important to
	// block on Close to not lose any publishings.
	defer conn.Close()

	c, err := conn.Channel()
	if err != nil {
		log.Fatalf("channel.open: %s", err)
	}

	// We declare our topology on both the publisher and consumer to ensure they
	// are the same.  This is part of AMQP being a programmable messaging model.
	//
	// See the Channel.Consume example for the complimentary declare.
	err = c.ExchangeDeclare("logs", "topic", true, false, false, false, nil)
	if err != nil {
		log.Fatalf("exchange.declare: %v", err)
	}

	// Prepare this message to be persistent.  Your publishing requirements may
	// be different.
	msg := amqp.Publishing{
		DeliveryMode: amqp.Persistent,
		Timestamp:    time.Now(),
		ContentType:  "text/plain",
		Body:         []byte("Go Go AMQP!"),
	}

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	// This is not a mandatory delivery, so it will be dropped if there are no
	// queues bound to the logs exchange.
	err = c.PublishWithContext(ctx, "logs", "info", false, false, msg)
	if err != nil {
		// Since publish is asynchronous this can happen if the network connection
		// is reset or if the server has run out of resources.
		log.Fatalf("basic.publish: %v", err)
	}
}

func (*Channel) PublishWithDeferredConfirm

func (ch *Channel) PublishWithDeferredConfirm(exchange, key string, mandatory, immediate bool, msg Publishing) (*DeferredConfirmation, error)

PublishWithDeferredConfirm behaves identically to Publish, but additionally returns a DeferredConfirmation, allowing the caller to wait on the publisher confirmation for this message. If the channel has not been put into confirm mode, the DeferredConfirmation will be nil.

func (*Channel) PublishWithDeferredConfirmWithContext

func (ch *Channel) PublishWithDeferredConfirmWithContext(_ context.Context, exchange, key string, mandatory, immediate bool, msg Publishing) (*DeferredConfirmation, error)

PublishWithDeferredConfirmWithContext behaves identically to Publish but additionally returns a DeferredConfirmation, allowing the caller to wait on the publisher confirmation for this message. If the channel has not been put into confirm mode, the DeferredConfirmation will be nil.

NOTE: PublishWithDeferredConfirmWithContext is equivalent to its non-context variant. The context passed to this function is not honoured.

func (*Channel) Qos

func (ch *Channel) Qos(prefetchCount, prefetchSize int, global bool) 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.

When global is true, these Qos settings apply to all existing and future consumers on all channels on the same connection. When false, the Channel.Qos settings will apply to all existing and future consumers on this channel.

Please see the RabbitMQ Consumer Prefetch documentation for an explanation of how the global flag is implemented in RabbitMQ, as it differs from the AMQP 0.9.1 specification in that global Qos settings are limited in scope to channels, not connections (https://www.rabbitmq.com/consumer-prefetch.html).

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 (*Channel) QueueBind

func (ch *Channel) QueueBind(name, key, exchange string, noWait bool, args Table) 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.

When noWait is false and the queue could not be bound, the channel will be closed with an error.

func (*Channel) QueueDeclare

func (ch *Channel) QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, args Table) (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.

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

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.

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.

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.

Example (ClassicQueueV2)

package main

import (
	"log"

	amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
	conn, _ := amqp.Dial("amqp://localhost")
	ch, _ := conn.Channel()
	q, _ := ch.QueueDeclare(
		"my-classic-queue-v2", // queue name
		true,                  // durable
		false,                 // auto-delete
		false,                 // exclusive
		false,                 // noWait
		amqp.Table{
			amqp.QueueTypeArg:    amqp.QueueTypeClassic,
			amqp.QueueVersionArg: 2,
		},
	)
	log.Printf("Declared Classic Queue v2: %s", q.Name)
}

Example (ConsumerTimeout)

package main

import (
	"log"

	amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
	conn, _ := amqp.Dial("amqp://localhost")
	ch, _ := conn.Channel()
	// this works only with RabbitMQ 3.12+
	q, _ := ch.QueueDeclare(
		"my-classic-queue-v2", // queue name
		true,                  // durable
		false,                 // auto-delete
		false,                 // exclusive
		false,                 // noWait
		amqp.Table{
			amqp.QueueTypeArg:       amqp.QueueTypeQuorum, // also works with classic queues
			amqp.ConsumerTimeoutArg: 600_000,              // 10 minute consumer timeout
		},
	)
	log.Printf("Declared Classic Queue v2: %s", q.Name)
}

Example (Quorum)

package main

import (
	"log"

	amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
	conn, _ := amqp.Dial("amqp://localhost")
	ch, _ := conn.Channel()
	args := amqp.Table{ // queue args
		amqp.QueueTypeArg: amqp.QueueTypeQuorum,
	}
	q, _ := ch.QueueDeclare(
		"my-quorum-queue", // queue name
		true,              // durable
		false,             // auto-delete
		false,             // exclusive
		false,             // noWait
		args,
	)
	log.Printf("Declared queue: %s with arguments: %v", q.Name, args)
}

Example (Stream)

package main

import (
	"log"

	amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
	conn, _ := amqp.Dial("amqp://localhost")
	ch, _ := conn.Channel()
	q, _ := ch.QueueDeclare(
		"my-stream-queue", // queue name
		true,              // durable
		false,             // auto-delete
		false,             // exclusive
		false,             // noWait
		amqp.Table{ // queue args
			amqp.QueueTypeArg:                 amqp.QueueTypeStream,
			amqp.StreamMaxLenBytesArg:         int64(5_000_000_000), // 5 Gb
			amqp.StreamMaxSegmentSizeBytesArg: 500_000_000,          // 500 Mb
			amqp.StreamMaxAgeArg:              "3D",                 // 3 days
		},
	)
	log.Printf("Declared queue: %s", q.Name)
}

func (*Channel) QueueDeclarePassive

func (ch *Channel) QueueDeclarePassive(name string, durable, autoDelete, exclusive, noWait bool, args Table) (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 (*Channel) QueueDelete

func (ch *Channel) QueueDelete(name string, ifUnused, ifEmpty, noWait bool) (int, 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.

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.

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.

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.

func (*Channel) QueueInspect

func (ch *Channel) QueueInspect(name string) (Queue, error)

QueueInspect passively declares a queue by name to inspect the current message count and consumer count.

Use this method to check how many messages ready for delivery reside in the queue, how many consumers are receiving deliveries, and whether a queue by this name already exists.

If the queue by this name exists, use Channel.QueueDeclare check if it is declared with specific parameters.

If a queue by this name does not exist, an error will be returned and the channel will be closed.

Deprecated: Use QueueDeclare with "Passive: true" instead.

func (*Channel) QueuePurge

func (ch *Channel) QueuePurge(name string, noWait bool) (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.

If noWait is true, do not wait for the server response and the number of messages purged will not be meaningful.

func (*Channel) QueueUnbind

func (ch *Channel) QueueUnbind(name, key, exchange string, args Table) error

QueueUnbind removes a binding between an exchange and queue matching the key and arguments.

func (*Channel) Recover

func (ch *Channel) Recover(requeue bool) error

Recover redelivers all unacknowledged deliveries on this channel.

When requeue is false, messages will be redelivered to the original consumer.

When requeue is true, messages will be redelivered to any available consumer, potentially including the original.

If the deliveries cannot be recovered, an error will be returned and the channel will be closed.

Note: this method is not implemented on RabbitMQ, use Delivery.Nack instead

Deprecated: This method is deprecated in RabbitMQ. RabbitMQ used Recover(true) as a mechanism for consumers to tell the broker that they were ready for more deliveries, back in 2008-2009. Support for this will be removed from RabbitMQ in a future release. Use Nack() with requeue=true instead.

func (*Channel) Reject

func (ch *Channel) Reject(tag uint64, requeue bool) error

Reject negatively acknowledges a delivery by its delivery tag. Prefer Nack over Reject when communicating with a RabbitMQ server because you can Nack multiple messages, reducing the amount of protocol messages to exchange.

See also Delivery.Reject

func (*Channel) Tx

func (ch *Channel) 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 a 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 (*Channel) TxCommit

func (ch *Channel) 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 (*Channel) TxRollback

func (ch *Channel) 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 Config

type Config struct {
	// The SASL mechanisms to try in the client request, and the successful
	// mechanism used on the Connection object.
	// If SASL is nil, PlainAuth from the URL is used.
	SASL []Authentication

	// Vhost specifies the namespace of permissions, exchanges, queues and
	// bindings on the server.  Dial sets this to the path parsed from the URL.
	Vhost string

	ChannelMax uint16        // 0 max channels means 2^16 - 1
	FrameSize  int           // 0 max bytes means unlimited
	Heartbeat  time.Duration // less than 1s uses the server's interval

	// TLSClientConfig specifies the client configuration of the TLS connection
	// when establishing a tls transport.
	// If the URL uses an amqps scheme, then an empty tls.Config with the
	// ServerName from the URL is used.
	TLSClientConfig *tls.Config

	// Properties is table of properties that the client advertises to the server.
	// This is an optional setting - if the application does not set this,
	// the underlying library will use a generic set of client properties.
	Properties Table

	// Connection locale that we expect to always be en_US
	// Even though servers must return it as per the AMQP 0-9-1 spec,
	// we are not aware of it being used other than to satisfy the spec requirements
	Locale string

	// Dial returns a net.Conn prepared for a TLS handshake with TSLClientConfig,
	// then an AMQP connection handshake.
	// If Dial is nil, net.DialTimeout with a 30s connection and 30s deadline is
	// used during TLS and AMQP handshaking.
	Dial func(network, addr string) (net.Conn, error)
}

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.

Example (Timeout)

package main

import (
	"log"
	"net"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
	// Provide your own anonymous Dial function that delegates to net.DialTimout
	// for custom timeouts

	conn, err := amqp.DialConfig("amqp:///", amqp.Config{
		Dial: func(network, addr string) (net.Conn, error) {
			return net.DialTimeout(network, addr, 2*time.Second)
		},
	})

	log.Printf("conn: %v, err: %v", conn, err)
}

type Confirmation

type Confirmation struct {
	DeliveryTag uint64 // A 1 based counter of publishings from when the channel was put in Confirm mode
	Ack         bool   // True when the server successfully received the publishing
}

Confirmation notifies the acknowledgment or negative acknowledgement of a publishing identified by its delivery tag. Use NotifyPublish on the Channel to consume these events.

type Connection

type Connection struct {
	Config Config // The negotiated Config after connection.open

	Major      int      // Server's major version
	Minor      int      // Server's minor version
	Properties Table    // Server properties
	Locales    []string // Server locales
	// contains filtered or unexported fields
}

Connection manages the serialization and deserialization of frames from IO and dispatches the frames to the appropriate channel. All RPC methods and asynchronous Publishing, Delivery, Ack, Nack and Return messages are multiplexed on this channel. There must always be active receivers for every asynchronous message on this connection.

Example (Reconnect)

package main

import (
	"context"
	"fmt"
	"os"

	amqp "github.com/rabbitmq/amqp091-go"
)

// Every connection should declare the topology they expect
func setup(url, queue string) (*amqp.Connection, *amqp.Channel, error) {
	conn, err := amqp.Dial(url)
	if err != nil {
		return nil, nil, err
	}

	ch, err := conn.Channel()
	if err != nil {
		return nil, nil, err
	}

	if _, err := ch.QueueDeclare(queue, false, true, false, false, nil); err != nil {
		return nil, nil, err
	}

	return conn, ch, nil
}

func consume(url, queue string) (*amqp.Connection, <-chan amqp.Delivery, error) {
	conn, ch, err := setup(url, queue)
	if err != nil {
		return nil, nil, err
	}

	// Indicate we only want 1 message to acknowledge at a time.
	if err := ch.Qos(1, 0, false); err != nil {
		return nil, nil, err
	}

	// Exclusive consumer
	deliveries, err := ch.Consume(queue, "", false, true, false, false, nil)

	return conn, deliveries, err
}

func main() {
	if url := os.Getenv("AMQP_URL"); url != "" {
		queue := "example.reconnect"

		// The connection/channel for publishing to interleave the ingress messages
		// between reconnects, shares the same topology as the consumer.  If we rather
		// sent all messages up front, the first consumer would receive every message.
		// We would rather show how the messages are not lost between reconnects.
		con, pub, err := setup(url, queue)
		if err != nil {
			fmt.Println("err publisher setup:", err)
			return
		}
		defer con.Close()

		// Purge the queue from the publisher side to establish initial state
		if _, err := pub.QueuePurge(queue, false); err != nil {
			fmt.Println("err purge:", err)
			return
		}

		// Reconnect simulation, should be for { ... } in production
		for i := 1; i <= 3; i++ {
			fmt.Println("connect")

			conn, deliveries, err := consume(url, queue)
			if err != nil {
				fmt.Println("err consume:", err)
				return
			}

			// Simulate a producer on a different connection showing that consumers
			// continue where they were left off after each reconnect.
			if err := pub.PublishWithContext(context.TODO(), "", queue, false, false, amqp.Publishing{
				Body: []byte(fmt.Sprintf("%d", i)),
			}); err != nil {
				fmt.Println("err publish:", err)
				return
			}

			// Simulates a consumer that when the range finishes, will setup a new
			// session and begin ranging over the deliveries again.
			for msg := range deliveries {
				fmt.Println(string(msg.Body))
				if e := msg.Ack(false); e != nil {
					fmt.Println("ack error: ", e)
				}

				// Simulate an error like a server restart, loss of route or operator
				// intervention that results in the connection terminating
				go conn.Close()
			}
		}
	} else {
		// pass with expected output when not running in an integration
		// environment.
		fmt.Println("connect")
		fmt.Println("1")
		fmt.Println("connect")
		fmt.Println("2")
		fmt.Println("connect")
		fmt.Println("3")
	}

}

Output:

connect
1
connect
2
connect
3

func Dial

func Dial(url string) (*Connection, error)

Dial accepts a string in the AMQP URI format and returns a new Connection over TCP using PlainAuth. Defaults to a server heartbeat interval of 10 seconds and sets the handshake deadline to 30 seconds. After handshake, deadlines are cleared.

Dial uses the zero value of tls.Config when it encounters an amqps:// scheme. It is equivalent to calling DialTLS(amqp, nil).

func DialConfig

func DialConfig(url string, config Config) (*Connection, error)

DialConfig accepts a string in the AMQP URI format and a configuration for the transport and connection setup, returning a new Connection. Defaults to a server heartbeat interval of 10 seconds and sets the initial read deadline to 30 seconds. The heartbeat interval specified in the AMQP URI takes precedence over the value specified in the config. To disable heartbeats, you must use the AMQP URI and set heartbeat=0 there.

func DialTLS

func DialTLS(url string, amqps *tls.Config) (*Connection, error)

DialTLS accepts a string in the AMQP URI format and returns a new Connection over TCP using PlainAuth. Defaults to a server heartbeat interval of 10 seconds and sets the initial read deadline to 30 seconds.

DialTLS uses the provided tls.Config when encountering an amqps:// scheme.

Example

package main

import (
	"crypto/tls"
	"crypto/x509"
	"log"
	"os"

	amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
	// This example assume you have a RabbitMQ node running on localhost
	// with TLS enabled.
	//
	// The easiest way to create the CA, certificates and keys required for these
	// examples is by using tls-gen: https://github.com/michaelklishin/tls-gen
	//
	// A comprehensive RabbitMQ TLS guide can be found at
	// http://www.rabbitmq.com/ssl.html
	//
	// Once you have the required TLS files in place, use the following
	// rabbitmq.config example for the RabbitMQ node that you will run on
	// localhost:
	//
	//   [
	//   {rabbit, [
	//     {tcp_listeners, []},     % listens on 127.0.0.1:5672
	//     {ssl_listeners, [5671]}, % listens on 0.0.0.0:5671
	//     {ssl_options, [{cacertfile,"/path/to/your/testca/cacert.pem"},
	//                    {certfile,"/path/to/your/server/cert.pem"},
	//                    {keyfile,"/path/to/your/server/key.pem"},
	//                    {verify,verify_peer},
	//                    {fail_if_no_peer_cert,true}]}
	//     ]}
	//   ].
	//
	//
	// In the above rabbitmq.config example, we are disabling the plain AMQP port
	// and verifying that clients and fail if no certificate is presented.
	//
	// The self-signing certificate authority's certificate (cacert.pem) must be
	// included in the RootCAs to be trusted, otherwise the server certificate
	// will fail certificate verification.
	//
	// Alternatively to adding it to the tls.Config. you can add the CA's cert to
	// your system's root CAs.  The tls package will use the system roots
	// specific to each support OS.  Under OS X, add (drag/drop) cacert.pem
	// file to the 'Certificates' section of KeyChain.app to add and always
	// trust.  You can also add it via the command line:
	//
	//   security add-certificate testca/cacert.pem
	//   security add-trusted-cert testca/cacert.pem
	//
	// If you depend on the system root CAs, then use nil for the RootCAs field
	// so the system roots will be loaded instead.
	//
	// Server names are validated by the crypto/tls package, so the server
	// certificate must be made for the hostname in the URL.  Find the commonName
	// (CN) and make sure the hostname in the URL matches this common name.  Per
	// the RabbitMQ instructions (or tls-gen) for a self-signed cert, this defaults to the
	// current hostname.
	//
	//   openssl x509 -noout -in /path/to/certificate.pem -subject
	//
	// If your server name in your certificate is different than the host you are
	// connecting to, set the hostname used for verification in
	// ServerName field of the tls.Config struct.
	cfg := new(tls.Config)

	// see at the top
	cfg.RootCAs = x509.NewCertPool()

	if ca, err := os.ReadFile("testca/cacert.pem"); err == nil {
		cfg.RootCAs.AppendCertsFromPEM(ca)
	}

	// Move the client cert and key to a location specific to your application
	// and load them here.

	if cert, err := tls.LoadX509KeyPair("client/cert.pem", "client/key.pem"); err == nil {
		cfg.Certificates = append(cfg.Certificates, cert)
	}

	// see a note about Common Name (CN) at the top
	conn, err := amqp.DialTLS("amqps://server-name-from-certificate/", cfg)

	log.Printf("conn: %v, err: %v", conn, err)
}

func DialTLS_ExternalAuth

func DialTLS_ExternalAuth(url string, amqps *tls.Config) (*Connection, error)

DialTLS_ExternalAuth accepts a string in the AMQP URI format and returns a new Connection over TCP using EXTERNAL auth. Defaults to a server heartbeat interval of 10 seconds and sets the initial read deadline to 30 seconds.

This mechanism is used, when RabbitMQ is configured for EXTERNAL auth with ssl_cert_login plugin for userless/passwordless logons

DialTLS_ExternalAuth uses the provided tls.Config when encountering an amqps:// scheme.

func Open

func Open(conn io.ReadWriteCloser, config Config) (*Connection, error)

Open accepts an already established connection, or other io.ReadWriteCloser as a transport. Use this method if you have established a TLS connection or wish to use your own custom transport.

func (*Connection) Channel

func (c *Connection) Channel() (*Channel, error)

Channel opens a unique, concurrent server channel to process the bulk of AMQP messages. Any error from methods on this receiver will render the receiver invalid and a new Channel should be opened.

func (*Connection) Close

func (c *Connection) Close() error

Close requests and waits for the response to close the AMQP connection.

It's advisable to use this message when publishing to ensure all kernel buffers have been flushed on the server and client before exiting.

An error indicates that server may not have received this request to close but the connection should be treated as closed regardless.

After returning from this call, all resources associated with this connection, including the underlying io, Channels, Notify listeners and Channel consumers will also be closed.

func (*Connection) CloseDeadline

func (c *Connection) CloseDeadline(deadline time.Time) error

CloseDeadline requests and waits for the response to close this AMQP connection.

Accepts a deadline for waiting the server response. The deadline is passed to the low-level connection i.e. network socket.

Regardless of the error returned, the connection is considered closed, and it should not be used after calling this function.

In the event of an I/O timeout, connection-closed listeners are NOT informed.

After returning from this call, all resources associated with this connection, including the underlying io, Channels, Notify listeners and Channel consumers will also be closed.

func (*Connection) ConnectionState

func (c *Connection) ConnectionState() tls.ConnectionState

ConnectionState returns basic TLS details of the underlying transport. Returns a zero value when the underlying connection does not implement ConnectionState() tls.ConnectionState.

func (*Connection) IsClosed

func (c *Connection) IsClosed() bool

IsClosed returns true if the connection is marked as closed, otherwise false is returned.

func (*Connection) LocalAddr

func (c *Connection) LocalAddr() net.Addr

LocalAddr returns the local TCP peer address, or ":0" (the zero value of net.TCPAddr) as a fallback default value if the underlying transport does not support LocalAddr().

func (*Connection) NotifyBlocked

func (c *Connection) NotifyBlocked(receiver chan Blocking) chan Blocking

NotifyBlocked registers a listener for RabbitMQ specific TCP flow control method extensions connection.blocked and connection.unblocked. Flow control is active with a reason when Blocking.Blocked is true. When a Connection is blocked, all methods will block across all connections until server resources become free again.

This optional extension is supported by the server when the "connection.blocked" server capability key is true.

Example

package main

import (
	"log"

	amqp "github.com/rabbitmq/amqp091-go"
)

func publishAllTheThings(conn *amqp.Connection) {

}

func main() {
	// Simply logs when the server throttles the TCP connection for publishers

	// Test this by tuning your server to have a low memory watermark:
	// rabbitmqctl set_vm_memory_high_watermark 0.00000001

	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	if err != nil {
		log.Fatalf("connection.open: %s", err)
	}
	defer conn.Close()

	blockings := conn.NotifyBlocked(make(chan amqp.Blocking))
	go func() {
		for b := range blockings {
			if b.Active {
				log.Printf("TCP blocked: %q", b.Reason)
			} else {
				log.Printf("TCP unblocked")
			}
		}
	}()

	// Your application domain channel setup publishings
	publishAllTheThings(conn)
}

func (*Connection) NotifyClose

func (c *Connection) NotifyClose(receiver chan *Error) chan *Error

NotifyClose registers a listener for close events either initiated by an error accompanying a connection.close method or by a normal shutdown.

The chan provided will be closed when the Connection is closed and on a graceful close, no error will be sent.

In case of a non graceful close the error will be notified synchronously by the library so that it will be necessary to consume the Channel from the caller in order to avoid deadlocks

To reconnect after a transport or protocol error, register a listener here and re-run your setup process.

func (*Connection) RemoteAddr

func (c *Connection) RemoteAddr() net.Addr

RemoteAddr returns the remote TCP peer address, if known.

func (*Connection) UpdateSecret

func (c *Connection) UpdateSecret(newSecret, reason string) error

UpdateSecret updates the secret used to authenticate this connection. It is used when secrets have an expiration date and need to be renewed, like OAuth 2 tokens.

It returns an error if the operation is not successful, or if the connection is closed.

Example

package main

import (
	"context"
	"fmt"

	amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
	// In order to authenticate into RabbitMQ, the application must acquire a JWT token.
	// This may be different, depending on the library used to communicate with the OAuth2
	// server. This examples assumes that it's possible to obtain tokens using username+password.
	//
	// The authentication is successful if RabbitMQ can validate the JWT with the OAuth2 server.
	// The permissions are determined from the scopes. Check the OAuth2 plugin readme for more details:
	// https://github.com/rabbitmq/rabbitmq-server/tree/main/deps/rabbitmq_auth_backend_oauth2#scope-to-permission-translation
	//
	// Once the app has a JWT token, this can be used as credentials in the URI used in Connection.Dial()
	//
	// The app should have a long-running task that checks the validity of the JWT token, and renew it before
	// the refresher time expires. Once a new JWT token is obtained, it shall be used in Connection.UpdateSecret().

	token, _ := getJWToken("username", "password")

	uri := fmt.Sprintf("amqp://%s:%s@localhost:5672", "client_id", token)
	c, _ := amqp.Dial(uri)

	defer c.Close()

	// It also calls Connection.UpdateSecret()
	tokenRefresherTask := func(conn *amqp.Connection, token string) {
		// if token is expired
		// then
		renewedToken, _ := refreshJWToken(token)
		_ = conn.UpdateSecret(renewedToken, "Token refreshed!")
	}

	go tokenRefresherTask(c, "my-JWT-token")

	ch, _ := c.Channel()
	defer ch.Close()

	_, _ = ch.QueueDeclare(
		"test-amqp",
		false,
		false,
		false,
		false,
		amqp.Table{},
	)
	_ = ch.PublishWithContext(
		context.Background(),
		"",
		"test-amqp",
		false,
		false,
		amqp.Publishing{
			Headers:         amqp.Table{},
			ContentType:     "text/plain",
			ContentEncoding: "",
			DeliveryMode:    amqp.Persistent,
			Body:            []byte("message"),
		},
	)
}

func getJWToken(username, password string) (string, error) {

	return "a-token", nil
}

func refreshJWToken(token string) (string, error) {

	return "so fresh!", nil
}

type Decimal

type Decimal struct {
	Scale uint8
	Value int32
}

Decimal matches the AMQP decimal type. Scale is the number of decimal digits Scale == 2, Value == 12345, Decimal == 123.45

type DeferredConfirmation

type DeferredConfirmation struct {
	DeliveryTag uint64
	// contains filtered or unexported fields
}

DeferredConfirmation represents a future publisher confirm for a message. It allows users to directly correlate a publishing to a confirmation. These are returned from PublishWithDeferredConfirm on Channels.

func (*DeferredConfirmation) Acked

func (d *DeferredConfirmation) Acked() bool

Acked returns the publisher confirmation in a non-blocking manner. It returns false if the confirmation was not acknowledged yet or received negative acknowledgement.

func (*DeferredConfirmation) Done

func (d *DeferredConfirmation) Done() <-chan struct{}

Done returns the channel that can be used to wait for the publisher confirmation.

func (*DeferredConfirmation) Wait

func (d *DeferredConfirmation) Wait() bool

Wait blocks until the publisher confirmation. It returns true if the server successfully received the publishing.

func (*DeferredConfirmation) WaitContext

func (d *DeferredConfirmation) WaitContext(ctx context.Context) (bool, error)

WaitContext waits until the publisher confirmation. It returns true if the server successfully received the publishing. If the context expires before that, ctx.Err() is returned.

type Delivery

type Delivery struct {
	Acknowledger Acknowledger // the channel from which this delivery arrived

	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.

func (Delivery) Ack

func (d Delivery) Ack(multiple bool) error

Ack delegates an acknowledgement through the Acknowledger interface that the client or server has finished work on a delivery.

All deliveries in AMQP must be acknowledged. If you called Channel.Consume with autoAck true then the server will be automatically ack each message and this method should not be called. Otherwise, you must call Delivery.Ack after you have successfully processed this delivery.

When multiple is true, this delivery and all prior unacknowledged deliveries on the same channel will be acknowledged. This is useful for batch processing of deliveries.

An error will indicate that the acknowledge could not be delivered to the channel it was sent from.

Either Delivery.Ack, Delivery.Reject or Delivery.Nack must be called for every delivery that is not automatically acknowledged.

func (Delivery) Nack

func (d Delivery) Nack(multiple, requeue bool) error

Nack negatively acknowledge the delivery of message(s) identified by the delivery tag from either the client or server.

When multiple is true, nack messages up to and including delivered messages up until the delivery tag delivered on the same channel.

When requeue is true, request the server to deliver this message to a different consumer. If it is not possible or requeue is false, the message will be dropped or delivered to a server configured dead-letter queue.

This method must not be used to select or requeue messages the client wishes not to handle, rather it is to inform the server that the client is incapable of handling this message at this time.

Either Delivery.Ack, Delivery.Reject or Delivery.Nack must be called for every delivery that is not automatically acknowledged.

func (Delivery) Reject

func (d Delivery) Reject(requeue bool) error

Reject delegates a negatively acknowledgement through the Acknowledger interface.

When requeue is true, queue this message to be delivered to a consumer on a different channel. When requeue is false or the server is unable to queue this message, it will be dropped.

If you are batch processing deliveries, and your server supports it, prefer Delivery.Nack.

Either Delivery.Ack, Delivery.Reject or Delivery.Nack must be called for every delivery that is not automatically acknowledged.

type Error

type Error struct {
	Code    int    // constant code from the specification
	Reason  string // description of the error
	Server  bool   // true when initiated from the server, false when from this library
	Recover bool   // true when this error can be recovered by retrying later or with different parameters
}

Error captures the code and reason a channel or connection has been closed by the server.

func (Error) Error

func (e Error) Error() string

type ExternalAuth

type ExternalAuth struct {
}

ExternalAuth for RabbitMQ-auth-mechanism-ssl.

func (*ExternalAuth) Mechanism

func (*ExternalAuth) Mechanism() string

Mechanism returns "EXTERNAL"

func (*ExternalAuth) Response

func (*ExternalAuth) Response() string

Response returns an AMQP encoded credentials table, without the field table size.

type Logging

type Logging interface {
	Printf(format string, v ...interface{})
}

var Logger Logging = NullLogger{}

type NullLogger

type NullLogger struct {
}

func (NullLogger) Printf

func (l NullLogger) Printf(format string, v ...interface{})

type PlainAuth

type PlainAuth struct {
	Username string
	Password string
}

PlainAuth is a similar to Basic Auth in HTTP.

func (*PlainAuth) Mechanism

func (auth *PlainAuth) Mechanism() string

Mechanism returns "PLAIN"

func (*PlainAuth) Response

func (auth *PlainAuth) Response() string

Response returns the null character delimited encoding for the SASL PLAIN Mechanism.

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  // Transient (0 or 1) or Persistent (2)
	Priority        uint8  // 0 to 9
	CorrelationId   string // correlation identifier
	ReplyTo         string // address to to reply to (ex: RPC)
	// Expiration represents the message TTL in milliseconds. A value of "0"
	// indicates that the message will immediately expire if the message arrives
	// at its destination and the message is not directly handled by a consumer
	// that currently has the capacatity to do so. If you wish the message to
	// not expire on its own, set this value to any ttl value, empty string or
	// use the corresponding constants NeverExpire and ImmediatelyExpire. This
	// does not influence queue configured TTL values.
	Expiration string
	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
}

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

type Return struct {
	ReplyCode  uint16 // reason
	ReplyText  string // description
	Exchange   string // basic.publish exchange
	RoutingKey string // basic.publish routing key

	// Properties
	ContentType     string    // MIME content type
	ContentEncoding string    // MIME content encoding
	Headers         Table     // Application or header exchange table
	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 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 id
	AppId           string    // application use - creating application

	Body []byte
}

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

func NewConnectionProperties() Table

NewConnectionProperties creates an amqp.Table to be used as amqp.Config.Properties.

Defaults to library-defined values. For empty properties, use make(amqp.Table) instead.

func (Table) SetClientConnectionName

func (t Table) SetClientConnectionName(connName string)

Sets the connection name property. This property can be used in amqp.Config to set a custom connection name during amqp.DialConfig(). This can be helpful to identify specific connections in RabbitMQ, for debugging or tracing purposes.

Example

package main

import (
	"log"

	amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
	// Sets the well-known connection_name property in amqp.Config. The connection
	// name will be visible in RabbitMQ Management UI.
	config := amqp.Config{Properties: amqp.NewConnectionProperties()}
	config.Properties.SetClientConnectionName("my-client-app")
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	if err != nil {
		log.Fatalf("connection.open: %s", err)
	}
	defer conn.Close()
}

func (Table) Validate

func (t Table) Validate() error

Validate returns and error if any Go types in the table are incompatible with AMQP types.

type URI

type URI struct {
	Scheme            string
	Host              string
	Port              int
	Username          string
	Password          string
	Vhost             string
	CertFile          string // client TLS auth - path to certificate (PEM)
	CACertFile        string // client TLS auth - path to CA certificate (PEM)
	KeyFile           string // client TLS auth - path to private key (PEM)
	ServerName        string // client TLS auth - server name
	AuthMechanism     []string
	Heartbeat         heartbeatDuration
	ConnectionTimeout int
	ChannelMax        uint16
}

URI represents a parsed AMQP URI string.

func ParseURI

func ParseURI(uri string) (URI, error)

ParseURI attempts to parse the given AMQP URI according to the spec. See http://www.rabbitmq.com/uri-spec.html.

Default values for the fields are:

Scheme: amqp
Host: localhost
Port: 5672
Username: guest
Password: guest
Vhost: /

Supports TLS query parameters. See https://www.rabbitmq.com/uri-query-parameters.html

certfile: <path/to/client_cert.pem>
keyfile: <path/to/client_key.pem>
cacertfile: <path/to/ca.pem>
server_name_indication: <server name>
auth_mechanism: <one or more: plain, amqplain, external>
heartbeat: <seconds (integer)>
connection_timeout: <milliseconds (integer)>
channel_max: <max number of channels (integer)>

If cacertfile is not provided, system CA certificates will be used. Mutual TLS (client auth) will be enabled only in case keyfile AND certfile provided.

If Config.TLSClientConfig is set, TLS parameters from URI will be ignored.

func (URI) AMQPlainAuth

func (uri URI) AMQPlainAuth() *AMQPlainAuth

AMQPlainAuth returns a PlainAuth structure based on the parsed URI's Username and Password fields.

func (URI) PlainAuth

func (uri URI) PlainAuth() *PlainAuth

PlainAuth returns a PlainAuth structure based on the parsed URI's Username and Password fields.

func (URI) String

func (uri URI) String() string