Help us shape the future of Go! Take the 2026 Go Developer Survey

amqp

package module
v1.2.1 Latest Latest
Warning

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

 Go to latest Published: Oct 6, 2026 License: MIT Imports: 29 Imported by: 0

README

amqp-client.go

A fast AMQP 0-9-1 client for Go, without dependencies. It works with any AMQP 0-9-1 broker, such as LavinMQ and RabbitMQ, self-hosted or managed by CloudAMQP.

It's a sibling of amqp-client.rb, amqp-client.cr and amqp-client.js, with the same two APIs:

  • High-level (Client): reconnects automatically and recovers exchanges, queues, bindings and consumers. Safe by default: messages are persistent and confirmed by the broker, and consumers acknowledge a message after its handler has processed it.
  • Low-level (Connection and Channel): methods that map one-to-one to the protocol, for full control.

Both are idiomatic Go: every blocking call takes a context.Context, errors work with errors.Is/errors.As, and everything is safe for concurrent use.

API reference

Install

go get github.com/cloudamqp/amqp-client.go

Go 1.24 or later is required.

High-level API

import amqp "github.com/cloudamqp/amqp-client.go"

ctx := context.Background()

// Connects, and reconnects if the connection is lost
c, err := amqp.NewClient(ctx, "amqp://guest:guest@localhost", nil)
if err != nil {
	log.Fatal(err)
}
defer c.Close()

// A durable queue (a queue named "" is server-named, exclusive and auto-delete)
q, err := c.Queue(ctx, "jobs", nil)

// A durable topic exchange, and a binding that is recovered after reconnects
events, err := c.TopicExchange(ctx, "events")
err = q.Bind(ctx, events.Name(), "jobs.#", nil)

// Calls the handler for each message. The message is acked when the handler
// returns nil, and rejected and requeued when it returns an error or panics.
// The handler can also ack/reject the message itself.
sub, err := q.Subscribe(ctx, func(ctx context.Context, d *amqp.Delivery) error {
	log.Printf("received %s", d.Body)
	return nil
}, &amqp.SubscribeOptions{Prefetch: 50, Workers: 4})
defer sub.Cancel(ctx)

// Publishes a persistent message and waits until the broker confirms it
err = q.Publish(ctx, amqp.Message{Body: "hello"})
err = events.Publish(ctx, "jobs.created", amqp.Message{
	Body:       `{"id": 1}`,
	Properties: amqp.Properties{ContentType: "application/json", Headers: amqp.Table{"retries": 0}},
})

Publish returns once the broker has confirmed the message. Publish from several goroutines to get a higher throughput, because the confirms are pipelined on a shared channel. The first publish to an exchange the client hasn't declared checks that it exists, so that publishing to a missing exchange fails on its own instead of closing the shared channel for every publish in flight.

While the client is reconnecting, operations wait until the connection is back or their context is done. If the channel or connection is lost after a message is sent but before it's confirmed, Publish returns an error that matches amqp.ErrUnconfirmed. The message may or may not have reached the broker.

Reconnection
c, err := amqp.NewClient(ctx, url, &amqp.ClientOptions{
	ReconnectInterval:    time.Second,      // doubled after each failed attempt...
	MaxReconnectInterval: 30 * time.Second, // ...up to this
	MaxRetries:           0,                // give up after N failed attempts, 0 = never
	OnConnect: func(ctx context.Context, c *amqp.Client) error {
		return nil // runs after every (re)connection, after recovery
	},
	OnDisconnect: func(err error) { log.Println("disconnected:", err) },
	OnFailed:     func(err error) { log.Println("gave up:", err) },
})

Topology that OnConnect declares with its context isn't recovered, because OnConnect declares it again on every connection. Subscriptions it starts end with their connection, and OnConnect starts them again.

Each reconnect attempt times out after Config.ConnectTimeout (30s by default). A connection that the broker closes soon after it's established, for example because of a connection limit, counts as a failed attempt, so the backoff and MaxRetries apply.

When the client reconnects, it declares exchanges, queues and bindings again. A server-named queue gets a new name, and Queue.Name() returns the new one. Subscriptions consume again, also after their channel is closed while the connection stays up, for example after a consumer timeout. If the broker has deleted the queue in the meantime (an auto-delete queue is deleted when its last consumer goes away), it's declared again with its bindings. Messages that were in process when the connection was lost are redelivered, so make handlers idempotent.

Codecs

Opt in to codecs to have message bodies serialized and compressed according to their content-type and content-encoding:

c, err := amqp.NewClient(ctx, url, &amqp.ClientOptions{
	Config:                 amqp.Config{Codecs: amqp.DefaultCodecs()}, // JSON, text, gzip, deflate
	DefaultContentType:     "application/json",
	DefaultContentEncoding: "gzip",
})

// Serialized as JSON and gzip compressed
err = q.Publish(ctx, amqp.Message{Body: Order{ID: 1}})

_, err = q.Subscribe(ctx, func(ctx context.Context, d *amqp.Delivery) error {
	var o Order
	if err := d.Decode(&o); err != nil { // gunzip and unmarshal
		d.Reject(false) // can't be processed, dead-letter it instead of requeuing
		return err
	}
	return handle(o)
}, nil)

Register your own codecs with Codecs.RegisterSerializer (by content type) and Codecs.RegisterEncoder (by content encoding). The built-in gzip and deflate encoders refuse to decode more than 128 MiB, to protect against compression bombs; use amqp.NewGzip(limit) or amqp.NewDeflate(limit) for another limit.

RPC

Request/reply over direct reply-to:

srv, err := c.RPCServer(ctx, "upcase", func(ctx context.Context, req *amqp.Delivery) (amqp.Message, error) {
	return amqp.Message{Body: strings.ToUpper(string(req.Body))}, nil
}, &amqp.SubscribeOptions{Workers: 8})

ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
reply, err := c.RPCCall(ctx, "upcase", amqp.Message{Body: "hi"})

RPCCall returns amqp.ErrUnroutable right away if no queue exists for the request. If you make many calls, use your own RPCClient from c.RPCClient(ctx). Both the servers and the clients recover after reconnects.

Low-level API

The low-level API doesn't recover anything. Methods map one-to-one to the protocol:

conn, err := amqp.Dial(ctx, "amqp://guest:guest@localhost", &amqp.Config{
	ConnectionName: "my-app",
	Heartbeat:      30 * time.Second,
})
defer conn.Close()

ch, err := conn.Channel(ctx)
q, err := ch.QueueDeclare(ctx, "", amqp.QueueDeclareOptions{Exclusive: true, AutoDelete: true})
err = ch.QueueBind(ctx, q.Name, "amq.topic", "a.#", nil)
err = ch.ExchangeDeclare(ctx, "logs", amqp.Fanout, amqp.ExchangeDeclareOptions{Durable: true})

// Consume. Deliveries are buffered per consumer, so always set a prefetch limit
err = ch.BasicQos(ctx, 100, false)
consumer, err := ch.BasicConsume(ctx, q.Name, amqp.ConsumeOptions{})
go func() {
	for {
		d, err := consumer.Next(ctx) // or range over consumer.Deliveries()
		if err != nil {
			return // amqp.ErrConsumerCancelled, a channel or connection error
		}
		d.Ack()
	}
}()

// Fire and forget: returns when the message is buffered for sending
err = ch.BasicPublish(ctx, "amq.topic", "a.b", amqp.Publishing{Body: []byte("hi")})

// Publisher confirms: publish many messages, then wait for their confirms
conf, err := ch.BasicPublishConfirm(ctx, "", q.Name, amqp.Publishing{
	Properties: amqp.Properties{DeliveryMode: amqp.Persistent},
	Body:       []byte("hi"),
})
err = conf.Wait(ctx)         // or wait for everything published so far:
err = ch.WaitForConfirms(ctx)

// Polling
msg, ok, err := ch.BasicGet(ctx, q.Name, false)

// Mandatory messages that can't be routed are returned
ch.OnReturn(func(r *amqp.Return) { log.Printf("returned: %s", r.ReplyText) })

Other methods are QueuePurge, QueueDelete, QueueUnbind, ExchangeDelete, ExchangeBind/Unbind, BasicNack, BasicReject, BasicRecover, BasicCancel, TxSelect/Commit/Rollback, Flow and Connection.UpdateSecret, which renews an OAuth 2 token without reconnecting.

Lifecycles follow the context pattern. Done() returns a channel that's closed when the connection, channel, consumer or subscription stops, and Err() returns the reason.

Errors
_, err := ch.QueueDeclare(ctx, "q", amqp.QueueDeclareOptions{Passive: true})
switch {
case amqp.IsCode(err, amqp.NotFound): // an *amqp.Error from the broker, the channel is now closed
case errors.Is(err, amqp.ErrClosed):  // the channel or connection is closed
}

Broker exceptions are returned as *amqp.Error, with Code, Reason, ClassID/MethodID and whether the whole Connection was closed. All errors that come from a closed channel or connection match amqp.ErrClosed, including network errors.

Performance

The library is designed to avoid allocations and syscalls:

  • Frames are encoded straight into the write buffer. Encoding a publish takes about 90 ns and allocates nothing.
  • A background goroutine flushes the write buffer, so concurrent or rapid publishes and acks are sent in batches with few syscalls. Publishes are still bounded by TCP backpressure.
  • Received frames are parsed in place in the read buffer. Receiving a message allocates only the Delivery and its body; repeated strings such as the exchange, routing key and content type are reused.
  • Each consumer has its own buffer, so a slow consumer never stalls the connection, and synchronous calls on a channel are pipelined.

The table below compares 1-byte messages with no properties, with one goroutine publishing or consuming, against LavinMQ 2.2.0 on the same machine (Intel i9-13900H). Consuming includes acknowledging each message, with a prefetch of 1000.

Client Publish rate Consume rate
amqp-client.go 1 550 000 msgs/s 1 200 000 msgs/s
amqp091-go v1.15.0 200 000 msgs/s 45 000 msgs/s

Run the benchmarks against a broker with go test -run=^$ -bench=. -benchmem.

Development

Most tests need a broker, by default at amqp://guest:guest@localhost:5672. Set AMQP_URL to use another one. If no broker is reachable, those tests are skipped.

go test -race ./...
go test -short ./...  # skips the slow heartbeat tests

License

The library is available as open source under the terms of the MIT License.

Documentation

Overview

Package amqp is a fast AMQP 0-9-1 client without dependencies, for brokers such as LavinMQ and RabbitMQ.

It has two APIs.

High-level API

NewClient returns a Client that keeps a connection to the broker. It reconnects when the connection is lost and then recovers the exchanges, queues, bindings and subscriptions declared through it. Messages are published as persistent and confirmed by the broker by default, and subscriptions acknowledge messages when the handler returns: 

c, err := amqp.NewClient(ctx, "amqp://guest:guest@localhost", nil)
if err != nil {
	return err
}
defer c.Close()

q, err := c.Queue(ctx, "jobs", nil) // durable
if err != nil {
	return err
}
sub, err := q.Subscribe(ctx, func(ctx context.Context, d *amqp.Delivery) error {
	return process(d.Body) // ack on nil, requeue on error
}, &amqp.SubscribeOptions{Prefetch: 50, Workers: 4})
if err != nil {
	return err
}
defer sub.Cancel(ctx)

err = q.Publish(ctx, amqp.Message{Body: "hello"}) // waits for the confirm

With Codecs message bodies are serialized and encoded according to their content type and content encoding, see ClientOptions and Delivery.Decode. Client.RPCServer and Client.RPCCall implement request/reply over direct reply-to.

Low-level API

Dial returns a Connection and Connection.Channel opens a Channel with methods that map one-to-one to the protocol, e.g. Channel.QueueDeclare, Channel.BasicPublish and Channel.BasicConsume. Nothing is recovered if the connection is lost: 

conn, err := amqp.Dial(ctx, "amqp://guest:guest@localhost", nil)
if err != nil {
	return err
}
defer conn.Close()
ch, err := conn.Channel(ctx)
if err != nil {
	return err
}
q, err := ch.QueueDeclare(ctx, "", amqp.QueueDeclareOptions{Exclusive: true})
if err != nil {
	return err
}
err = ch.BasicPublish(ctx, "", q.Name, amqp.Publishing{Body: []byte("hi")})

Concurrency and performance

Connections, channels and clients are safe for concurrent use. Synchronous methods called concurrently on one channel are pipelined. Frames are written to a buffer and flushed by a background goroutine, so that concurrent or rapid publishes and acks are sent in batches with few syscalls. Received messages are buffered per consumer, so a slow consumer never stalls the connection; limit the buffer with a prefetch count.

Errors

Exceptions from the broker are returned as *Error, with a reply code such as NotFound or AccessRefused, check them with IsCode. A channel exception closes the channel, a connection exception the connection. All errors caused by a closed channel or connection match ErrClosed with errors.Is.

Example 
package main

import (
	"context"
	"fmt"
	"log"

	amqp "github.com/cloudamqp/amqp-client.go"
)

func main() {
	ctx := context.Background()
	c, err := amqp.NewClient(ctx, "amqp://guest:guest@localhost", nil)
	if err != nil {
		log.Fatal(err)
	}
	defer c.Close()

	// A server-named, exclusive, auto-delete queue
	q, err := c.Queue(ctx, "", nil)
	if err != nil {
		log.Fatal(err)
	}
	received := make(chan string)
	sub, err := q.Subscribe(ctx, func(ctx context.Context, d *amqp.Delivery) error {
		received <- string(d.Body)
		return nil // acknowledges the message
	}, nil)
	if err != nil {
		log.Fatal(err)
	}
	defer sub.Cancel(ctx)

	// Waits until the broker has confirmed the message
	if err := q.Publish(ctx, amqp.Message{Body: "Hello world"}); err != nil {
		log.Fatal(err)
	}
	fmt.Println(<-received)
}

Index

Examples

Constants

View Source 
const (
	Direct  = "direct"
	Fanout  = "fanout"
	Topic   = "topic"
	Headers = "headers"
)

Exchange types

View Source 
const (
	Transient  uint8 = 1
	Persistent uint8 = 2
)

Delivery modes for Properties.DeliveryMode.

View Source 
const (
	ReplySuccess       = 200
	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
)

Reply codes used by the broker when closing a connection or channel, see Error.Code.

View Source 
const DefaultMaxDecodedSize = 128 << 20

DefaultMaxDecodedSize is the decoded size limit of Gzip and Deflate.

View Source 
const Version = "1.2.1"

Version of the library, sent to the broker in the client properties.

Variables

View Source 
var (
	// ErrUnsupportedContentType is returned when no Serializer is
	// registered for a message's content type.
	ErrUnsupportedContentType = errors.New("amqp: unsupported content type")
	// ErrUnsupportedContentEncoding is returned when no Encoder is
	// registered for a message's content encoding.
	ErrUnsupportedContentEncoding = errors.New("amqp: unsupported content encoding")
	// ErrDecodedTooLarge is returned when a compressed body decodes to more
	// than the encoder's limit, see [NewGzip].
	ErrDecodedTooLarge = errors.New("amqp: decoded body exceeds the size limit")
)
View Source 
var (
	// ErrClosed reports that the connection, channel or client is closed.
	// Every error caused by a closed connection or channel matches it with
	// errors.Is, including an [*Error] from the broker and network errors.
	ErrClosed = errors.New("amqp: closed")

	// ErrPublishNacked is returned when the broker negatively acknowledges
	// a published message, i.e. it couldn't take responsibility for it.
	ErrPublishNacked = errors.New("amqp: message nacked by the broker")

	// ErrConsumerCancelled is returned by a consumer after it has been
	// cancelled by the client and all buffered deliveries are read.
	ErrConsumerCancelled = errors.New("amqp: consumer cancelled")

	// ErrConsumerCancelledByServer is returned by a consumer the broker
	// cancelled, e.g. because its queue was deleted.
	ErrConsumerCancelledByServer = errors.New("amqp: consumer cancelled by the broker")

	// ErrUnconfirmed is matched by the error for a published message whose
	// channel closed before the broker confirmed it. The message may or may
	// not have been routed. The error also matches ErrClosed.
	ErrUnconfirmed = errors.New("amqp: channel closed before the message was confirmed")

	// ErrNoConfirmMode is returned when waiting for confirms on a channel
	// that isn't in confirm mode.
	ErrNoConfirmMode = errors.New("amqp: channel is not in confirm mode")

	// ErrChannelMax is returned when all channel ids are in use.
	ErrChannelMax = errors.New("amqp: channel max reached")
)
View Source 
var Deflate = NewDeflate(DefaultMaxDecodedSize)

Deflate encodes bodies with zlib-wrapped deflate (RFC 1950), which is what HTTP's "deflate" content encoding means. It decodes at most DefaultMaxDecodedSize bytes.

View Source 
var ErrAlreadyAcknowledged = errors.New("amqp: delivery already acknowledged or rejected")

ErrAlreadyAcknowledged is returned when acknowledging or rejecting a delivery a second time, or one delivered with no-ack, which would otherwise make the broker close the channel.

View Source 
var ErrUnroutable = errors.New("amqp: message could not be routed to any queue")

ErrUnroutable is returned by an RPC call when no queue received the request, i.e. there's no server for the method.

View Source 
var Gzip = NewGzip(DefaultMaxDecodedSize)

Gzip encodes bodies with gzip (RFC 1952). It decodes at most DefaultMaxDecodedSize bytes, protecting against compression bombs.

Functions

func IsCode 

func IsCode(err error, code uint16) bool

IsCode reports whether err is an *Error with the given reply code.

Example 
package main

import (
	"context"
	"errors"
	"fmt"
	"log"

	amqp "github.com/cloudamqp/amqp-client.go"
)

func main() {
	ctx := context.Background()
	conn, err := amqp.Dial(ctx, "amqp://localhost", nil)
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	ch, err := conn.Channel(ctx)
	if err != nil {
		log.Fatal(err)
	}
	_, err = ch.QueueDeclare(ctx, "might-not-exist", amqp.QueueDeclareOptions{Passive: true})
	switch {
	case amqp.IsCode(err, amqp.NotFound):
		fmt.Println("no such queue") // the channel is closed now
	case errors.Is(err, amqp.ErrClosed):
		fmt.Println("connection lost")
	case err != nil:
		log.Fatal(err)
	}
}

Types

type Channel 

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

Channel is an AMQP channel, a lightweight virtual connection multiplexed over a Connection. It's safe for concurrent use: synchronous methods (declarations, bindings etc.) from multiple goroutines are pipelined, the broker answers them in order.

A channel is closed by the broker if an operation fails, e.g. declaring a queue passively that doesn't exist. Its methods then return an *Error, and a new channel has to be opened.

func (*Channel) BasicAck 

func (ch *Channel) BasicAck(deliveryTag uint64, multiple bool) error

BasicAck acknowledges a delivery, or with multiple all deliveries up to and including deliveryTag.

func (*Channel) BasicCancel 

func (ch *Channel) BasicCancel(ctx context.Context, consumerTag string) error

BasicCancel cancels a consumer. Deliveries already received can still be read from the consumer.

func (*Channel) BasicConsume 

func (ch *Channel) BasicConsume(ctx context.Context, queue string, opts ConsumeOptions) (*Consumer, error)

BasicConsume starts a consumer on a queue. Read deliveries with Consumer.Next or Consumer.Deliveries. Set a prefetch limit with Channel.BasicQos first, as deliveries are buffered in memory until read.

func (*Channel) BasicGet 

func (ch *Channel) BasicGet(ctx context.Context, queue string, noAck bool) (msg *Delivery, ok bool, err error)

BasicGet polls a queue for a message. ok is false if the queue is empty. Unless noAck the message has to be acknowledged.

func (*Channel) BasicNack 

func (ch *Channel) BasicNack(deliveryTag uint64, multiple, requeue bool) error

BasicNack rejects a delivery, or with multiple all deliveries up to and including deliveryTag. With requeue the messages are put back in the queue, otherwise they're dropped or dead-lettered.

func (*Channel) BasicPublish 

func (ch *Channel) BasicPublish(ctx context.Context, exchange, routingKey string, msg Publishing) error

BasicPublish publishes a message to an exchange. It returns when the message is written to the connection's buffer, it's then sent to the broker in the background. If the broker has blocked the connection it waits until it's unblocked or ctx is done.

Use Channel.BasicPublishConfirm to know when the broker has taken responsibility for the message.

func (*Channel) BasicPublishConfirm 

func (ch *Channel) BasicPublishConfirm(ctx context.Context, exchange, routingKey string, msg Publishing) (*Confirmation, error)

BasicPublishConfirm publishes a message and returns a confirmation to wait on. It puts the channel in confirm mode if it isn't already. Waiting for each confirmation in turn limits the throughput to one message per round trip, so publish many messages concurrently (or pipeline them) and wait for the confirmations afterwards.

Example 
package main

import (
	"context"
	"fmt"
	"log"

	amqp "github.com/cloudamqp/amqp-client.go"
)

func main() {
	ctx := context.Background()
	conn, err := amqp.Dial(ctx, "amqp://localhost", nil)
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	ch, err := conn.Channel(ctx)
	if err != nil {
		log.Fatal(err)
	}

	// Publish a batch, then wait for all confirms; waiting for each one in
	// turn would limit the throughput to one message per round trip.
	confirms := make([]*amqp.Confirmation, 0, 1000)
	for i := range 1000 {
		c, err := ch.BasicPublishConfirm(ctx, "amq.topic", "numbers", amqp.Publishing{Body: fmt.Append(nil, i)})
		if err != nil {
			log.Fatal(err)
		}
		confirms = append(confirms, c)
	}
	for _, c := range confirms {
		if err := c.Wait(ctx); err != nil {
			log.Fatal(err)
		}
	}
}

func (*Channel) BasicQos 

func (ch *Channel) BasicQos(ctx context.Context, prefetchCount uint16, global bool) error

BasicQos limits the number of unacknowledged messages delivered to each consumer on the channel (or, with global, shared by all consumers on the channel in RabbitMQ). Zero means unlimited.

func (*Channel) BasicRecover 

func (ch *Channel) BasicRecover(ctx context.Context, requeue bool) error

BasicRecover asks the broker to redeliver all unacknowledged messages on the channel. Not supported by all brokers with requeue=false.

func (*Channel) BasicReject 

func (ch *Channel) BasicReject(deliveryTag uint64, requeue bool) error

BasicReject rejects a single delivery, see Channel.BasicNack.

func (*Channel) Close 

func (ch *Channel) Close() error

Close closes the channel gracefully. Unacknowledged messages are requeued by the broker. It returns nil if the channel is already closed.

func (*Channel) CloseReason 

func (ch *Channel) CloseReason(code uint16, reason string) error

CloseReason closes the channel with a reply code and text, which the broker might log.

func (*Channel) ConfirmSelect 

func (ch *Channel) ConfirmSelect(ctx context.Context) error

ConfirmSelect puts the channel in confirm mode, the broker then acks or nacks every published message.

func (*Channel) Connection 

func (ch *Channel) Connection() *Connection

Connection returns the connection the channel belongs to.

func (*Channel) Done 

func (ch *Channel) Done() <-chan struct{}

Done returns a channel that's closed when the channel is closed.

func (*Channel) Err 

func (ch *Channel) Err() error

Err returns why the channel was closed, or nil if it's open. See Connection.Err.

func (*Channel) ExchangeBind 

func (ch *Channel) ExchangeBind(ctx context.Context, destination, source, routingKey string, args Table) error

ExchangeBind binds the destination exchange to the source exchange, so that messages routed by the source with a matching routing key are routed by the destination too.

func (*Channel) ExchangeDeclare 

func (ch *Channel) ExchangeDeclare(ctx context.Context, name, kind string, opts ExchangeDeclareOptions) error

ExchangeDeclare declares an exchange of the given kind, e.g. Direct, Fanout, Topic, Headers or a plugin type such as "x-consistent-hash".

func (*Channel) ExchangeDelete 

func (ch *Channel) ExchangeDelete(ctx context.Context, name string, ifUnused bool) error

ExchangeDelete deletes an exchange. With ifUnused it's only deleted if it has no bindings.

func (*Channel) ExchangeUnbind 

func (ch *Channel) ExchangeUnbind(ctx context.Context, destination, source, routingKey string, args Table) error

ExchangeUnbind removes an exchange-to-exchange binding.

func (*Channel) Flow 

func (ch *Channel) Flow(ctx context.Context, active bool) error

Flow asks the broker to pause (active=false) or resume deliveries. Not supported by RabbitMQ.

func (*Channel) ID 

func (ch *Channel) ID() uint16

ID returns the channel id.

func (*Channel) IsClosed 

func (ch *Channel) IsClosed() bool

IsClosed reports whether the channel is closed.

func (*Channel) OnReturn 

func (ch *Channel) OnReturn(fn func(*Return))

OnReturn sets a function that's called with messages published with Mandatory that the broker couldn't route. It's called from the connection's read loop, so it must not block or call synchronous methods on the connection. The broker sends the return before the publish confirm, so when a confirmation arrives the return has been processed.

func (*Channel) QueueBind 

func (ch *Channel) QueueBind(ctx context.Context, queue, exchange, routingKey string, args Table) error

QueueBind binds a queue to an exchange. Args are matched against message headers by headers exchanges.

func (*Channel) QueueDeclare 

func (ch *Channel) QueueDeclare(ctx context.Context, name string, opts QueueDeclareOptions) (QueueInfo, error)

QueueDeclare declares a queue, creating it if it doesn't exist. Pass an empty name to let the broker generate a unique name.

func (*Channel) QueueDelete 

func (ch *Channel) QueueDelete(ctx context.Context, queue string, opts QueueDeleteOptions) (uint32, error)

QueueDelete deletes a queue and returns the number of messages it had.

func (*Channel) QueuePurge 

func (ch *Channel) QueuePurge(ctx context.Context, queue string) (uint32, error)

QueuePurge removes all ready messages from a queue and returns how many were removed.

func (*Channel) QueueUnbind 

func (ch *Channel) QueueUnbind(ctx context.Context, queue, exchange, routingKey string, args Table) error

QueueUnbind removes a binding between a queue and an exchange.

func (*Channel) TxCommit 

func (ch *Channel) TxCommit(ctx context.Context) error

TxCommit commits the current transaction.

func (*Channel) TxRollback 

func (ch *Channel) TxRollback(ctx context.Context) error

TxRollback abandons the current transaction.

func (*Channel) TxSelect 

func (ch *Channel) TxSelect(ctx context.Context) error

TxSelect puts the channel in transaction mode.

func (*Channel) WaitForConfirms 

func (ch *Channel) WaitForConfirms(ctx context.Context) error

WaitForConfirms waits until all messages published on the channel before the call are confirmed. It returns ErrPublishNacked if any message was nacked since the previous call, and an error matching ErrUnconfirmed if the channel closed with messages unconfirmed.

type Client 

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

Client is the high-level API. It keeps a connection to the broker, reconnecting when it's lost, and recovers the exchanges, queues, bindings and subscriptions declared through it. Messages are published as persistent and confirmed by the broker by default.

Operations wait (bounded by their context) while the client is reconnecting. A Client is safe for concurrent use.

Example (Codecs) 
package main

import (
	"context"
	"fmt"
	"log"

	amqp "github.com/cloudamqp/amqp-client.go"
)

func main() {
	ctx := context.Background()
	c, err := amqp.NewClient(ctx, "amqp://localhost", &amqp.ClientOptions{
		Config:                 amqp.Config{Codecs: amqp.DefaultCodecs()},
		DefaultContentType:     "application/json",
		DefaultContentEncoding: "gzip",
	})
	if err != nil {
		log.Fatal(err)
	}
	defer c.Close()

	type Order struct {
		ID    int      `json:"id"`
		Items []string `json:"items"`
	}
	q, err := c.Queue(ctx, "orders", nil)
	if err != nil {
		log.Fatal(err)
	}
	// Serialized as JSON and gzip compressed
	if err := q.Publish(ctx, amqp.Message{Body: Order{ID: 1, Items: []string{"book"}}}); err != nil {
		log.Fatal(err)
	}
	_, err = q.Subscribe(ctx, func(ctx context.Context, d *amqp.Delivery) error {
		var o Order
		if err := d.Decode(&o); err != nil {
			d.Reject(false) // can't be processed, don't requeue
			return err
		}
		fmt.Println(o.ID, o.Items)
		return nil
	}, nil)
	if err != nil {
		log.Fatal(err)
	}
}

func NewClient 

func NewClient(ctx context.Context, url string, opts *ClientOptions) (*Client, error)

NewClient connects to the broker at url and returns a client that keeps the connection alive. It fails if the first connection attempt fails. opts may be nil.

func (*Client) Close 

func (c *Client) Close() error

Close stops the client: subscriptions are stopped (waiting for running handlers to return, their context is cancelled), and the connection is closed. Unacknowledged messages are requeued by the broker. As Close waits for handlers it must not be called from one, do that in a new goroutine.

func (*Client) Connection 

func (c *Client) Connection(ctx context.Context) (*Connection, error)

Connection returns the current connection, waiting for the client to reconnect if necessary. Channels opened on it aren't recovered.

func (*Client) DefaultExchange 

func (c *Client) DefaultExchange() *Exchange

DefaultExchange returns the default exchange, which routes messages to the queue named by the routing key.

func (*Client) DirectExchange 

func (c *Client) DirectExchange(ctx context.Context, name string) (*Exchange, error)

DirectExchange declares a durable direct exchange, use "amq.direct" for the built-in one.

func (*Client) Done 

func (c *Client) Done() <-chan struct{}

Done returns a channel that's closed when the client is closed or gave up reconnecting.

func (*Client) Err 

func (c *Client) Err() error

Err returns why the client stopped, nil while it's running.

func (*Client) Exchange 

func (c *Client) Exchange(ctx context.Context, name, kind string, opts *ExchangeOptions) (*Exchange, error)

Exchange declares an exchange of the given kind (Direct, Fanout, Topic, Headers or a plugin type) and returns a handle to it. With nil options the exchange is durable. Exchanges named "amq.*" are built in, they're only checked for existence. The exchange is redeclared when the client reconnects.

func (*Client) FanoutExchange 

func (c *Client) FanoutExchange(ctx context.Context, name string) (*Exchange, error)

FanoutExchange declares a durable fanout exchange, use "amq.fanout" for the built-in one.

func (*Client) HeadersExchange 

func (c *Client) HeadersExchange(ctx context.Context, name string) (*Exchange, error)

HeadersExchange declares a durable headers exchange, use "amq.headers" for the built-in one.

func (*Client) Publish 

func (c *Client) Publish(ctx context.Context, exchange, routingKey string, msg Message) error

Publish publishes a message to an exchange and waits for the broker to confirm it. Messages are persistent unless DeliveryMode is set to Transient. Publish from multiple goroutines for higher throughput, the confirms are pipelined.

If the connection is lost before the message is confirmed an error is returned, the message may or may not have reached the broker.

func (*Client) Queue 

func (c *Client) Queue(ctx context.Context, name string, opts *QueueOptions) (*Queue, error)

Queue declares a queue and returns a handle to it. The queue (and its bindings) are redeclared when the client reconnects.

With nil options a named queue is declared durable, and a queue with an empty name is declared exclusive and auto-delete with a name generated by the broker (which changes on reconnect).

func (*Client) RPCCall 

func (c *Client) RPCCall(ctx context.Context, queue string, msg Message) (*Delivery, error)

RPCCall sends a request to an RPC server consuming from queue and waits for the reply, use a context with a deadline to not wait forever. It uses a shared RPCClient.

Example 
package main

import (
	"context"
	"fmt"
	"log"
	"time"

	amqp "github.com/cloudamqp/amqp-client.go"
)

func main() {
	ctx := context.Background()
	c, err := amqp.NewClient(ctx, "amqp://localhost", nil)
	if err != nil {
		log.Fatal(err)
	}
	defer c.Close()

	_, err = c.RPCServer(ctx, "upcase", func(ctx context.Context, req *amqp.Delivery) (amqp.Message, error) {
		return amqp.Message{Body: fmt.Sprintf("%X", req.Body)}, nil
	}, nil)
	if err != nil {
		log.Fatal(err)
	}

	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
	defer cancel()
	reply, err := c.RPCCall(ctx, "upcase", amqp.Message{Body: "hi"})
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(string(reply.Body))
}

func (*Client) RPCClient 

func (c *Client) RPCClient(ctx context.Context) (*RPCClient, error)

RPCClient returns a new RPC client, close it when done.

func (*Client) RPCServer 

func (c *Client) RPCServer(ctx context.Context, queue string, handler RPCHandler, opts *SubscribeOptions) (*Subscription, error)

RPCServer serves RPC requests from a queue (declared durable if it doesn't exist). Replies are sent to the request's ReplyTo with its CorrelationID, and with its content type and encoding unless the reply sets them. Cancel the returned subscription to stop serving.

func (*Client) TopicExchange 

func (c *Client) TopicExchange(ctx context.Context, name string) (*Exchange, error)

TopicExchange declares a durable topic exchange, use "amq.topic" for the built-in one.

Example 
package main

import (
	"context"
	"log"

	amqp "github.com/cloudamqp/amqp-client.go"
)

func main() {
	ctx := context.Background()
	c, err := amqp.NewClient(ctx, "amqp://localhost", nil)
	if err != nil {
		log.Fatal(err)
	}
	defer c.Close()

	events, err := c.TopicExchange(ctx, "events")
	if err != nil {
		log.Fatal(err)
	}
	q, err := c.Queue(ctx, "user-events", nil)
	if err != nil {
		log.Fatal(err)
	}
	// Bindings are recovered after reconnects too
	if err := q.Bind(ctx, events.Name(), "user.*", nil); err != nil {
		log.Fatal(err)
	}
	err = events.Publish(ctx, "user.created", amqp.Message{
		Body:       `{"id": 1}`,
		Properties: amqp.Properties{ContentType: "application/json"},
	})
	if err != nil {
		log.Fatal(err)
	}
}

type ClientOptions 

type ClientOptions struct {
	// Config is used for every connection the client makes.
	Config

	// ReconnectInterval is the delay before the second reconnect attempt
	// (the first is immediate), doubled for each failed attempt up to
	// MaxReconnectInterval. Defaults to 1s and 30s. A connection that's
	// lost within MaxReconnectInterval of being established counts as a
	// failed attempt, so a broker that closes connections right away isn't
	// reconnected to in a tight loop.
	ReconnectInterval    time.Duration
	MaxReconnectInterval time.Duration
	// MaxRetries is the number of consecutive failed reconnect attempts
	// before the client gives up and closes. Zero retries forever. Each
	// attempt is bounded by Config.ConnectTimeout, 30s by default.
	MaxRetries int

	// OnConnect is called after each successful connection, including the
	// first one, once the topology is recovered. The client's methods can
	// be used in it, e.g. to declare topology that isn't recovered
	// automatically. Its context is cancelled if the connection is lost.
	// What's declared with that context (queues, exchanges, bindings) isn't
	// recovered after a reconnect, as OnConnect declares it again, and
	// subscriptions started with it stop when the connection is lost.
	// An error from the first call makes NewClient fail, later errors are
	// logged. Later calls run in their own goroutine, one at a time.
	OnConnect func(ctx context.Context, c *Client) error
	// OnDisconnect is called when the connection is lost.
	OnDisconnect func(err error)
	// OnFailed is called when the client gives up reconnecting, see
	// MaxRetries.
	OnFailed func(err error)
	// OnReturn is called with messages published with Mandatory that
	// couldn't be routed. It must not block.
	OnReturn func(*Return)

	// DefaultContentType and DefaultContentEncoding are used when a
	// published message doesn't set them.
	DefaultContentType     string
	DefaultContentEncoding string
}

ClientOptions are the options for NewClient.

type Codecs 

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

Codecs is a registry of serializers by content type and encoders by content encoding. Register all codecs before the registry is used, it's not safe to modify concurrently with use.

func DefaultCodecs 

func DefaultCodecs() *Codecs

DefaultCodecs returns a registry with the built-in codecs: "application/json" (JSON), "text/plain" (Text), "gzip" (Gzip) and "deflate" (Deflate).

func NewCodecs 

func NewCodecs() *Codecs

NewCodecs returns an empty registry.

func (*Codecs) Marshal 

func (c *Codecs) Marshal(v any, p *Properties) ([]byte, error)

Marshal serializes v according to p.ContentType and then encodes it according to p.ContentEncoding. A []byte or string v isn't serialized, only encoded.

func (*Codecs) RegisterEncoder 

func (c *Codecs) RegisterEncoder(contentEncoding string, e Encoder) *Codecs

RegisterEncoder registers an encoder for a content encoding.

func (*Codecs) RegisterSerializer 

func (c *Codecs) RegisterSerializer(contentType string, s Serializer) *Codecs

RegisterSerializer registers a serializer for a content type. Parameters such as "; charset=utf-8" are ignored when looking up a content type.

func (*Codecs) Unmarshal 

func (c *Codecs) Unmarshal(body []byte, p *Properties, v any) error

Unmarshal decodes body according to p.ContentEncoding and deserializes it into v according to p.ContentType. If v is a *[]byte or *string it's set to the decoded body.

type Config 

type Config struct {
	// ConnectionName is shown in the broker's management UI. Defaults to
	// the program name.
	ConnectionName string
	// Heartbeat is the requested heartbeat interval. Zero uses the broker's
	// suggestion, a negative value disables heartbeats.
	Heartbeat time.Duration
	// FrameMax is the largest frame size the client accepts, defaults to
	// 131072. The smaller of the client's and broker's values is used.
	FrameMax uint32
	// ChannelMax is the maximum number of open channels, defaults to 2047.
	// The smaller of the client's and broker's values is used.
	ChannelMax uint16
	// ConnectTimeout bounds the TCP connect, TLS handshake and AMQP
	// handshake, in addition to the context passed to Dial. Defaults to
	// the URI's connection_timeout, and for the Client's reconnects to 30s.
	ConnectTimeout time.Duration
	// TLSConfig is used for amqps:// URIs. ServerName defaults to the host.
	TLSConfig *tls.Config
	// Dial overrides how the network connection is established, e.g. to
	// use a proxy or a UNIX socket. The connection is wrapped in TLS
	// afterwards if the URI scheme is amqps.
	Dial func(ctx context.Context, network, addr string) (net.Conn, error)
	// ClientProperties are added to the client properties sent to the broker.
	ClientProperties Table
	// Codecs used by [Delivery.Decode] (and the [Client] when publishing).
	Codecs *Codecs
	// Logger receives warnings, such as frames for unknown channels.
	// Defaults to slog.Default().
	Logger *slog.Logger
	// OnBlocked is called when the broker blocks publishing on the
	// connection, e.g. because of a memory or disk alarm. It's called from
	// the connection's read loop and must not block.
	OnBlocked func(reason string)
	// OnUnblocked is called when the broker lifts the block.
	OnUnblocked func()
}

Config holds the optional settings for a connection. Settings in the URI query string are used for the fields left unset here.

type Confirmation 

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

Confirmation is the broker's confirmation of a published message, see Channel.BasicPublishConfirm.

func (*Confirmation) Acked 

func (c *Confirmation) Acked() bool

Acked reports whether the broker acked the message, valid after Done is closed.

func (*Confirmation) DeliveryTag 

func (c *Confirmation) DeliveryTag() uint64

DeliveryTag returns the publish sequence number of the message.

func (*Confirmation) Done 

func (c *Confirmation) Done() <-chan struct{}

Done returns a channel that's closed when the broker has acked or nacked the message, or the channel was closed before that.

func (*Confirmation) Wait 

func (c *Confirmation) Wait(ctx context.Context) error

Wait waits for the confirmation. It returns nil if the broker acked the message, ErrPublishNacked if it was nacked, or an error matching ErrUnconfirmed (and ErrClosed) if the channel closed before the message was confirmed. The message may or may not have been routed then, and Channel.Err tells why the channel closed.

type Connection 

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

Connection is an AMQP connection. It's safe for concurrent use.

Frames are appended to a write buffer and written to the socket by a background goroutine, which coalesces frames written concurrently, or in quick succession, into fewer syscalls (see writer.go).

func Dial 

func Dial(ctx context.Context, uri string, cfg *Config) (*Connection, error)

Dial connects to the broker at uri, e.g. "amqp://guest:guest@localhost/vhost". cfg may be nil. The context bounds the TCP connect, the TLS handshake and the AMQP handshake.

Example 
package main

import (
	"context"
	"fmt"
	"log"
	"time"

	amqp "github.com/cloudamqp/amqp-client.go"
)

func main() {
	ctx := context.Background()
	conn, err := amqp.Dial(ctx, "amqp://guest:guest@localhost", &amqp.Config{
		ConnectionName: "example",
		Heartbeat:      10 * time.Second,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()

	ch, err := conn.Channel(ctx)
	if err != nil {
		log.Fatal(err)
	}
	q, err := ch.QueueDeclare(ctx, "", amqp.QueueDeclareOptions{Exclusive: true, AutoDelete: true})
	if err != nil {
		log.Fatal(err)
	}
	if err := ch.BasicQos(ctx, 100, false); err != nil {
		log.Fatal(err)
	}
	consumer, err := ch.BasicConsume(ctx, q.Name, amqp.ConsumeOptions{})
	if err != nil {
		log.Fatal(err)
	}

	err = ch.BasicPublish(ctx, "", q.Name, amqp.Publishing{
		Properties: amqp.Properties{ContentType: "text/plain"},
		Body:       []byte("Hello world"),
	})
	if err != nil {
		log.Fatal(err)
	}

	d, err := consumer.Next(ctx)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(string(d.Body))
	d.Ack()
}

func DialURI 

func DialURI(ctx context.Context, u URI, cfg *Config) (*Connection, error)

DialURI is like Dial but takes a parsed URI.

func (*Connection) Blocked 

func (c *Connection) Blocked() bool

Blocked reports whether the broker has blocked publishing on the connection.

func (*Connection) Channel 

func (c *Connection) Channel(ctx context.Context) (*Channel, error)

Channel opens a new channel.

func (*Connection) ChannelMax 

func (c *Connection) ChannelMax() uint16

ChannelMax returns the negotiated maximum channel id.

func (*Connection) Close 

func (c *Connection) Close() error

Close closes the connection gracefully, waiting for the broker to confirm. All channels and consumers are closed.

func (*Connection) CloseReason 

func (c *Connection) CloseReason(reason string) error

CloseReason is like Close, the reason might be logged by the broker.

func (*Connection) Done 

func (c *Connection) Done() <-chan struct{}

Done returns a channel that's closed when the connection is closed, for any reason. Err then returns the reason.

func (*Connection) Err 

func (c *Connection) Err() error

Err returns why the connection was closed: ErrClosed if closed by the client, an *Error if closed by the broker, or a network error (that matches ErrClosed). It returns nil while the connection is open.

func (*Connection) FrameMax 

func (c *Connection) FrameMax() uint32

FrameMax returns the negotiated maximum frame size.

func (*Connection) Heartbeat 

func (c *Connection) Heartbeat() time.Duration

Heartbeat returns the negotiated heartbeat interval, 0 if disabled.

func (*Connection) IsClosed 

func (c *Connection) IsClosed() bool

IsClosed reports whether the connection is closed.

func (*Connection) LocalAddr 

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

LocalAddr returns the local network address.

func (*Connection) RemoteAddr 

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

RemoteAddr returns the broker's network address.

func (*Connection) ServerProperties 

func (c *Connection) ServerProperties() Table

ServerProperties returns the properties the broker sent in the handshake, such as "product", "version" and "capabilities".

func (*Connection) UpdateSecret 

func (c *Connection) UpdateSecret(ctx context.Context, secret, reason string) error

UpdateSecret updates the credentials of the connection, e.g. a renewed OAuth 2 token, without reconnecting.

type ConsumeOptions 

type ConsumeOptions struct {
	// Tag identifies the consumer, the broker generates one if empty.
	Tag string
	// NoAck makes the broker consider messages acknowledged as soon as
	// they're delivered.
	NoAck bool
	// Exclusive requests to be the only consumer of the queue.
	Exclusive bool
	// NoLocal isn't supported by RabbitMQ or LavinMQ.
	NoLocal bool
	// Args are optional arguments, e.g. "x-stream-offset" or "x-priority".
	Args Table
}

ConsumeOptions are the options for Channel.BasicConsume.

type Consumer 

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

Consumer receives deliveries from a queue, see Channel.BasicConsume. Deliveries are buffered in memory until read, so the connection never waits for a slow consumer. Limit the buffer with Channel.BasicQos.

func (*Consumer) Cancel 

func (c *Consumer) Cancel(ctx context.Context) error

Cancel cancels the consumer. Deliveries already buffered can still be read, after that Next returns ErrConsumerCancelled.

func (*Consumer) Channel 

func (c *Consumer) Channel() *Channel

Channel returns the consumer's channel.

func (*Consumer) Deliveries 

func (c *Consumer) Deliveries() <-chan *Delivery

Deliveries returns a Go channel of deliveries, which is closed when the consumer is cancelled or its channel closed, Err then tells why. The returned channel must be read until it's closed. Consumer.Next is slightly faster.

Example 
package main

import (
	"context"
	"fmt"
	"log"

	amqp "github.com/cloudamqp/amqp-client.go"
)

func main() {
	ctx := context.Background()
	conn, err := amqp.Dial(ctx, "amqp://localhost", nil)
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	ch, err := conn.Channel(ctx)
	if err != nil {
		log.Fatal(err)
	}
	ch.BasicQos(ctx, 10, false)
	consumer, err := ch.BasicConsume(ctx, "jobs", amqp.ConsumeOptions{})
	if err != nil {
		log.Fatal(err)
	}
	shutdown := make(chan struct{})
	for {
		select {
		case d, ok := <-consumer.Deliveries():
			if !ok {
				log.Println("consumer stopped:", consumer.Err())
				return
			}
			fmt.Println(string(d.Body))
			d.Ack()
		case <-shutdown:
			consumer.Cancel(ctx)
		}
	}
}

func (*Consumer) Done 

func (c *Consumer) Done() <-chan struct{}

Done returns a channel that's closed when the consumer stops receiving deliveries, there might still be buffered deliveries to read.

func (*Consumer) Err 

func (c *Consumer) Err() error

Err returns why the consumer stopped, or nil while it's active.

func (*Consumer) Next 

func (c *Consumer) Next(ctx context.Context) (*Delivery, error)

Next returns the next delivery, waiting for one if none is buffered. When the consumer is cancelled it returns the buffered deliveries first and then the reason: ErrConsumerCancelled, ErrConsumerCancelledByServer or the channel's error. Next can be called from multiple goroutines.

func (*Consumer) Queue 

func (c *Consumer) Queue() string

Queue returns the name of the queue consumed from.

func (*Consumer) Tag 

func (c *Consumer) Tag() string

Tag returns the consumer tag.

type Decimal 

type Decimal struct {
	Scale uint8
	Value int32
}

Decimal is an AMQP decimal value, Value × 10^-Scale.

func (Decimal) String 

func (d Decimal) String() string

String formats the decimal, e.g. Decimal{Scale: 2, Value: 1234} is "12.34".

type Delivery 

type Delivery struct {
	Properties
	Body         []byte
	ConsumerTag  string // empty for basic.get
	DeliveryTag  uint64 // identifies the delivery on its channel
	Redelivered  bool   // the message has been delivered before
	Exchange     string // the exchange the message was published to
	RoutingKey   string // the routing key the message was published with
	MessageCount uint32 // messages left in the queue, only for basic.get
	// contains filtered or unexported fields
}

Delivery is a message delivered to a consumer or fetched with Channel.BasicGet.

func (*Delivery) Ack 

func (d *Delivery) Ack() error

Ack acknowledges the delivery, the broker then removes the message. It returns ErrAlreadyAcknowledged if the delivery is already acknowledged or rejected, or was delivered with no-ack.

func (*Delivery) Acknowledged 

func (d *Delivery) Acknowledged() bool

Acknowledged reports whether the delivery is acknowledged or rejected, with its own methods or the channel's with multiple, or was delivered with no-ack.

func (*Delivery) Channel 

func (d *Delivery) Channel() *Channel

Channel returns the channel the message was delivered on.

func (*Delivery) Decode 

func (d *Delivery) Decode(v any) error

Decode decodes the body into v according to the message's content encoding and content type, using the connection's Config.Codecs or DefaultCodecs. v can also be a *[]byte or *string to get the body with only the content encoding decoded.

func (*Delivery) Nack 

func (d *Delivery) Nack(requeue bool) error

Nack rejects the delivery. With requeue the message is put back in the queue, otherwise it's dropped or dead-lettered.

func (*Delivery) Reject 

func (d *Delivery) Reject(requeue bool) error

Reject is like Nack, using basic.reject.

type Encoder 

type Encoder interface {
	Encode(data []byte) ([]byte, error)
	Decode(data []byte) ([]byte, error)
}

Encoder applies a content encoding, such as "gzip", to message bodies.

func NewDeflate added in v1.2.0 

func NewDeflate(maxDecodedSize int64) Encoder

NewDeflate returns a deflate encoder that fails to decode bodies larger than maxDecodedSize with ErrDecodedTooLarge.

func NewGzip added in v1.2.0 

func NewGzip(maxDecodedSize int64) Encoder

NewGzip returns a gzip encoder that fails to decode bodies larger than maxDecodedSize with ErrDecodedTooLarge.

type Error 

type Error struct {
	Code     uint16
	Reason   string
	ClassID  uint16 // class of the method that caused the error, or 0
	MethodID uint16 // method that caused the error, or 0
	// Connection is true when the whole connection was closed, otherwise
	// only the channel was closed.
	Connection bool
	// Server is true when the broker sent the exception.
	Server bool
}

Error is an exception from the broker (or the client) that closed a channel or a connection. Code is one of the reply codes, such as NotFound or PreconditionFailed.

func (*Error) Error 

func (e *Error) Error() string

func (*Error) Is 

func (e *Error) Is(target error) bool

Is makes errors.Is(err, ErrClosed) true for broker exceptions.

type Exchange 

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

Exchange is a handle to an exchange declared with Client.Exchange.

func (*Exchange) Bind 

func (x *Exchange) Bind(ctx context.Context, source, routingKey string, args Table) error

Bind binds the exchange (as destination) to the source exchange, so that messages matching the routing key are routed to this exchange too. The binding is recovered when the client reconnects.

func (*Exchange) Delete 

func (x *Exchange) Delete(ctx context.Context, ifUnused bool) error

Delete deletes the exchange. With ifUnused it's only deleted if it has no bindings. It's no longer recovered on reconnect.

func (*Exchange) Kind 

func (x *Exchange) Kind() string

Kind returns the exchange type.

func (*Exchange) Name 

func (x *Exchange) Name() string

Name returns the exchange's name.

func (*Exchange) Publish 

func (x *Exchange) Publish(ctx context.Context, routingKey string, msg Message) error

Publish publishes a message to the exchange and waits for the broker to confirm it. See Client.Publish.

func (*Exchange) Unbind 

func (x *Exchange) Unbind(ctx context.Context, source, routingKey string, args Table) error

Unbind removes a binding to the source exchange.

type ExchangeDeclareOptions 

type ExchangeDeclareOptions struct {
	// Passive only checks if the exchange exists, the channel is closed
	// with a NotFound error otherwise.
	Passive bool
	// Durable exchanges survive broker restarts.
	Durable bool
	// AutoDelete exchanges are deleted when their last binding is removed.
	AutoDelete bool
	// Internal exchanges can't be published to directly, only via
	// exchange-to-exchange bindings.
	Internal bool
	// Args are optional arguments, e.g. "alternate-exchange".
	Args Table
}

ExchangeDeclareOptions are the options for Channel.ExchangeDeclare.

type ExchangeOptions 

type ExchangeOptions struct {
	Durable    bool
	AutoDelete bool
	Internal   bool
	Passive    bool
	Args       Table
}

ExchangeOptions are the options for Client.Exchange, see ExchangeDeclareOptions for their meaning.

type Handler 

type Handler func(ctx context.Context, d *Delivery) error

Handler processes a delivery for a Subscription. The message is acknowledged when the handler returns nil and rejected and requeued when it returns an error (or panics), unless the handler acknowledged or rejected it itself, e.g. with d.Reject(false) to dead-letter a message that can't be processed. ctx is cancelled when the client is closed.

type Message 

type Message struct {
	Properties
	// Body is a []byte or string, which is sent as is, or a value that's
	// serialized according to ContentType by the client's codecs.
	// Bodies are also encoded according to ContentEncoding (e.g. gzip) if
	// there are codecs.
	Body any
	// Mandatory makes the broker return the message, see
	// [ClientOptions.OnReturn], if it can't be routed to a queue.
	Mandatory bool
}

Message is a message published with the high-level API.

type Properties 

type Properties struct {
	ContentType     string    // MIME content type, e.g. "application/json"
	ContentEncoding string    // MIME content encoding, e.g. "gzip"
	Headers         Table     // Application headers
	DeliveryMode    uint8     // Transient (1) or Persistent (2)
	Priority        uint8     // 0 to 9
	CorrelationID   string    // Correlate a reply with a request
	ReplyTo         string    // Address to reply to, usually a queue name
	Expiration      string    // Message TTL in milliseconds, as a string
	MessageID       string    // Application message identifier
	Timestamp       time.Time // Message timestamp, second precision
	Type            string    // Message type name
	UserID          string    // Creating user id, validated by the broker
	AppID           string    // Creating application id
}

Properties are the basic class content properties of a message. Zero values are not sent on the wire.

type Publishing 

type Publishing struct {
	Properties
	Body []byte
	// Mandatory makes the broker return the message, see
	// [Channel.OnReturn], if it can't be routed to any queue.
	Mandatory bool
}

Publishing is a message to publish.

type Queue 

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

Queue is a handle to a queue declared with Client.Queue.

func (*Queue) Bind 

func (q *Queue) Bind(ctx context.Context, exchange, bindingKey string, args Table) error

Bind binds the queue to an exchange. The binding is recovered when the client reconnects.

func (*Queue) Delete 

func (q *Queue) Delete(ctx context.Context, opts QueueDeleteOptions) (n uint32, err error)

Delete deletes the queue and returns the number of messages it had. It is no longer recovered on reconnect.

func (*Queue) Get 

func (q *Queue) Get(ctx context.Context, noAck bool) (msg *Delivery, ok bool, err error)

Get fetches a message from the queue, ok is false if it's empty. Unless noAck the message has to be acknowledged. Gets share a channel, which the broker closes if a get fails (e.g. because the queue was deleted); the messages fetched on it but not yet acknowledged are then requeued.

func (*Queue) Name 

func (q *Queue) Name() string

Name returns the queue's name. The name of a queue declared with an empty name changes when the client reconnects.

func (*Queue) Publish 

func (q *Queue) Publish(ctx context.Context, msg Message) error

Publish publishes a message directly to the queue, via the default exchange, and waits for the broker to confirm it. See Client.Publish.

func (*Queue) Purge 

func (q *Queue) Purge(ctx context.Context) (n uint32, err error)

Purge removes all ready messages from the queue and returns how many were removed.

func (*Queue) Subscribe 

func (q *Queue) Subscribe(ctx context.Context, handler Handler, opts *SubscribeOptions) (*Subscription, error)

Subscribe consumes messages from the queue, calling handler for each, see Subscription.

func (*Queue) Unbind 

func (q *Queue) Unbind(ctx context.Context, exchange, bindingKey string, args Table) error

Unbind removes a binding to an exchange.

type QueueDeclareOptions 

type QueueDeclareOptions struct {
	// Passive only checks if the queue exists, the channel is closed with
	// a NotFound error otherwise.
	Passive bool
	// Durable queues survive broker restarts. Note that messages also have
	// to be published as persistent to survive.
	Durable bool
	// Exclusive queues can only be used by this connection and are deleted
	// when it closes.
	Exclusive bool
	// AutoDelete queues are deleted when their last consumer is cancelled.
	AutoDelete bool
	// Args are optional arguments, e.g. "x-queue-type", "x-max-length" or
	// "x-dead-letter-exchange".
	Args Table
}

QueueDeclareOptions are the options for Channel.QueueDeclare.

type QueueDeleteOptions 

type QueueDeleteOptions struct {
	IfUnused bool // only delete if the queue has no consumers
	IfEmpty  bool // only delete if the queue has no messages
}

QueueDeleteOptions are the options for Channel.QueueDelete.

type QueueInfo 

type QueueInfo struct {
	Name          string // the queue name, generated by the broker if declared with ""
	MessageCount  uint32 // number of messages ready in the queue
	ConsumerCount uint32 // number of consumers
}

QueueInfo is the broker's reply to a queue declaration.

type QueueOptions 

type QueueOptions struct {
	Durable    bool
	Exclusive  bool
	AutoDelete bool
	Passive    bool
	Args       Table
}

QueueOptions are the options for Client.Queue, see QueueDeclareOptions for their meaning.

type RPCClient 

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

RPCClient makes RPC calls using direct reply-to. Calls can be made concurrently, and the client recovers after reconnects.

func (*RPCClient) Call 

func (r *RPCClient) Call(ctx context.Context, queue string, msg Message) (*Delivery, error)

Call sends a request to the RPC server consuming from queue and waits for the reply, use a context with a deadline to not wait forever. If no queue receives the request ErrUnroutable is returned.

func (*RPCClient) Close 

func (r *RPCClient) Close() error

Close closes the RPC client's channel, failing calls in progress.

type RPCHandler 

type RPCHandler func(ctx context.Context, req *Delivery) (Message, error)

RPCHandler handles an RPC request and returns the reply. If it returns an error the request is rejected (dead-lettered if the queue has a dead-letter exchange) and no reply is sent.

type Return 

type Return struct {
	Properties
	Body       []byte
	ReplyCode  uint16 // e.g. NoRoute
	ReplyText  string
	Exchange   string
	RoutingKey string
}

Return is a message the broker couldn't route, published with the Mandatory flag.

type Serializer 

type Serializer interface {
	Marshal(v any) ([]byte, error)
	Unmarshal(data []byte, v any) error
}

Serializer converts values to and from message bodies of a content type, such as "application/json". 

var JSON Serializer = jsonSerializer{}

JSON serializes values with encoding/json. 

var Text Serializer = textSerializer{}

Text serializes values with fmt, e.g. numbers and fmt.Stringers. It can only decode into *string and *[]byte.

type SubscribeOptions 

type SubscribeOptions struct {
	// Prefetch is the maximum number of unacknowledged messages, defaults
	// to 20, or twice the number of workers if that's higher.
	Prefetch uint16
	// Workers is the number of goroutines calling the handler, defaults to
	// one, which processes messages in order.
	Workers int
	// NoAck makes the broker consider messages acknowledged when they're
	// delivered, they're lost if the client fails to process them.
	NoAck bool
	// Exclusive requests to be the only consumer of the queue.
	Exclusive bool
	// ConsumerTag identifies the consumer, generated by the broker if empty.
	ConsumerTag string
	// Args are optional arguments, e.g. {"x-stream-offset": "first"}.
	Args Table
}

SubscribeOptions are the options for Queue.Subscribe.

type Subscription 

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

Subscription is a consumer that survives reconnects: when the connection (or just the consumer's channel) is lost it consumes again as soon as possible. Messages that were being processed when the connection was lost are redelivered by the broker, so handlers should be idempotent.

func (*Subscription) Cancel 

func (s *Subscription) Cancel(ctx context.Context) error

Cancel stops the subscription. It waits for running handlers to return (or ctx to be done), messages that weren't processed are requeued. When called from the subscription's handler, with the handler's context, it doesn't wait for the handlers.

func (*Subscription) ConsumerTag 

func (s *Subscription) ConsumerTag() string

ConsumerTag returns the tag of the current consumer, it changes when the subscription is recovered if the broker generated it.

func (*Subscription) Done 

func (s *Subscription) Done() <-chan struct{}

Done returns a channel that's closed when the subscription stops, either because it was cancelled, the client was closed or it couldn't be recovered, e.g. because the queue was deleted. Err then tells why.

func (*Subscription) Err 

func (s *Subscription) Err() error

Err returns why the subscription stopped, or nil while it's active.

func (*Subscription) Queue 

func (s *Subscription) Queue() *Queue

Queue returns the queue the subscription consumes from.

type Table 

type Table map[string]any

Table is an AMQP field table, used for message headers and for the arguments to declarations, bindings and consumers.

These Go types can be encoded: 

nil                 void
bool                boolean ('t')
int8, uint8         short-short int ('b', 'B')
int16, uint16       short int ('s', 'u')
int32, uint32       long int ('I', 'i')
int, int64, uint,   long-long int ('l')
uint64 (≤ MaxInt64)
float32, float64    float ('f'), double ('d')
Decimal             decimal ('D')
string              long string ('S')
[]byte              byte array ('x')
time.Time           timestamp ('T'), second precision
Table,              nested table ('F')
map[string]any
[]any, []string,    array ('A')
[]Table, []int,
[]int64

Decoded tables only contain bool, int8, uint8, int16, uint16, int32, uint32, int64, float32, float64, Decimal, string, []byte, time.Time, Table, []any and nil values.

type URI 

type URI struct {
	TLS      bool
	Host     string
	Port     int
	Username string
	Password string
	Vhost    string
	// Query parameters recognized by ParseURI, zero when absent.
	Heartbeat      time.Duration // heartbeat=<seconds>, -1 when explicitly 0 (disabled)
	FrameMax       uint32        // frame_max
	ChannelMax     uint16        // channel_max
	ConnectionName string        // name or connection_name
	ConnectTimeout time.Duration // connection_timeout=<milliseconds>
	AuthMechanism  string        // auth_mechanism, "PLAIN" or "EXTERNAL"
	InsecureTLS    bool          // verify=verify_none
	ServerName     string        // server_name_indication
}

URI is a parsed AMQP URI, see https://www.rabbitmq.com/docs/uri-spec.

func ParseURI 

func ParseURI(s string) (URI, error)

ParseURI parses an amqp:// or amqps:// URI. Unset parts get the default values: guest/guest on localhost port 5672 (5671 for amqps) and vhost "/".

func (URI) Addr 

func (u URI) Addr() string

Addr returns the host:port to dial.

func (URI) String 

func (u URI) String() string

String formats the URI without the password.

Source Files

View all Source files

Jump to

Keyboard shortcuts

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