Switch branches/tags
Nothing to show
Find file
Fetching contributors…
Cannot retrieve contributors at this time
315 lines (266 sloc) 10.9 KB
// Copyright (c) 2012, Sean Treadway, SoundCloud Ltd.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// Source code and contact info at
package amqp
import (
var (
// Errors that this library could return/emit from a channel or connection
ErrClosed = &Error{Code: ChannelError, Reason: "channel/connection is not open"}
ErrSASL = &Error{Code: AccessRefused, Reason: "SASL could not negotiate a shared mechanism"}
ErrCredentials = &Error{Code: AccessRefused, Reason: "username or password not allowed"}
ErrVhost = &Error{Code: AccessRefused, Reason: "no access to this vhost"}
ErrSyntax = &Error{Code: SyntaxError, Reason: "invalid field or value inside of a frame"}
ErrFrame = &Error{Code: FrameError, Reason: "frame could not be parsed"}
ErrCommandInvalid = &Error{Code: CommandInvalid, Reason: "unexpected command received"}
ErrUnexpectedFrame = &Error{Code: UnexpectedFrame, Reason: "unexpected frame received"}
// The code and reason a channel or connection has been closed by the server.
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 differnet parameters
func newError(code uint16, text string) *Error {
return &Error{
Code: int(code),
Reason: text,
Recover: isSoftExceptionCode(int(code)),
Server: true,
func (me Error) Error() string {
return fmt.Sprintf("Exception (%d) Reason: %q", me.Code, me.Reason)
// Used by header frames to capture routing and header information
type properties struct {
ContentType string // MIME content type
ContentEncoding string // MIME content encoding
Headers Table // Application or header exchange table
DeliveryMode uint8 // queue implemention use - Transient (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
reserved1 string // was cluster-id - process for buffer consumption
// DeliveryMode. Transient means higher throughput but messages will not be
// restored on broker restart. The delivery mode of publishings is unrelated
// to the durability of the queues they reside on. Transient messages will
// not be restored to durable queues, persistent messages will be restored to
// durable queues and lost on non-durable queues during server restart.
// This remains typed as uint8 to match Publishing.DeliveryMode. Other
// delivery modes specific to custom queue implementations are not enumerated
// here.
const (
Transient uint8 = 1
Persistent uint8 = 2
// The property flags are an array of bits that indicate the presence or
// absence of each property value in sequence. The bits are ordered from most
// high to low - bit 15 indicates the first property.
const (
flagContentType = 0x8000
flagContentEncoding = 0x4000
flagHeaders = 0x2000
flagDeliveryMode = 0x1000
flagPriority = 0x0800
flagCorrelationId = 0x0400
flagReplyTo = 0x0200
flagExpiration = 0x0100
flagMessageId = 0x0080
flagTimestamp = 0x0040
flagType = 0x0020
flagUserId = 0x0010
flagAppId = 0x0008
flagReserved1 = 0x0004
// Current state of the queue on the server returned from Channel.QueueDeclare or
// Channel.QueueInspect.
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
// A published message from the client 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 convienence and efficiency.
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
Body []byte
// The golang type that matches the amqp type. Scale is the number of decimal digits
// Scale == 2, Value == 12345, Decimal == 123.45
type Decimal struct {
Scale uint8
Value int32
// The amqp type that represents a string to field. Most Go types are supported in
// table serialization.
type Table map[string]interface{}
// Heap interface for maintaining delivery tags
type tagSet []uint64
func (me tagSet) Len() int { return len(me) }
func (me tagSet) Less(i, j int) bool { return (me)[i] < (me)[j] }
func (me tagSet) Swap(i, j int) { (me)[i], (me)[j] = (me)[j], (me)[i] }
func (me *tagSet) Push(tag interface{}) { *me = append(*me, tag.(uint64)) }
func (me *tagSet) Pop() interface{} {
val := (*me)[len(*me)-1]
*me = (*me)[:len(*me)-1]
return val
type message interface {
id() (uint16, uint16)
wait() bool
read(io.Reader) error
write(io.Writer) error
type messageWithContent interface {
getContent() (properties, []byte)
setContent(properties, []byte)
The base interface implemented as:
2.3.5 frame Details
All frames consist of a header (7 octets), a payload of arbitrary size, and a 'frame-end' octet that detects
malformed frames:
0 1 3 7 size+7 size+8
+------+---------+-------------+ +------------+ +-----------+
| type | channel | size | | payload | | frame-end |
+------+---------+-------------+ +------------+ +-----------+
octet short long size octets octet
To read a frame, we:
1. Read the header and check the frame type and channel.
2. Depending on the frame type, we read the payload and process it.
3. Read the frame end octet.
In realistic implementations where performance is a concern, we would use
“read-ahead buffering” or “gathering reads” to avoid doing three separate
system calls to read a frame.
type frame interface {
write(io.Writer) error
channel() uint16
type reader struct {
r io.Reader
type writer struct {
w io.Writer
// Implements the frame interface for Connection RPC
type protocolHeader struct{}
func (protocolHeader) write(w io.Writer) error {
_, err := w.Write([]byte{'A', 'M', 'Q', 'P', 0, 0, 9, 1})
return err
func (protocolHeader) channel() uint16 {
panic("only valid as initial handshake")
Method frames carry the high-level protocol commands (which we call "methods").
One method frame carries one command. The method frame payload has this format:
0 2 4
+----------+-----------+-------------- - -
| class-id | method-id | arguments...
+----------+-----------+-------------- - -
short short ...
To process a method frame, we:
1. Read the method frame payload.
2. Unpack it into a structure. A given method always has the same structure,
so we can unpack the method rapidly. 3. Check that the method is allowed in
the current context.
4. Check that the method arguments are valid.
5. Execute the method.
Method frame bodies are constructed as a list of AMQP data fields (bits,
integers, strings and string tables). The marshalling code is trivially
generated directly from the protocol specifications, and can be very rapid.
type methodFrame struct {
ChannelId uint16
ClassId uint16
MethodId uint16
Method message
func (me *methodFrame) channel() uint16 { return me.ChannelId }
Heartbeating is a technique designed to undo one of TCP/IP's features, namely
its ability to recover from a broken physical connection by closing only after
a quite long time-out. In some scenarios we need to know very rapidly if a
peer is disconnected or not responding for other reasons (e.g. it is looping).
Since heartbeating can be done at a low level, we implement this as a special
type of frame that peers exchange at the transport level, rather than as a
class method.
type heartbeatFrame struct {
ChannelId uint16
func (me *heartbeatFrame) channel() uint16 { return me.ChannelId }
Certain methods (such as Basic.Publish, Basic.Deliver, etc.) are formally
defined as carrying content. When a peer sends such a method frame, it always
follows it with a content header and zero or more content body frames.
A content header frame has this format:
0 2 4 12 14
+----------+--------+-----------+----------------+------------- - -
| class-id | weight | body size | property flags | property list...
+----------+--------+-----------+----------------+------------- - -
short short long long short remainder...
We place content body in distinct frames (rather than including it in the
method) so that AMQP may support "zero copy" techniques in which content is
never marshalled or encoded. We place the content properties in their own
frame so that recipients can selectively discard contents they do not want to
type headerFrame struct {
ChannelId uint16
ClassId uint16
weight uint16
Size uint64
Properties properties
func (me *headerFrame) channel() uint16 { return me.ChannelId }
Content is the application data we carry from client-to-client via the AMQP
server. Content is, roughly speaking, a set of properties plus a binary data
part. The set of allowed properties are defined by the Basic class, and these
form the "content header frame". The data can be any size, and MAY be broken
into several (or many) chunks, each forming a "content body frame".
Looking at the frames for a specific channel, as they pass on the wire, we
might see something like this:
[method] [header] [body] [body]
type bodyFrame struct {
ChannelId uint16
Body []byte
func (me *bodyFrame) channel() uint16 { return me.ChannelId }