amqp

package module
v1.1.0 Latest Latest
Warning

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

 Go to latest Published: Jun 19, 2023 License: BSD-2-Clause Imports: 18 Imported by: 7,733

README

Build Status GoDoc

Go RabbitMQ Client Library (Unmaintained Fork)

Beware of Abandonware

This repository is NOT ACTIVELY MAINTAINED. Consider using a different fork instead: rabbitmq/amqp091-go. In case of questions, start a discussion in that repo or use other RabbitMQ community resources.

Project Maturity

This project has been used in production systems for many years. As of 2022, this repository is NOT ACTIVELY MAINTAINED.

This repository is very strict about any potential public API changes. You may want to consider rabbitmq/amqp091-go which is more willing to adapt the API.

Supported Go Versions

This library supports two most recent Go release series, currently 1.10 and 1.11.

Supported RabbitMQ Versions

This project supports RabbitMQ versions starting with 2.0 but primarily tested against reasonably recent 3.x releases. 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. Versions 0.10 and 1.0 are divergent specifications that change the semantics and wire format of the protocol. We will accept patches for other protocol support but have no plans for implementation ourselves.
  • 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.

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

Use Godoc documentation for reference and usage.

RabbitMQ tutorials in Go are also available.

Contributing

Pull requests are very much welcomed. Create your pull request on a non-master 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.

To run the integration tests, make sure you have RabbitMQ running on any host, export the environment variable AMQP_URL=amqp://host/ and run go test -tags integration. TravisCI will also run the integration tests.

Thanks to the community of contributors.

External packages

License

BSD 2 clause - see LICENSE for more details.

Documentation

Overview

Package amqp 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 sychronous 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.Publish 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 dial your own connection. 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

Example

This exports a Session 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 (
	"errors"
	"fmt"
	"log"
	"os"
	"time"

	"github.com/streadway/amqp"
)

// This exports a Session 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.
func main() {
	name := "job_queue"
	addr := "amqp://guest:guest@localhost:5672/"
	queue := New(name, addr)
	message := []byte("message")
	// Attempt to push a message every 2 seconds
	for {
		time.Sleep(time.Second * 2)
		if err := queue.Push(message); err != nil {
			fmt.Printf("Push failed: %s\n", err)
		} else {
			fmt.Println("Push succeeded!")
		}
	}
}

type Session struct {
	name            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 (
	// When reconnecting to the server after connection failure
	reconnectDelay = 5 * time.Second

	// When setting up the channel after a channel exception
	reInitDelay = 2 * time.Second

	// When resending messages the server didn't confirm
	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("session is shutting down")
)

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

// handleReconnect will wait for a connection error on
// notifyConnClose, and then continuously attempt to reconnect.
func (session *Session) handleReconnect(addr string) {
	for {
		session.isReady = false
		log.Println("Attempting to connect")

		conn, err := session.connect(addr)

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

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

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

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

	if err != nil {
		return nil, err
	}

	session.changeConnection(conn)
	log.Println("Connected!")
	return conn, nil
}

// handleReconnect will wait for a channel error
// and then continuously attempt to re-initialize both channels
func (session *Session) handleReInit(conn *amqp.Connection) bool {
	for {
		session.isReady = false

		err := session.init(conn)

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

			select {
			case <-session.done:
				return true
			case <-time.After(reInitDelay):
			}
			continue
		}

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

// init will initialize channel & declare queue
func (session *Session) 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(
		session.name,
		false, // Durable
		false, // Delete when unused
		false, // Exclusive
		false, // No-wait
		nil,   // Arguments
	)

	if err != nil {
		return err
	}

	session.changeChannel(ch)
	session.isReady = true
	log.Println("Setup!")

	return nil
}

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

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

// Push will push data onto the queue, and wait for a confirm.
// If no confirms are received until within the resendTimeout,
// it continuously re-sends messages until a confirm is received.
// This will block until the server sends a confirm. Errors are
// only returned if the push action itself fails, see UnsafePush.
func (session *Session) Push(data []byte) error {
	if !session.isReady {
		return errors.New("failed to push: not connected")
	}
	for {
		err := session.UnsafePush(data)
		if err != nil {
			session.logger.Println("Push failed. Retrying...")
			select {
			case <-session.done:
				return errShutdown
			case <-time.After(resendDelay):
			}
			continue
		}
		select {
		case confirm := <-session.notifyConfirm:
			if confirm.Ack {
				session.logger.Println("Push confirmed!")
				return nil
			}
		case <-time.After(resendDelay):
		}
		session.logger.Println("Push didn't confirm. Retrying...")
	}
}

// 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
// recieve the message.
func (session *Session) UnsafePush(data []byte) error {
	if !session.isReady {
		return errNotConnected
	}
	return session.channel.Publish(
		"",           // Exchange
		session.name, // Routing key
		false,        // Mandatory
		false,        // Immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        data,
		},
	)
}

// Stream 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 (session *Session) Stream() (<-chan amqp.Delivery, error) {
	if !session.isReady {
		return nil, errNotConnected
	}
	return session.channel.Consume(
		session.name,
		"",    // Consumer
		false, // Auto-Ack
		false, // Exclusive
		false, // No-local
		false, // No-Wait
		nil,   // Args
	)
}

// Close will cleanly shutdown the channel and connection.
func (session *Session) Close() error {
	if !session.isReady {
		return errAlreadyClosed
	}
	err := session.channel.Close()
	if err != nil {
		return err
	}
	err = session.connection.Close()
	if err != nil {
		return err
	}
	close(session.done)
	session.isReady = false
	return nil
}

Output:

Index

Examples

Constants

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

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

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

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

Variables

View Source 
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 negoated.
	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

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 the null character delimited encoding for the SASL PLAIN Mechanism.

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 delivieries 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 (
	"log"

	"github.com/streadway/amqp"
)

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)
	}

	// 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.Publish("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 {
			msg.Nack(false, false)
			log.Fatalf("basic.publish destination: %+v", msg)
		}

		// only ack the source delivery when the destination acks the publishing
		if confirmed := <-confirms; confirmed.Ack {
			msg.Ack(false)
		} else {
			msg.Nack(false, false)
		}
	}
}

Output:

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.

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"

	"github.com/streadway/amqp"
)

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 log := range pages {
			// ... this consumer is responsible for sending pages per log
			log.Ack(false)
		}
	}()

	// 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 log := range emails {
			// ... this consumer is responsible for sending emails per log
			log.Ack(false)
		}
	}()

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

Output:

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

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

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.

Example 
package main

import (
	"log"
	"time"

	"github.com/streadway/amqp"
)

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!"),
	}

	// This is not a mandatory delivery, so it will be dropped if there are no
	// queues bound to the logs exchange.
	err = c.Publish("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)
	}
}

Output:

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.

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.

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.

It is possible to send and empty string for the exchange name which means to unbind the queue from the default exchange.

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

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 int           // 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"

	"github.com/streadway/amqp"
)

func main() {
	// Provide your own anonymous Dial function that delgates 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)
}

Output:

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 (
	"fmt"
	"github.com/streadway/amqp"
	"os"
)

// 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.
		_, pub, err := setup(url, queue)
		if err != nil {
			fmt.Println("err publisher setup:", err)
			return
		}

		// 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.Publish("", 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))
				msg.Ack(false)

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

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"
	"io/ioutil"
	"log"

	"github.com/streadway/amqp"
)

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 := ioutil.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)
}

Output:

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) 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"

	"github.com/streadway/amqp"
)

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)
}

Output:

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.

On normal shutdowns, the chan will be closed.

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

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 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 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      string    // message expiration spec
	MessageId       string    // message identifier
	Timestamp       time.Time // message timestamp
	Type            string    // message type name
	UserId          string    // creating user id - ex: "guest"
	AppId           string    // creating application id

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

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
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 (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
}

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: /

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

Source Files

View all Source files

Directories

Path Synopsis
_examples
pubsub
simple-consumer
This example declares a durable Exchange, an ephemeral (auto-delete) Queue, binds the Queue to the Exchange with a binding key, and consumes every message published to that Exchange with that routing key.
 This example declares a durable Exchange, an ephemeral (auto-delete) Queue, binds the Queue to the Exchange with a binding key, and consumes every message published to that Exchange with that routing key.
simple-producer
This example declares a durable Exchange, and publishes a single message to that Exchange with a given routing key.
 This example declares a durable Exchange, and publishes a single message to that Exchange with a given routing key.

Jump to

Keyboard shortcuts

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