src: Add DMA localagent

Change-Id: Ibcee814fbc9a904448eeb368a1a26bbb69cf54aa
Signed-off-by: Toshiaki Takahashi <takahashi.tsc@ncos.nec.co.jp>

23 files changed, 8309 insertions, 0 deletions

diff --git a/src/dma/vendor/github.com/streadway/amqp/.gitignore b/src/dma/vendor/github.com/streadway/amqp/.gitignore
new file mode 100644
index 00000000..ba8a7056
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/.gitignore
@@ -0,0 +1,4 @@
+certs/*
+spec/spec
+examples/simple-consumer/simple-consumer
+examples/simple-producer/simple-producer
diff --git a/src/dma/vendor/github.com/streadway/amqp/.travis.yml b/src/dma/vendor/github.com/streadway/amqp/.travis.yml
new file mode 100644
index 00000000..7166964c
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/.travis.yml
@@ -0,0 +1,18 @@
+language: go
+
+go:
+  - 1.9.x
+  - 1.10.x
+
+services:
+  - rabbitmq
+
+env:
+ - AMQP_URL=amqp://guest:guest@127.0.0.1:5672/
+
+before_install:
+  - go get -v github.com/golang/lint/golint
+
+script:
+  - ./pre-commit
+  - go test -cpu=1,2 -v -tags integration ./...
diff --git a/src/dma/vendor/github.com/streadway/amqp/CONTRIBUTING.md b/src/dma/vendor/github.com/streadway/amqp/CONTRIBUTING.md
new file mode 100644
index 00000000..c87f3d7e
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/CONTRIBUTING.md
@@ -0,0 +1,35 @@
+## Prequisites
+
+1. Go: [https://golang.org/dl/](https://golang.org/dl/)
+1. Golint `go get -u -v github.com/golang/lint/golint`
+
+## Contributing
+
+The workflow is pretty standard:
+
+1. Fork github.com/streadway/amqp
+1. Add the pre-commit hook: `ln -s ../../pre-commit .git/hooks/pre-commit`
+1. Create your feature branch (`git checkout -b my-new-feature`)
+1. Run integration tests (see below)
+1. **Implement tests**
+1. Implement fixs
+1. Commit your changes (`git commit -am 'Add some feature'`)
+1. Push to a branch (`git push -u origin my-new-feature`)
+1. Submit a pull request
+
+## Running Tests
+
+The test suite assumes that:
+
+ * A RabbitMQ node is running on localhost with all defaults: [https://www.rabbitmq.com/download.html](https://www.rabbitmq.com/download.html)
+ * `AMQP_URL` is exported to `amqp://guest:guest@127.0.0.1:5672/`
+
+### Integration Tests
+
+After starting a local RabbitMQ, run integration tests with the following:
+
+    env AMQP_URL=amqp://guest:guest@127.0.0.1:5672/ go test -v -cpu 2 -tags integration -race
+
+All integration tests should use the `integrationConnection(...)` test
+helpers defined in `integration_test.go` to setup the integration environment
+and logging.
diff --git a/src/dma/vendor/github.com/streadway/amqp/LICENSE b/src/dma/vendor/github.com/streadway/amqp/LICENSE
new file mode 100644
index 00000000..243c0ce7
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/LICENSE
@@ -0,0 +1,23 @@
+Copyright (c) 2012, Sean Treadway, SoundCloud Ltd.
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+Redistributions of source code must retain the above copyright notice, this
+list of conditions and the following disclaimer.
+
+Redistributions in binary form must reproduce the above copyright notice, this
+list of conditions and the following disclaimer in the documentation and/or
+other materials provided with the distribution.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
diff --git a/src/dma/vendor/github.com/streadway/amqp/README.md b/src/dma/vendor/github.com/streadway/amqp/README.md
new file mode 100644
index 00000000..099db276
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/README.md
@@ -0,0 +1,93 @@
+[![Build Status](https://api.travis-ci.org/streadway/amqp.svg)](http://travis-ci.org/streadway/amqp) [![GoDoc](https://godoc.org/github.com/streadway/amqp?status.svg)](http://godoc.org/github.com/streadway/amqp)
+
+# Go RabbitMQ Client Library
+
+This is an AMQP 0.9.1 client with RabbitMQ extensions in Go.
+
+## Project Maturity
+
+This project has been used in production systems for many years. It is reasonably mature
+and feature complete, and as of November 2016 has [a team of maintainers](https://github.com/streadway/amqp/issues/215).
+
+Future API changes are unlikely but possible. They will be discussed on [Github
+issues](https://github.com/streadway/amqp/issues) along with any bugs or
+enhancements.
+
+## Supported Go Versions
+
+This library supports two most recent Go release series, currently 1.8 and 1.9.
+
+
+## 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](http://godoc.org/github.com/streadway/amqp) for
+reference and usage.
+
+[RabbitMQ tutorials in
+Go](https://github.com/rabbitmq/rabbitmq-tutorials/tree/master/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](https://github.com/streadway/amqp/graphs/contributors).
+
+## External packages
+
+  * [Google App Engine Dialer support](https://github.com/soundtrackyourbrand/gaeamqp)
+  * [RabbitMQ examples in Go](https://github.com/rabbitmq/rabbitmq-tutorials/tree/master/go)
+
+## License
+
+BSD 2 clause - see LICENSE for more details.
+
+
diff --git a/src/dma/vendor/github.com/streadway/amqp/allocator.go b/src/dma/vendor/github.com/streadway/amqp/allocator.go
new file mode 100644
index 00000000..53620e7d
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/allocator.go
@@ -0,0 +1,106 @@
+package amqp
+
+import (
+	"bytes"
+	"fmt"
+	"math/big"
+)
+
+const (
+	free      = 0
+	allocated = 1
+)
+
+// allocator maintains a bitset of allocated numbers.
+type allocator struct {
+	pool *big.Int
+	last int
+	low  int
+	high int
+}
+
+// NewAllocator reserves and frees integers out of a range between low and
+// high.
+//
+// O(N) worst case space used, where N is maximum allocated, divided by
+// sizeof(big.Word)
+func newAllocator(low, high int) *allocator {
+	return &allocator{
+		pool: big.NewInt(0),
+		last: low,
+		low:  low,
+		high: high,
+	}
+}
+
+// String returns a string describing the contents of the allocator like
+// "allocator[low..high] reserved..until"
+//
+// O(N) where N is high-low
+func (a allocator) String() string {
+	b := &bytes.Buffer{}
+	fmt.Fprintf(b, "allocator[%d..%d]", a.low, a.high)
+
+	for low := a.low; low <= a.high; low++ {
+		high := low
+		for a.reserved(high) && high <= a.high {
+			high++
+		}
+
+		if high > low+1 {
+			fmt.Fprintf(b, " %d..%d", low, high-1)
+		} else if high > low {
+			fmt.Fprintf(b, " %d", high-1)
+		}
+
+		low = high
+	}
+	return b.String()
+}
+
+// Next reserves and returns the next available number out of the range between
+// low and high.  If no number is available, false is returned.
+//
+// O(N) worst case runtime where N is allocated, but usually O(1) due to a
+// rolling index into the oldest allocation.
+func (a *allocator) next() (int, bool) {
+	wrapped := a.last
+
+	// Find trailing bit
+	for ; a.last <= a.high; a.last++ {
+		if a.reserve(a.last) {
+			return a.last, true
+		}
+	}
+
+	// Find preceding free'd pool
+	a.last = a.low
+
+	for ; a.last < wrapped; a.last++ {
+		if a.reserve(a.last) {
+			return a.last, true
+		}
+	}
+
+	return 0, false
+}
+
+// reserve claims the bit if it is not already claimed, returning true if
+// successfully claimed.
+func (a *allocator) reserve(n int) bool {
+	if a.reserved(n) {
+		return false
+	}
+	a.pool.SetBit(a.pool, n-a.low, allocated)
+	return true
+}
+
+// reserved returns true if the integer has been allocated
+func (a *allocator) reserved(n int) bool {
+	return a.pool.Bit(n-a.low) == allocated
+}
+
+// release frees the use of the number for another allocation
+func (a *allocator) release(n int) {
+	a.pool.SetBit(a.pool, n-a.low, free)
+}
diff --git a/src/dma/vendor/github.com/streadway/amqp/auth.go b/src/dma/vendor/github.com/streadway/amqp/auth.go
new file mode 100644
index 00000000..ebc765b6
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/auth.go
@@ -0,0 +1,46 @@
+// 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 http://github.com/streadway/amqp
+
+package amqp
+
+import (
+	"fmt"
+)
+
+// Authentication interface provides a means for different SASL authentication
+// mechanisms to be used during connection tuning.
+type Authentication interface {
+	Mechanism() string
+	Response() string
+}
+
+// PlainAuth is a similar to Basic Auth in HTTP.
+type PlainAuth struct {
+	Username string
+	Password string
+}
+
+// Mechanism returns "PLAIN"
+func (auth *PlainAuth) Mechanism() string {
+	return "PLAIN"
+}
+
+// Response returns the null character delimited encoding for the SASL PLAIN Mechanism.
+func (auth *PlainAuth) Response() string {
+	return fmt.Sprintf("\000%s\000%s", auth.Username, auth.Password)
+}
+
+// Finds the first mechanism preferred by the client that the server supports.
+func pickSASLMechanism(client []Authentication, serverMechanisms []string) (auth Authentication, ok bool) {
+	for _, auth = range client {
+		for _, mech := range serverMechanisms {
+			if auth.Mechanism() == mech {
+				return auth, true
+			}
+		}
+	}
+
+	return
+}
diff --git a/src/dma/vendor/github.com/streadway/amqp/certs.sh b/src/dma/vendor/github.com/streadway/amqp/certs.sh
new file mode 100755
index 00000000..834f4224
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/certs.sh
@@ -0,0 +1,159 @@
+#!/bin/sh
+#
+# Creates the CA, server and client certs to be used by tls_test.go
+# http://www.rabbitmq.com/ssl.html
+#
+# Copy stdout into the const section of tls_test.go or use for RabbitMQ
+#
+root=$PWD/certs
+
+if [ -f $root/ca/serial ]; then
+  echo >&2 "Previous installation found"
+  echo >&2 "Remove $root/ca and rerun to overwrite"
+  exit 1
+fi
+
+mkdir -p $root/ca/private
+mkdir -p $root/ca/certs
+mkdir -p $root/server
+mkdir -p $root/client
+
+cd $root/ca
+
+chmod 700 private
+touch index.txt
+echo 'unique_subject = no' > index.txt.attr
+echo '01' > serial
+echo >openssl.cnf '
+[ ca ]
+default_ca = testca
+
+[ testca ]
+dir = .
+certificate = $dir/cacert.pem
+database = $dir/index.txt
+new_certs_dir = $dir/certs
+private_key = $dir/private/cakey.pem
+serial = $dir/serial
+
+default_crl_days = 7
+default_days = 3650
+default_md = sha1
+
+policy = testca_policy
+x509_extensions = certificate_extensions
+
+[ testca_policy ]
+commonName = supplied
+stateOrProvinceName = optional
+countryName = optional
+emailAddress = optional
+organizationName = optional
+organizationalUnitName = optional
+
+[ certificate_extensions ]
+basicConstraints = CA:false
+
+[ req ]
+default_bits = 2048
+default_keyfile = ./private/cakey.pem
+default_md = sha1
+prompt = yes
+distinguished_name = root_ca_distinguished_name
+x509_extensions = root_ca_extensions
+
+[ root_ca_distinguished_name ]
+commonName = hostname
+
+[ root_ca_extensions ]
+basicConstraints = CA:true
+keyUsage = keyCertSign, cRLSign
+
+[ client_ca_extensions ]
+basicConstraints = CA:false
+keyUsage = digitalSignature
+extendedKeyUsage = 1.3.6.1.5.5.7.3.2
+
+[ server_ca_extensions ]
+basicConstraints = CA:false
+keyUsage = keyEncipherment
+extendedKeyUsage = 1.3.6.1.5.5.7.3.1
+subjectAltName = @alt_names
+
+[ alt_names ]
+IP.1 = 127.0.0.1
+'
+
+openssl req \
+  -x509 \
+  -nodes \
+  -config openssl.cnf \
+  -newkey rsa:2048 \
+  -days 3650 \
+  -subj "/CN=MyTestCA/" \
+  -out cacert.pem \
+  -outform PEM
+
+openssl x509 \
+  -in cacert.pem \
+  -out cacert.cer \
+  -outform DER
+
+openssl genrsa -out $root/server/key.pem 2048
+openssl genrsa -out $root/client/key.pem 2048
+
+openssl req \
+  -new \
+  -nodes \
+  -config openssl.cnf \
+  -subj "/CN=127.0.0.1/O=server/" \
+  -key $root/server/key.pem \
+  -out $root/server/req.pem \
+  -outform PEM
+
+openssl req \
+  -new \
+  -nodes \
+  -config openssl.cnf \
+  -subj "/CN=127.0.0.1/O=client/" \
+  -key $root/client/key.pem \
+  -out $root/client/req.pem \
+  -outform PEM
+
+openssl ca \
+  -config openssl.cnf \
+  -in $root/server/req.pem \
+  -out $root/server/cert.pem \
+  -notext \
+  -batch \
+  -extensions server_ca_extensions
+
+openssl ca \
+  -config openssl.cnf \
+  -in $root/client/req.pem \
+  -out $root/client/cert.pem \
+  -notext \
+  -batch \
+  -extensions client_ca_extensions
+
+cat <<-END
+const caCert = \`
+`cat $root/ca/cacert.pem`
+\`
+
+const serverCert = \`
+`cat $root/server/cert.pem`
+\`
+
+const serverKey = \`
+`cat $root/server/key.pem`
+\`
+
+const clientCert = \`
+`cat $root/client/cert.pem`
+\`
+
+const clientKey = \`
+`cat $root/client/key.pem`
+\`
+END
diff --git a/src/dma/vendor/github.com/streadway/amqp/channel.go b/src/dma/vendor/github.com/streadway/amqp/channel.go
new file mode 100644
index 00000000..dd2552ca
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/channel.go
@@ -0,0 +1,1587 @@
+// 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 http://github.com/streadway/amqp
+
+package amqp
+
+import (
+	"reflect"
+	"sync"
+	"sync/atomic"
+)
+
+// 0      1         3             7                  size+7 size+8
+// +------+---------+-------------+  +------------+  +-----------+
+// | type | channel |     size    |  |  payload   |  | frame-end |
+// +------+---------+-------------+  +------------+  +-----------+
+//  octet   short         long         size octets       octet
+const frameHeaderSize = 1 + 2 + 4 + 1
+
+/*
+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.
+
+*/
+type Channel struct {
+	destructor sync.Once
+	m          sync.Mutex // struct field mutex
+	confirmM   sync.Mutex // publisher confirms state mutex
+	notifyM    sync.RWMutex
+
+	connection *Connection
+
+	rpc       chan message
+	consumers *consumers
+
+	id uint16
+
+	// closed is set to 1 when the channel has been closed - see Channel.send()
+	closed int32
+
+	// true when we will never notify again
+	noNotify bool
+
+	// Channel and Connection exceptions will be broadcast on these listeners.
+	closes []chan *Error
+
+	// Listeners for active=true flow control.  When true is sent to a listener,
+	// publishing should pause until false is sent to listeners.
+	flows []chan bool
+
+	// Listeners for returned publishings for unroutable messages on mandatory
+	// publishings or undeliverable messages on immediate publishings.
+	returns []chan Return
+
+	// Listeners for when the server notifies the client that
+	// a consumer has been cancelled.
+	cancels []chan string
+
+	// Allocated when in confirm mode in order to track publish counter and order confirms
+	confirms   *confirms
+	confirming bool
+
+	// Selects on any errors from shutdown during RPC
+	errors chan *Error
+
+	// State machine that manages frame order, must only be mutated by the connection
+	recv func(*Channel, frame) error
+
+	// Current state for frame re-assembly, only mutated from recv
+	message messageWithContent
+	header  *headerFrame
+	body    []byte
+}
+
+// Constructs a new channel with the given framing rules
+func newChannel(c *Connection, id uint16) *Channel {
+	return &Channel{
+		connection: c,
+		id:         id,
+		rpc:        make(chan message),
+		consumers:  makeConsumers(),
+		confirms:   newConfirms(),
+		recv:       (*Channel).recvMethod,
+		errors:     make(chan *Error, 1),
+	}
+}
+
+// shutdown is called by Connection after the channel has been removed from the
+// connection registry.
+func (ch *Channel) shutdown(e *Error) {
+	ch.destructor.Do(func() {
+		ch.m.Lock()
+		defer ch.m.Unlock()
+
+		// Grab an exclusive lock for the notify channels
+		ch.notifyM.Lock()
+		defer ch.notifyM.Unlock()
+
+		// Broadcast abnormal shutdown
+		if e != nil {
+			for _, c := range ch.closes {
+				c <- e
+			}
+		}
+
+		// Signal that from now on, Channel.send() should call
+		// Channel.sendClosed()
+		atomic.StoreInt32(&ch.closed, 1)
+
+		// Notify RPC if we're selecting
+		if e != nil {
+			ch.errors <- e
+		}
+
+		ch.consumers.close()
+
+		for _, c := range ch.closes {
+			close(c)
+		}
+
+		for _, c := range ch.flows {
+			close(c)
+		}
+
+		for _, c := range ch.returns {
+			close(c)
+		}
+
+		for _, c := range ch.cancels {
+			close(c)
+		}
+
+		// Set the slices to nil to prevent the dispatch() range from sending on
+		// the now closed channels after we release the notifyM mutex
+		ch.flows = nil
+		ch.closes = nil
+		ch.returns = nil
+		ch.cancels = nil
+
+		if ch.confirms != nil {
+			ch.confirms.Close()
+		}
+
+		close(ch.errors)
+		ch.noNotify = true
+	})
+}
+
+// send calls Channel.sendOpen() during normal operation.
+//
+// After the channel has been closed, send calls Channel.sendClosed(), ensuring
+// only 'channel.close' is sent to the server.
+func (ch *Channel) send(msg message) (err error) {
+	// If the channel is closed, use Channel.sendClosed()
+	if atomic.LoadInt32(&ch.closed) == 1 {
+		return ch.sendClosed(msg)
+	}
+
+	return ch.sendOpen(msg)
+}
+
+func (ch *Channel) open() error {
+	return ch.call(&channelOpen{}, &channelOpenOk{})
+}
+
+// Performs a request/response call for when the message is not NoWait and is
+// specified as Synchronous.
+func (ch *Channel) call(req message, res ...message) error {
+	if err := ch.send(req); err != nil {
+		return err
+	}
+
+	if req.wait() {
+		select {
+		case e, ok := <-ch.errors:
+			if ok {
+				return e
+			}
+			return ErrClosed
+
+		case msg := <-ch.rpc:
+			if msg != nil {
+				for _, try := range res {
+					if reflect.TypeOf(msg) == reflect.TypeOf(try) {
+						// *res = *msg
+						vres := reflect.ValueOf(try).Elem()
+						vmsg := reflect.ValueOf(msg).Elem()
+						vres.Set(vmsg)
+						return nil
+					}
+				}
+				return ErrCommandInvalid
+			}
+			// RPC channel has been closed without an error, likely due to a hard
+			// error on the Connection.  This indicates we have already been
+			// shutdown and if were waiting, will have returned from the errors chan.
+			return ErrClosed
+		}
+	}
+
+	return nil
+}
+
+func (ch *Channel) sendClosed(msg message) (err error) {
+	// After a 'channel.close' is sent or received the only valid response is
+	// channel.close-ok
+	if _, ok := msg.(*channelCloseOk); ok {
+		return ch.connection.send(&methodFrame{
+			ChannelId: ch.id,
+			Method:    msg,
+		})
+	}
+
+	return ErrClosed
+}
+
+func (ch *Channel) sendOpen(msg message) (err error) {
+	if content, ok := msg.(messageWithContent); ok {
+		props, body := content.getContent()
+		class, _ := content.id()
+
+		// catch client max frame size==0 and server max frame size==0
+		// set size to length of what we're trying to publish
+		var size int
+		if ch.connection.Config.FrameSize > 0 {
+			size = ch.connection.Config.FrameSize - frameHeaderSize
+		} else {
+			size = len(body)
+		}
+
+		if err = ch.connection.send(&methodFrame{
+			ChannelId: ch.id,
+			Method:    content,
+		}); err != nil {
+			return
+		}
+
+		if err = ch.connection.send(&headerFrame{
+			ChannelId:  ch.id,
+			ClassId:    class,
+			Size:       uint64(len(body)),
+			Properties: props,
+		}); err != nil {
+			return
+		}
+
+		// chunk body into size (max frame size - frame header size)
+		for i, j := 0, size; i < len(body); i, j = j, j+size {
+			if j > len(body) {
+				j = len(body)
+			}
+
+			if err = ch.connection.send(&bodyFrame{
+				ChannelId: ch.id,
+				Body:      body[i:j],
+			}); err != nil {
+				return
+			}
+		}
+	} else {
+		err = ch.connection.send(&methodFrame{
+			ChannelId: ch.id,
+			Method:    msg,
+		})
+	}
+
+	return
+}
+
+// Eventually called via the state machine from the connection's reader
+// goroutine, so assumes serialized access.
+func (ch *Channel) dispatch(msg message) {
+	switch m := msg.(type) {
+	case *channelClose:
+		// lock before sending connection.close-ok
+		// to avoid unexpected interleaving with basic.publish frames if
+		// publishing is happening concurrently
+		ch.m.Lock()
+		ch.send(&channelCloseOk{})
+		ch.m.Unlock()
+		ch.connection.closeChannel(ch, newError(m.ReplyCode, m.ReplyText))
+
+	case *channelFlow:
+		ch.notifyM.RLock()
+		for _, c := range ch.flows {
+			c <- m.Active
+		}
+		ch.notifyM.RUnlock()
+		ch.send(&channelFlowOk{Active: m.Active})
+
+	case *basicCancel:
+		ch.notifyM.RLock()
+		for _, c := range ch.cancels {
+			c <- m.ConsumerTag
+		}
+		ch.notifyM.RUnlock()
+		ch.consumers.cancel(m.ConsumerTag)
+
+	case *basicReturn:
+		ret := newReturn(*m)
+		ch.notifyM.RLock()
+		for _, c := range ch.returns {
+			c <- *ret
+		}
+		ch.notifyM.RUnlock()
+
+	case *basicAck:
+		if ch.confirming {
+			if m.Multiple {
+				ch.confirms.Multiple(Confirmation{m.DeliveryTag, true})
+			} else {
+				ch.confirms.One(Confirmation{m.DeliveryTag, true})
+			}
+		}
+
+	case *basicNack:
+		if ch.confirming {
+			if m.Multiple {
+				ch.confirms.Multiple(Confirmation{m.DeliveryTag, false})
+			} else {
+				ch.confirms.One(Confirmation{m.DeliveryTag, false})
+			}
+		}
+
+	case *basicDeliver:
+		ch.consumers.send(m.ConsumerTag, newDelivery(ch, m))
+		// TODO log failed consumer and close channel, this can happen when
+		// deliveries are in flight and a no-wait cancel has happened
+
+	default:
+		ch.rpc <- msg
+	}
+}
+
+func (ch *Channel) transition(f func(*Channel, frame) error) error {
+	ch.recv = f
+	return nil
+}
+
+func (ch *Channel) recvMethod(f frame) error {
+	switch frame := f.(type) {
+	case *methodFrame:
+		if msg, ok := frame.Method.(messageWithContent); ok {
+			ch.body = make([]byte, 0)
+			ch.message = msg
+			return ch.transition((*Channel).recvHeader)
+		}
+
+		ch.dispatch(frame.Method) // termination state
+		return ch.transition((*Channel).recvMethod)
+
+	case *headerFrame:
+		// drop
+		return ch.transition((*Channel).recvMethod)
+
+	case *bodyFrame:
+		// drop
+		return ch.transition((*Channel).recvMethod)
+	}
+
+	panic("unexpected frame type")
+}
+
+func (ch *Channel) recvHeader(f frame) error {
+	switch frame := f.(type) {
+	case *methodFrame:
+		// interrupt content and handle method
+		return ch.recvMethod(f)
+
+	case *headerFrame:
+		// start collecting if we expect body frames
+		ch.header = frame
+
+		if frame.Size == 0 {
+			ch.message.setContent(ch.header.Properties, ch.body)
+			ch.dispatch(ch.message) // termination state
+			return ch.transition((*Channel).recvMethod)
+		}
+		return ch.transition((*Channel).recvContent)
+
+	case *bodyFrame:
+		// drop and reset
+		return ch.transition((*Channel).recvMethod)
+	}
+
+	panic("unexpected frame type")
+}
+
+// state after method + header and before the length
+// defined by the header has been reached
+func (ch *Channel) recvContent(f frame) error {
+	switch frame := f.(type) {
+	case *methodFrame:
+		// interrupt content and handle method
+		return ch.recvMethod(f)
+
+	case *headerFrame:
+		// drop and reset
+		return ch.transition((*Channel).recvMethod)
+
+	case *bodyFrame:
+		ch.body = append(ch.body, frame.Body...)
+
+		if uint64(len(ch.body)) >= ch.header.Size {
+			ch.message.setContent(ch.header.Properties, ch.body)
+			ch.dispatch(ch.message) // termination state
+			return ch.transition((*Channel).recvMethod)
+		}
+
+		return ch.transition((*Channel).recvContent)
+	}
+
+	panic("unexpected frame type")
+}
+
+/*
+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 (ch *Channel) Close() error {
+	defer ch.connection.closeChannel(ch, nil)
+	return ch.call(
+		&channelClose{ReplyCode: replySuccess},
+		&channelCloseOk{},
+	)
+}
+
+/*
+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 (ch *Channel) NotifyClose(c chan *Error) chan *Error {
+	ch.notifyM.Lock()
+	defer ch.notifyM.Unlock()
+
+	if ch.noNotify {
+		close(c)
+	} else {
+		ch.closes = append(ch.closes, c)
+	}
+
+	return c
+}
+
+/*
+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 (ch *Channel) NotifyFlow(c chan bool) chan bool {
+	ch.notifyM.Lock()
+	defer ch.notifyM.Unlock()
+
+	if ch.noNotify {
+		close(c)
+	} else {
+		ch.flows = append(ch.flows, c)
+	}
+
+	return c
+}
+
+/*
+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 (ch *Channel) NotifyReturn(c chan Return) chan Return {
+	ch.notifyM.Lock()
+	defer ch.notifyM.Unlock()
+
+	if ch.noNotify {
+		close(c)
+	} else {
+		ch.returns = append(ch.returns, c)
+	}
+
+	return c
+}
+
+/*
+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 (ch *Channel) NotifyCancel(c chan string) chan string {
+	ch.notifyM.Lock()
+	defer ch.notifyM.Unlock()
+
+	if ch.noNotify {
+		close(c)
+	} else {
+		ch.cancels = append(ch.cancels, c)
+	}
+
+	return c
+}
+
+/*
+NotifyConfirm calls NotifyPublish and starts a goroutine sending
+ordered Ack and Nack DeliveryTag to the respective channels.
+
+For strict ordering, use NotifyPublish instead.
+*/
+func (ch *Channel) NotifyConfirm(ack, nack chan uint64) (chan uint64, chan uint64) {
+	confirms := ch.NotifyPublish(make(chan Confirmation, len(ack)+len(nack)))
+
+	go func() {
+		for c := range confirms {
+			if c.Ack {
+				ack <- c.DeliveryTag
+			} else {
+				nack <- c.DeliveryTag
+			}
+		}
+		close(ack)
+		if nack != ack {
+			close(nack)
+		}
+	}()
+
+	return ack, nack
+}
+
+/*
+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 (ch *Channel) NotifyPublish(confirm chan Confirmation) chan Confirmation {
+	ch.notifyM.Lock()
+	defer ch.notifyM.Unlock()
+
+	if ch.noNotify {
+		close(confirm)
+	} else {
+		ch.confirms.Listen(confirm)
+	}
+
+	return confirm
+
+}
+
+/*
+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 (ch *Channel) Qos(prefetchCount, prefetchSize int, global bool) error {
+	return ch.call(
+		&basicQos{
+			PrefetchCount: uint16(prefetchCount),
+			PrefetchSize:  uint32(prefetchSize),
+			Global:        global,
+		},
+		&basicQosOk{},
+	)
+}
+
+/*
+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 (ch *Channel) Cancel(consumer string, noWait bool) error {
+	req := &basicCancel{
+		ConsumerTag: consumer,
+		NoWait:      noWait,
+	}
+	res := &basicCancelOk{}
+
+	if err := ch.call(req, res); err != nil {
+		return err
+	}
+
+	if req.wait() {
+		ch.consumers.cancel(res.ConsumerTag)
+	} else {
+		// Potentially could drop deliveries in flight
+		ch.consumers.cancel(consumer)
+	}
+
+	return nil
+}
+
+/*
+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 (ch *Channel) QueueDeclare(name string, durable, autoDelete, exclusive, noWait bool, args Table) (Queue, error) {
+	if err := args.Validate(); err != nil {
+		return Queue{}, err
+	}
+
+	req := &queueDeclare{
+		Queue:      name,
+		Passive:    false,
+		Durable:    durable,
+		AutoDelete: autoDelete,
+		Exclusive:  exclusive,
+		NoWait:     noWait,
+		Arguments:  args,
+	}
+	res := &queueDeclareOk{}
+
+	if err := ch.call(req, res); err != nil {
+		return Queue{}, err
+	}
+
+	if req.wait() {
+		return Queue{
+			Name:      res.Queue,
+			Messages:  int(res.MessageCount),
+			Consumers: int(res.ConsumerCount),
+		}, nil
+	}
+
+	return Queue{Name: name}, nil
+}
+
+/*
+
+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 (ch *Channel) QueueDeclarePassive(name string, durable, autoDelete, exclusive, noWait bool, args Table) (Queue, error) {
+	if err := args.Validate(); err != nil {
+		return Queue{}, err
+	}
+
+	req := &queueDeclare{
+		Queue:      name,
+		Passive:    true,
+		Durable:    durable,
+		AutoDelete: autoDelete,
+		Exclusive:  exclusive,
+		NoWait:     noWait,
+		Arguments:  args,
+	}
+	res := &queueDeclareOk{}
+
+	if err := ch.call(req, res); err != nil {
+		return Queue{}, err
+	}
+
+	if req.wait() {
+		return Queue{
+			Name:      res.Queue,
+			Messages:  int(res.MessageCount),
+			Consumers: int(res.ConsumerCount),
+		}, nil
+	}
+
+	return Queue{Name: name}, nil
+}
+
+/*
+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 (ch *Channel) QueueInspect(name string) (Queue, error) {
+	req := &queueDeclare{
+		Queue:   name,
+		Passive: true,
+	}
+	res := &queueDeclareOk{}
+
+	err := ch.call(req, res)
+
+	state := Queue{
+		Name:      name,
+		Messages:  int(res.MessageCount),
+		Consumers: int(res.ConsumerCount),
+	}
+
+	return state, err
+}
+
+/*
+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 true and the queue could not be bound, the channel will be
+closed with an error.
+
+*/
+func (ch *Channel) QueueBind(name, key, exchange string, noWait bool, args Table) error {
+	if err := args.Validate(); err != nil {
+		return err
+	}
+
+	return ch.call(
+		&queueBind{
+			Queue:      name,
+			Exchange:   exchange,
+			RoutingKey: key,
+			NoWait:     noWait,
+			Arguments:  args,
+		},
+		&queueBindOk{},
+	)
+}
+
+/*
+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 (ch *Channel) QueueUnbind(name, key, exchange string, args Table) error {
+	if err := args.Validate(); err != nil {
+		return err
+	}
+
+	return ch.call(
+		&queueUnbind{
+			Queue:      name,
+			Exchange:   exchange,
+			RoutingKey: key,
+			Arguments:  args,
+		},
+		&queueUnbindOk{},
+	)
+}
+
+/*
+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 (ch *Channel) QueuePurge(name string, noWait bool) (int, error) {
+	req := &queuePurge{
+		Queue:  name,
+		NoWait: noWait,
+	}
+	res := &queuePurgeOk{}
+
+	err := ch.call(req, res)
+
+	return int(res.MessageCount), err
+}
+
+/*
+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 (ch *Channel) QueueDelete(name string, ifUnused, ifEmpty, noWait bool) (int, error) {
+	req := &queueDelete{
+		Queue:    name,
+		IfUnused: ifUnused,
+		IfEmpty:  ifEmpty,
+		NoWait:   noWait,
+	}
+	res := &queueDeleteOk{}
+
+	err := ch.call(req, res)
+
+	return int(res.MessageCount), err
+}
+
+/*
+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.
+
+*/
+func (ch *Channel) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args Table) (<-chan Delivery, error) {
+	// When we return from ch.call, there may be a delivery already for the
+	// consumer that hasn't been added to the consumer hash yet.  Because of
+	// this, we never rely on the server picking a consumer tag for us.
+
+	if err := args.Validate(); err != nil {
+		return nil, err
+	}
+
+	if consumer == "" {
+		consumer = uniqueConsumerTag()
+	}
+
+	req := &basicConsume{
+		Queue:       queue,
+		ConsumerTag: consumer,
+		NoLocal:     noLocal,
+		NoAck:       autoAck,
+		Exclusive:   exclusive,
+		NoWait:      noWait,
+		Arguments:   args,
+	}
+	res := &basicConsumeOk{}
+
+	deliveries := make(chan Delivery)
+
+	ch.consumers.add(consumer, deliveries)
+
+	if err := ch.call(req, res); err != nil {
+		ch.consumers.cancel(consumer)
+		return nil, err
+	}
+
+	return (<-chan Delivery)(deliveries), nil
+}
+
+/*
+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 (ch *Channel) ExchangeDeclare(name, kind string, durable, autoDelete, internal, noWait bool, args Table) error {
+	if err := args.Validate(); err != nil {
+		return err
+	}
+
+	return ch.call(
+		&exchangeDeclare{
+			Exchange:   name,
+			Type:       kind,
+			Passive:    false,
+			Durable:    durable,
+			AutoDelete: autoDelete,
+			Internal:   internal,
+			NoWait:     noWait,
+			Arguments:  args,
+		},
+		&exchangeDeclareOk{},
+	)
+}
+
+/*
+
+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 (ch *Channel) ExchangeDeclarePassive(name, kind string, durable, autoDelete, internal, noWait bool, args Table) error {
+	if err := args.Validate(); err != nil {
+		return err
+	}
+
+	return ch.call(
+		&exchangeDeclare{
+			Exchange:   name,
+			Type:       kind,
+			Passive:    true,
+			Durable:    durable,
+			AutoDelete: autoDelete,
+			Internal:   internal,
+			NoWait:     noWait,
+			Arguments:  args,
+		},
+		&exchangeDeclareOk{},
+	)
+}
+
+/*
+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 (ch *Channel) ExchangeDelete(name string, ifUnused, noWait bool) error {
+	return ch.call(
+		&exchangeDelete{
+			Exchange: name,
+			IfUnused: ifUnused,
+			NoWait:   noWait,
+		},
+		&exchangeDeleteOk{},
+	)
+}
+
+/*
+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 (ch *Channel) ExchangeBind(destination, key, source string, noWait bool, args Table) error {
+	if err := args.Validate(); err != nil {
+		return err
+	}
+
+	return ch.call(
+		&exchangeBind{
+			Destination: destination,
+			Source:      source,
+			RoutingKey:  key,
+			NoWait:      noWait,
+			Arguments:   args,
+		},
+		&exchangeBindOk{},
+	)
+}
+
+/*
+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 (ch *Channel) ExchangeUnbind(destination, key, source string, noWait bool, args Table) error {
+	if err := args.Validate(); err != nil {
+		return err
+	}
+
+	return ch.call(
+		&exchangeUnbind{
+			Destination: destination,
+			Source:      source,
+			RoutingKey:  key,
+			NoWait:      noWait,
+			Arguments:   args,
+		},
+		&exchangeUnbindOk{},
+	)
+}
+
+/*
+Publish sends a Publishing from the client to an exchange on the server.
+
+When you want a single message to be delivered to a single queue, you can
+publish to the default exchange with the routingKey of the queue name.  This is
+because every declared queue gets an implicit route to the default exchange.
+
+Since publishings are asynchronous, any undeliverable message will get returned
+by the server.  Add a listener with Channel.NotifyReturn to handle any
+undeliverable message when calling publish with either the mandatory or
+immediate parameters as true.
+
+Publishings can be undeliverable when the mandatory flag is true and no queue is
+bound that matches the routing key, or when the immediate flag is true and no
+consumer on the matched queue is ready to accept the delivery.
+
+This can return an error when the channel, connection or socket is closed.  The
+error or lack of an error does not indicate whether the server has received this
+publishing.
+
+It is possible for publishing to not reach the broker if the underlying socket
+is shut down without pending publishing packets being flushed from the kernel
+buffers.  The easy way of making it probable that all publishings reach the
+server is to always call Connection.Close before terminating your publishing
+application.  The way to ensure that all publishings reach the server is to add
+a listener to Channel.NotifyPublish and put the channel in confirm mode with
+Channel.Confirm.  Publishing delivery tags and their corresponding
+confirmations start at 1.  Exit when all publishings are confirmed.
+
+When Publish does not return an error and the channel is in confirm mode, the
+internal counter for DeliveryTags with the first confirmation starts at 1.
+
+*/
+func (ch *Channel) Publish(exchange, key string, mandatory, immediate bool, msg Publishing) error {
+	if err := msg.Headers.Validate(); err != nil {
+		return err
+	}
+
+	ch.m.Lock()
+	defer ch.m.Unlock()
+
+	if err := ch.send(&basicPublish{
+		Exchange:   exchange,
+		RoutingKey: key,
+		Mandatory:  mandatory,
+		Immediate:  immediate,
+		Body:       msg.Body,
+		Properties: properties{
+			Headers:         msg.Headers,
+			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,
+		},
+	}); err != nil {
+		return err
+	}
+
+	if ch.confirming {
+		ch.confirms.Publish()
+	}
+
+	return nil
+}
+
+/*
+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 (ch *Channel) Get(queue string, autoAck bool) (msg Delivery, ok bool, err error) {
+	req := &basicGet{Queue: queue, NoAck: autoAck}
+	res := &basicGetOk{}
+	empty := &basicGetEmpty{}
+
+	if err := ch.call(req, res, empty); err != nil {
+		return Delivery{}, false, err
+	}
+
+	if res.DeliveryTag > 0 {
+		return *(newDelivery(ch, res)), true, nil
+	}
+
+	return Delivery{}, false, nil
+}
+
+/*
+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 (ch *Channel) Tx() error {
+	return ch.call(
+		&txSelect{},
+		&txSelectOk{},
+	)
+}
+
+/*
+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 (ch *Channel) TxCommit() error {
+	return ch.call(
+		&txCommit{},
+		&txCommitOk{},
+	)
+}
+
+/*
+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.
+
+*/
+func (ch *Channel) TxRollback() error {
+	return ch.call(
+		&txRollback{},
+		&txRollbackOk{},
+	)
+}
+
+/*
+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 (ch *Channel) Flow(active bool) error {
+	return ch.call(
+		&channelFlow{Active: active},
+		&channelFlowOk{},
+	)
+}
+
+/*
+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.
+
+*/
+func (ch *Channel) Confirm(noWait bool) error {
+	if err := ch.call(
+		&confirmSelect{Nowait: noWait},
+		&confirmSelectOk{},
+	); err != nil {
+		return err
+	}
+
+	ch.confirmM.Lock()
+	ch.confirming = true
+	ch.confirmM.Unlock()
+
+	return nil
+}
+
+/*
+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 (ch *Channel) Recover(requeue bool) error {
+	return ch.call(
+		&basicRecover{Requeue: requeue},
+		&basicRecoverOk{},
+	)
+}
+
+/*
+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 (ch *Channel) Ack(tag uint64, multiple bool) error {
+	ch.m.Lock()
+	defer ch.m.Unlock()
+
+	return ch.send(&basicAck{
+		DeliveryTag: tag,
+		Multiple:    multiple,
+	})
+}
+
+/*
+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 (ch *Channel) Nack(tag uint64, multiple bool, requeue bool) error {
+	ch.m.Lock()
+	defer ch.m.Unlock()
+
+	return ch.send(&basicNack{
+		DeliveryTag: tag,
+		Multiple:    multiple,
+		Requeue:     requeue,
+	})
+}
+
+/*
+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 (ch *Channel) Reject(tag uint64, requeue bool) error {
+	return ch.send(&basicReject{
+		DeliveryTag: tag,
+		Requeue:     requeue,
+	})
+}
diff --git a/src/dma/vendor/github.com/streadway/amqp/confirms.go b/src/dma/vendor/github.com/streadway/amqp/confirms.go
new file mode 100644
index 00000000..06cbaa71
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/confirms.go
@@ -0,0 +1,94 @@
+package amqp
+
+import "sync"
+
+// confirms resequences and notifies one or multiple publisher confirmation listeners
+type confirms struct {
+	m         sync.Mutex
+	listeners []chan Confirmation
+	sequencer map[uint64]Confirmation
+	published uint64
+	expecting uint64
+}
+
+// newConfirms allocates a confirms
+func newConfirms() *confirms {
+	return &confirms{
+		sequencer: map[uint64]Confirmation{},
+		published: 0,
+		expecting: 1,
+	}
+}
+
+func (c *confirms) Listen(l chan Confirmation) {
+	c.m.Lock()
+	defer c.m.Unlock()
+
+	c.listeners = append(c.listeners, l)
+}
+
+// publish increments the publishing counter
+func (c *confirms) Publish() uint64 {
+	c.m.Lock()
+	defer c.m.Unlock()
+
+	c.published++
+	return c.published
+}
+
+// confirm confirms one publishing, increments the expecting delivery tag, and
+// removes bookkeeping for that delivery tag.
+func (c *confirms) confirm(confirmation Confirmation) {
+	delete(c.sequencer, c.expecting)
+	c.expecting++
+	for _, l := range c.listeners {
+		l <- confirmation
+	}
+}
+
+// resequence confirms any out of order delivered confirmations
+func (c *confirms) resequence() {
+	for c.expecting <= c.published {
+		sequenced, found := c.sequencer[c.expecting]
+		if !found {
+			return
+		}
+		c.confirm(sequenced)
+	}
+}
+
+// one confirms one publishing and all following in the publishing sequence
+func (c *confirms) One(confirmed Confirmation) {
+	c.m.Lock()
+	defer c.m.Unlock()
+
+	if c.expecting == confirmed.DeliveryTag {
+		c.confirm(confirmed)
+	} else {
+		c.sequencer[confirmed.DeliveryTag] = confirmed
+	}
+	c.resequence()
+}
+
+// multiple confirms all publishings up until the delivery tag
+func (c *confirms) Multiple(confirmed Confirmation) {
+	c.m.Lock()
+	defer c.m.Unlock()
+
+	for c.expecting <= confirmed.DeliveryTag {
+		c.confirm(Confirmation{c.expecting, confirmed.Ack})
+	}
+	c.resequence()
+}
+
+// Close closes all listeners, discarding any out of sequence confirmations
+func (c *confirms) Close() error {
+	c.m.Lock()
+	defer c.m.Unlock()
+
+	for _, l := range c.listeners {
+		close(l)
+	}
+	c.listeners = nil
+	return nil
+}
diff --git a/src/dma/vendor/github.com/streadway/amqp/connection.go b/src/dma/vendor/github.com/streadway/amqp/connection.go
new file mode 100644
index 00000000..ca1372d0
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/connection.go
@@ -0,0 +1,842 @@
+// 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 http://github.com/streadway/amqp
+
+package amqp
+
+import (
+	"bufio"
+	"crypto/tls"
+	"io"
+	"net"
+	"reflect"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+const (
+	maxChannelMax = (2 << 15) - 1
+
+	defaultHeartbeat         = 10 * time.Second
+	defaultConnectionTimeout = 30 * time.Second
+	defaultProduct           = "https://github.com/streadway/amqp"
+	defaultVersion           = "β"
+	// Safer default that makes channel leaks a lot easier to spot
+	// before they create operational headaches. See https://github.com/rabbitmq/rabbitmq-server/issues/1593.
+	defaultChannelMax = (2 << 10) - 1
+	defaultLocale     = "en_US"
+)
+
+// 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.
+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)
+}
+
+// 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.
+type Connection struct {
+	destructor sync.Once  // shutdown once
+	sendM      sync.Mutex // conn writer mutex
+	m          sync.Mutex // struct field mutex
+
+	conn io.ReadWriteCloser
+
+	rpc       chan message
+	writer    *writer
+	sends     chan time.Time     // timestamps of each frame sent
+	deadlines chan readDeadliner // heartbeater updates read deadlines
+
+	allocator *allocator // id generator valid after openTune
+	channels  map[uint16]*Channel
+
+	noNotify bool // true when we will never notify again
+	closes   []chan *Error
+	blocks   []chan Blocking
+
+	errors chan *Error
+
+	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
+
+	closed int32 // Will be 1 if the connection is closed, 0 otherwise. Should only be accessed as atomic
+}
+
+type readDeadliner interface {
+	SetReadDeadline(time.Time) error
+}
+
+// defaultDial establishes a connection when config.Dial is not provided
+func defaultDial(network, addr string) (net.Conn, error) {
+	conn, err := net.DialTimeout(network, addr, defaultConnectionTimeout)
+	if err != nil {
+		return nil, err
+	}
+
+	// Heartbeating hasn't started yet, don't stall forever on a dead server.
+	// A deadline is set for TLS and AMQP handshaking. After AMQP is established,
+	// the deadline is cleared in openComplete.
+	if err := conn.SetDeadline(time.Now().Add(defaultConnectionTimeout)); err != nil {
+		return nil, err
+	}
+
+	return conn, nil
+}
+
+// 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 Dial(url string) (*Connection, error) {
+	return DialConfig(url, Config{
+		Heartbeat: defaultHeartbeat,
+		Locale:    defaultLocale,
+	})
+}
+
+// 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.
+func DialTLS(url string, amqps *tls.Config) (*Connection, error) {
+	return DialConfig(url, Config{
+		Heartbeat:       defaultHeartbeat,
+		TLSClientConfig: amqps,
+		Locale:          defaultLocale,
+	})
+}
+
+// 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 DialConfig(url string, config Config) (*Connection, error) {
+	var err error
+	var conn net.Conn
+
+	uri, err := ParseURI(url)
+	if err != nil {
+		return nil, err
+	}
+
+	if config.SASL == nil {
+		config.SASL = []Authentication{uri.PlainAuth()}
+	}
+
+	if config.Vhost == "" {
+		config.Vhost = uri.Vhost
+	}
+
+	addr := net.JoinHostPort(uri.Host, strconv.FormatInt(int64(uri.Port), 10))
+
+	dialer := config.Dial
+	if dialer == nil {
+		dialer = defaultDial
+	}
+
+	conn, err = dialer("tcp", addr)
+	if err != nil {
+		return nil, err
+	}
+
+	if uri.Scheme == "amqps" {
+		if config.TLSClientConfig == nil {
+			config.TLSClientConfig = new(tls.Config)
+		}
+
+		// If ServerName has not been specified in TLSClientConfig,
+		// set it to the URI host used for this connection.
+		if config.TLSClientConfig.ServerName == "" {
+			config.TLSClientConfig.ServerName = uri.Host
+		}
+
+		client := tls.Client(conn, config.TLSClientConfig)
+		if err := client.Handshake(); err != nil {
+			conn.Close()
+			return nil, err
+		}
+
+		conn = client
+	}
+
+	return Open(conn, config)
+}
+
+/*
+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 Open(conn io.ReadWriteCloser, config Config) (*Connection, error) {
+	c := &Connection{
+		conn:      conn,
+		writer:    &writer{bufio.NewWriter(conn)},
+		channels:  make(map[uint16]*Channel),
+		rpc:       make(chan message),
+		sends:     make(chan time.Time),
+		errors:    make(chan *Error, 1),
+		deadlines: make(chan readDeadliner, 1),
+	}
+	go c.reader(conn)
+	return c, c.open(config)
+}
+
+/*
+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 (c *Connection) LocalAddr() net.Addr {
+	if conn, ok := c.conn.(interface {
+		LocalAddr() net.Addr
+	}); ok {
+		return conn.LocalAddr()
+	}
+	return &net.TCPAddr{}
+}
+
+// ConnectionState returns basic TLS details of the underlying transport.
+// Returns a zero value when the underlying connection does not implement
+// ConnectionState() tls.ConnectionState.
+func (c *Connection) ConnectionState() tls.ConnectionState {
+	if conn, ok := c.conn.(interface {
+		ConnectionState() tls.ConnectionState
+	}); ok {
+		return conn.ConnectionState()
+	}
+	return tls.ConnectionState{}
+}
+
+/*
+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.
+
+*/
+func (c *Connection) NotifyClose(receiver chan *Error) chan *Error {
+	c.m.Lock()
+	defer c.m.Unlock()
+
+	if c.noNotify {
+		close(receiver)
+	} else {
+		c.closes = append(c.closes, receiver)
+	}
+
+	return receiver
+}
+
+/*
+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.
+
+*/
+func (c *Connection) NotifyBlocked(receiver chan Blocking) chan Blocking {
+	c.m.Lock()
+	defer c.m.Unlock()
+
+	if c.noNotify {
+		close(receiver)
+	} else {
+		c.blocks = append(c.blocks, receiver)
+	}
+
+	return receiver
+}
+
+/*
+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 (c *Connection) Close() error {
+	if c.isClosed() {
+		return ErrClosed
+	}
+
+	defer c.shutdown(nil)
+	return c.call(
+		&connectionClose{
+			ReplyCode: replySuccess,
+			ReplyText: "kthxbai",
+		},
+		&connectionCloseOk{},
+	)
+}
+
+func (c *Connection) closeWith(err *Error) error {
+	if c.isClosed() {
+		return ErrClosed
+	}
+
+	defer c.shutdown(err)
+	return c.call(
+		&connectionClose{
+			ReplyCode: uint16(err.Code),
+			ReplyText: err.Reason,
+		},
+		&connectionCloseOk{},
+	)
+}
+
+func (c *Connection) isClosed() bool {
+	return (atomic.LoadInt32(&c.closed) == 1)
+}
+
+func (c *Connection) send(f frame) error {
+	if c.isClosed() {
+		return ErrClosed
+	}
+
+	c.sendM.Lock()
+	err := c.writer.WriteFrame(f)
+	c.sendM.Unlock()
+
+	if err != nil {
+		// shutdown could be re-entrant from signaling notify chans
+		go c.shutdown(&Error{
+			Code:   FrameError,
+			Reason: err.Error(),
+		})
+	} else {
+		// Broadcast we sent a frame, reducing heartbeats, only
+		// if there is something that can receive - like a non-reentrant
+		// call or if the heartbeater isn't running
+		select {
+		case c.sends <- time.Now():
+		default:
+		}
+	}
+
+	return err
+}
+
+func (c *Connection) shutdown(err *Error) {
+	atomic.StoreInt32(&c.closed, 1)
+
+	c.destructor.Do(func() {
+		c.m.Lock()
+		defer c.m.Unlock()
+
+		if err != nil {
+			for _, c := range c.closes {
+				c <- err
+			}
+		}
+
+		if err != nil {
+			c.errors <- err
+		}
+		// Shutdown handler goroutine can still receive the result.
+		close(c.errors)
+
+		for _, c := range c.closes {
+			close(c)
+		}
+
+		for _, c := range c.blocks {
+			close(c)
+		}
+
+		// Shutdown the channel, but do not use closeChannel() as it calls
+		// releaseChannel() which requires the connection lock.
+		//
+		// Ranging over c.channels and calling releaseChannel() that mutates
+		// c.channels is racy - see commit 6063341 for an example.
+		for _, ch := range c.channels {
+			ch.shutdown(err)
+		}
+
+		c.conn.Close()
+
+		c.channels = map[uint16]*Channel{}
+		c.allocator = newAllocator(1, c.Config.ChannelMax)
+		c.noNotify = true
+	})
+}
+
+// All methods sent to the connection channel should be synchronous so we
+// can handle them directly without a framing component
+func (c *Connection) demux(f frame) {
+	if f.channel() == 0 {
+		c.dispatch0(f)
+	} else {
+		c.dispatchN(f)
+	}
+}
+
+func (c *Connection) dispatch0(f frame) {
+	switch mf := f.(type) {
+	case *methodFrame:
+		switch m := mf.Method.(type) {
+		case *connectionClose:
+			// Send immediately as shutdown will close our side of the writer.
+			c.send(&methodFrame{
+				ChannelId: 0,
+				Method:    &connectionCloseOk{},
+			})
+
+			c.shutdown(newError(m.ReplyCode, m.ReplyText))
+		case *connectionBlocked:
+			for _, c := range c.blocks {
+				c <- Blocking{Active: true, Reason: m.Reason}
+			}
+		case *connectionUnblocked:
+			for _, c := range c.blocks {
+				c <- Blocking{Active: false}
+			}
+		default:
+			c.rpc <- m
+		}
+	case *heartbeatFrame:
+		// kthx - all reads reset our deadline.  so we can drop this
+	default:
+		// lolwat - channel0 only responds to methods and heartbeats
+		c.closeWith(ErrUnexpectedFrame)
+	}
+}
+
+func (c *Connection) dispatchN(f frame) {
+	c.m.Lock()
+	channel := c.channels[f.channel()]
+	c.m.Unlock()
+
+	if channel != nil {
+		channel.recv(channel, f)
+	} else {
+		c.dispatchClosed(f)
+	}
+}
+
+// section 2.3.7: "When a peer decides to close a channel or connection, it
+// sends a Close method.  The receiving peer MUST respond to a Close with a
+// Close-Ok, and then both parties can close their channel or connection.  Note
+// that if peers ignore Close, deadlock can happen when both peers send Close
+// at the same time."
+//
+// When we don't have a channel, so we must respond with close-ok on a close
+// method.  This can happen between a channel exception on an asynchronous
+// method like basic.publish and a synchronous close with channel.close.
+// In that case, we'll get both a channel.close and channel.close-ok in any
+// order.
+func (c *Connection) dispatchClosed(f frame) {
+	// Only consider method frames, drop content/header frames
+	if mf, ok := f.(*methodFrame); ok {
+		switch mf.Method.(type) {
+		case *channelClose:
+			c.send(&methodFrame{
+				ChannelId: f.channel(),
+				Method:    &channelCloseOk{},
+			})
+		case *channelCloseOk:
+			// we are already closed, so do nothing
+		default:
+			// unexpected method on closed channel
+			c.closeWith(ErrClosed)
+		}
+	}
+}
+
+// Reads each frame off the IO and hand off to the connection object that
+// will demux the streams and dispatch to one of the opened channels or
+// handle on channel 0 (the connection channel).
+func (c *Connection) reader(r io.Reader) {
+	buf := bufio.NewReader(r)
+	frames := &reader{buf}
+	conn, haveDeadliner := r.(readDeadliner)
+
+	for {
+		frame, err := frames.ReadFrame()
+
+		if err != nil {
+			c.shutdown(&Error{Code: FrameError, Reason: err.Error()})
+			return
+		}
+
+		c.demux(frame)
+
+		if haveDeadliner {
+			c.deadlines <- conn
+		}
+	}
+}
+
+// Ensures that at least one frame is being sent at the tuned interval with a
+// jitter tolerance of 1s
+func (c *Connection) heartbeater(interval time.Duration, done chan *Error) {
+	const maxServerHeartbeatsInFlight = 3
+
+	var sendTicks <-chan time.Time
+	if interval > 0 {
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		sendTicks = ticker.C
+	}
+
+	lastSent := time.Now()
+
+	for {
+		select {
+		case at, stillSending := <-c.sends:
+			// When actively sending, depend on sent frames to reset server timer
+			if stillSending {
+				lastSent = at
+			} else {
+				return
+			}
+
+		case at := <-sendTicks:
+			// When idle, fill the space with a heartbeat frame
+			if at.Sub(lastSent) > interval-time.Second {
+				if err := c.send(&heartbeatFrame{}); err != nil {
+					// send heartbeats even after close/closeOk so we
+					// tick until the connection starts erroring
+					return
+				}
+			}
+
+		case conn := <-c.deadlines:
+			// When reading, reset our side of the deadline, if we've negotiated one with
+			// a deadline that covers at least 2 server heartbeats
+			if interval > 0 {
+				conn.SetReadDeadline(time.Now().Add(maxServerHeartbeatsInFlight * interval))
+			}
+
+		case <-done:
+			return
+		}
+	}
+}
+
+// Convenience method to inspect the Connection.Properties["capabilities"]
+// Table for server identified capabilities like "basic.ack" or
+// "confirm.select".
+func (c *Connection) isCapable(featureName string) bool {
+	capabilities, _ := c.Properties["capabilities"].(Table)
+	hasFeature, _ := capabilities[featureName].(bool)
+	return hasFeature
+}
+
+// allocateChannel records but does not open a new channel with a unique id.
+// This method is the initial part of the channel lifecycle and paired with
+// releaseChannel
+func (c *Connection) allocateChannel() (*Channel, error) {
+	c.m.Lock()
+	defer c.m.Unlock()
+
+	if c.isClosed() {
+		return nil, ErrClosed
+	}
+
+	id, ok := c.allocator.next()
+	if !ok {
+		return nil, ErrChannelMax
+	}
+
+	ch := newChannel(c, uint16(id))
+	c.channels[uint16(id)] = ch
+
+	return ch, nil
+}
+
+// releaseChannel removes a channel from the registry as the final part of the
+// channel lifecycle
+func (c *Connection) releaseChannel(id uint16) {
+	c.m.Lock()
+	defer c.m.Unlock()
+
+	delete(c.channels, id)
+	c.allocator.release(int(id))
+}
+
+// openChannel allocates and opens a channel, must be paired with closeChannel
+func (c *Connection) openChannel() (*Channel, error) {
+	ch, err := c.allocateChannel()
+	if err != nil {
+		return nil, err
+	}
+
+	if err := ch.open(); err != nil {
+		c.releaseChannel(ch.id)
+		return nil, err
+	}
+	return ch, nil
+}
+
+// closeChannel releases and initiates a shutdown of the channel.  All channel
+// closures should be initiated here for proper channel lifecycle management on
+// this connection.
+func (c *Connection) closeChannel(ch *Channel, e *Error) {
+	ch.shutdown(e)
+	c.releaseChannel(ch.id)
+}
+
+/*
+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 (c *Connection) Channel() (*Channel, error) {
+	return c.openChannel()
+}
+
+func (c *Connection) call(req message, res ...message) error {
+	// Special case for when the protocol header frame is sent insted of a
+	// request method
+	if req != nil {
+		if err := c.send(&methodFrame{ChannelId: 0, Method: req}); err != nil {
+			return err
+		}
+	}
+
+	select {
+	case err, ok := <-c.errors:
+		if !ok {
+			return ErrClosed
+		}
+		return err
+
+	case msg := <-c.rpc:
+		// Try to match one of the result types
+		for _, try := range res {
+			if reflect.TypeOf(msg) == reflect.TypeOf(try) {
+				// *res = *msg
+				vres := reflect.ValueOf(try).Elem()
+				vmsg := reflect.ValueOf(msg).Elem()
+				vres.Set(vmsg)
+				return nil
+			}
+		}
+		return ErrCommandInvalid
+	}
+	// unreachable
+}
+
+//    Connection          = open-Connection *use-Connection close-Connection
+//    open-Connection     = C:protocol-header
+//                          S:START C:START-OK
+//                          *challenge
+//                          S:TUNE C:TUNE-OK
+//                          C:OPEN S:OPEN-OK
+//    challenge           = S:SECURE C:SECURE-OK
+//    use-Connection      = *channel
+//    close-Connection    = C:CLOSE S:CLOSE-OK
+//                        / S:CLOSE C:CLOSE-OK
+func (c *Connection) open(config Config) error {
+	if err := c.send(&protocolHeader{}); err != nil {
+		return err
+	}
+
+	return c.openStart(config)
+}
+
+func (c *Connection) openStart(config Config) error {
+	start := &connectionStart{}
+
+	if err := c.call(nil, start); err != nil {
+		return err
+	}
+
+	c.Major = int(start.VersionMajor)
+	c.Minor = int(start.VersionMinor)
+	c.Properties = Table(start.ServerProperties)
+	c.Locales = strings.Split(start.Locales, " ")
+
+	// eventually support challenge/response here by also responding to
+	// connectionSecure.
+	auth, ok := pickSASLMechanism(config.SASL, strings.Split(start.Mechanisms, " "))
+	if !ok {
+		return ErrSASL
+	}
+
+	// Save this mechanism off as the one we chose
+	c.Config.SASL = []Authentication{auth}
+
+	// Set the connection locale to client locale
+	c.Config.Locale = config.Locale
+
+	return c.openTune(config, auth)
+}
+
+func (c *Connection) openTune(config Config, auth Authentication) error {
+	if len(config.Properties) == 0 {
+		config.Properties = Table{
+			"product": defaultProduct,
+			"version": defaultVersion,
+		}
+	}
+
+	config.Properties["capabilities"] = Table{
+		"connection.blocked":     true,
+		"consumer_cancel_notify": true,
+	}
+
+	ok := &connectionStartOk{
+		ClientProperties: config.Properties,
+		Mechanism:        auth.Mechanism(),
+		Response:         auth.Response(),
+		Locale:           config.Locale,
+	}
+	tune := &connectionTune{}
+
+	if err := c.call(ok, tune); err != nil {
+		// per spec, a connection can only be closed when it has been opened
+		// so at this point, we know it's an auth error, but the socket
+		// was closed instead.  Return a meaningful error.
+		return ErrCredentials
+	}
+
+	// When the server and client both use default 0, then the max channel is
+	// only limited by uint16.
+	c.Config.ChannelMax = pick(config.ChannelMax, int(tune.ChannelMax))
+	if c.Config.ChannelMax == 0 {
+		c.Config.ChannelMax = defaultChannelMax
+	}
+	c.Config.ChannelMax = min(c.Config.ChannelMax, maxChannelMax)
+
+	// Frame size includes headers and end byte (len(payload)+8), even if
+	// this is less than FrameMinSize, use what the server sends because the
+	// alternative is to stop the handshake here.
+	c.Config.FrameSize = pick(config.FrameSize, int(tune.FrameMax))
+
+	// Save this off for resetDeadline()
+	c.Config.Heartbeat = time.Second * time.Duration(pick(
+		int(config.Heartbeat/time.Second),
+		int(tune.Heartbeat)))
+
+	// "The client should start sending heartbeats after receiving a
+	// Connection.Tune method"
+	go c.heartbeater(c.Config.Heartbeat, c.NotifyClose(make(chan *Error, 1)))
+
+	if err := c.send(&methodFrame{
+		ChannelId: 0,
+		Method: &connectionTuneOk{
+			ChannelMax: uint16(c.Config.ChannelMax),
+			FrameMax:   uint32(c.Config.FrameSize),
+			Heartbeat:  uint16(c.Config.Heartbeat / time.Second),
+		},
+	}); err != nil {
+		return err
+	}
+
+	return c.openVhost(config)
+}
+
+func (c *Connection) openVhost(config Config) error {
+	req := &connectionOpen{VirtualHost: config.Vhost}
+	res := &connectionOpenOk{}
+
+	if err := c.call(req, res); err != nil {
+		// Cannot be closed yet, but we know it's a vhost problem
+		return ErrVhost
+	}
+
+	c.Config.Vhost = config.Vhost
+
+	return c.openComplete()
+}
+
+// openComplete performs any final Connection initialization dependent on the
+// connection handshake and clears any state needed for TLS and AMQP handshaking.
+func (c *Connection) openComplete() error {
+	// We clear the deadlines and let the heartbeater reset the read deadline if requested.
+	// RabbitMQ uses TCP flow control at this point for pushback so Writes can
+	// intentionally block.
+	if deadliner, ok := c.conn.(interface {
+		SetDeadline(time.Time) error
+	}); ok {
+		_ = deadliner.SetDeadline(time.Time{})
+	}
+
+	c.allocator = newAllocator(1, c.Config.ChannelMax)
+	return nil
+}
+
+func max(a, b int) int {
+	if a > b {
+		return a
+	}
+	return b
+}
+
+func min(a, b int) int {
+	if a < b {
+		return a
+	}
+	return b
+}
+
+func pick(client, server int) int {
+	if client == 0 || server == 0 {
+		return max(client, server)
+	}
+	return min(client, server)
+}
diff --git a/src/dma/vendor/github.com/streadway/amqp/consumers.go b/src/dma/vendor/github.com/streadway/amqp/consumers.go
new file mode 100644
index 00000000..887ac749
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/consumers.go
@@ -0,0 +1,142 @@
+// 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 http://github.com/streadway/amqp
+
+package amqp
+
+import (
+	"os"
+	"strconv"
+	"sync"
+	"sync/atomic"
+)
+
+var consumerSeq uint64
+
+const consumerTagLengthMax = 0xFF // see writeShortstr
+
+func uniqueConsumerTag() string {
+	return commandNameBasedUniqueConsumerTag(os.Args[0])
+}
+
+func commandNameBasedUniqueConsumerTag(commandName string) string {
+	tagPrefix := "ctag-"
+	tagInfix := commandName
+	tagSuffix := "-" + strconv.FormatUint(atomic.AddUint64(&consumerSeq, 1), 10)
+
+	if len(tagPrefix)+len(tagInfix)+len(tagSuffix) > consumerTagLengthMax {
+		tagInfix = "streadway/amqp"
+	}
+
+	return tagPrefix + tagInfix + tagSuffix
+}
+
+type consumerBuffers map[string]chan *Delivery
+
+// Concurrent type that manages the consumerTag ->
+// ingress consumerBuffer mapping
+type consumers struct {
+	sync.WaitGroup               // one for buffer
+	closed         chan struct{} // signal buffer
+
+	sync.Mutex // protects below
+	chans      consumerBuffers
+}
+
+func makeConsumers() *consumers {
+	return &consumers{
+		closed: make(chan struct{}),
+		chans:  make(consumerBuffers),
+	}
+}
+
+func (subs *consumers) buffer(in chan *Delivery, out chan Delivery) {
+	defer close(out)
+	defer subs.Done()
+
+	var inflight = in
+	var queue []*Delivery
+
+	for delivery := range in {
+		queue = append(queue, delivery)
+
+		for len(queue) > 0 {
+			select {
+			case <-subs.closed:
+				// closed before drained, drop in-flight
+				return
+
+			case delivery, consuming := <-inflight:
+				if consuming {
+					queue = append(queue, delivery)
+				} else {
+					inflight = nil
+				}
+
+			case out <- *queue[0]:
+				queue = queue[1:]
+			}
+		}
+	}
+}
+
+// On key conflict, close the previous channel.
+func (subs *consumers) add(tag string, consumer chan Delivery) {
+	subs.Lock()
+	defer subs.Unlock()
+
+	if prev, found := subs.chans[tag]; found {
+		close(prev)
+	}
+
+	in := make(chan *Delivery)
+	subs.chans[tag] = in
+
+	subs.Add(1)
+	go subs.buffer(in, consumer)
+}
+
+func (subs *consumers) cancel(tag string) (found bool) {
+	subs.Lock()
+	defer subs.Unlock()
+
+	ch, found := subs.chans[tag]
+
+	if found {
+		delete(subs.chans, tag)
+		close(ch)
+	}
+
+	return found
+}
+
+func (subs *consumers) close() {
+	subs.Lock()
+	defer subs.Unlock()
+
+	close(subs.closed)
+
+	for tag, ch := range subs.chans {
+		delete(subs.chans, tag)
+		close(ch)
+	}
+
+	subs.Wait()
+}
+
+// Sends a delivery to a the consumer identified by `tag`.
+// If unbuffered channels are used for Consume this method
+// could block all deliveries until the consumer
+// receives on the other end of the channel.
+func (subs *consumers) send(tag string, msg *Delivery) bool {
+	subs.Lock()
+	defer subs.Unlock()
+
+	buffer, found := subs.chans[tag]
+	if found {
+		buffer <- msg
+	}
+
+	return found
+}
diff --git a/src/dma/vendor/github.com/streadway/amqp/delivery.go b/src/dma/vendor/github.com/streadway/amqp/delivery.go
new file mode 100644
index 00000000..304c8346
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/delivery.go
@@ -0,0 +1,173 @@
+// 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 http://github.com/streadway/amqp
+
+package amqp
+
+import (
+	"errors"
+	"time"
+)
+
+var errDeliveryNotInitialized = errors.New("delivery not initialized")
+
+// 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 Acknowledger interface {
+	Ack(tag uint64, multiple bool) error
+	Nack(tag uint64, multiple bool, requeue bool) error
+	Reject(tag uint64, requeue bool) error
+}
+
+// Delivery captures the fields for a previously delivered message resident in
+// a queue to be delivered by the server to a consumer from Channel.Consume or
+// Channel.Get.
+type Delivery struct {
+	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 exhange
+	RoutingKey  string // basic.publish routing key
+
+	Body []byte
+}
+
+func newDelivery(channel *Channel, msg messageWithContent) *Delivery {
+	props, body := msg.getContent()
+
+	delivery := Delivery{
+		Acknowledger: channel,
+
+		Headers:         props.Headers,
+		ContentType:     props.ContentType,
+		ContentEncoding: props.ContentEncoding,
+		DeliveryMode:    props.DeliveryMode,
+		Priority:        props.Priority,
+		CorrelationId:   props.CorrelationId,
+		ReplyTo:         props.ReplyTo,
+		Expiration:      props.Expiration,
+		MessageId:       props.MessageId,
+		Timestamp:       props.Timestamp,
+		Type:            props.Type,
+		UserId:          props.UserId,
+		AppId:           props.AppId,
+
+		Body: body,
+	}
+
+	// Properties for the delivery types
+	switch m := msg.(type) {
+	case *basicDeliver:
+		delivery.ConsumerTag = m.ConsumerTag
+		delivery.DeliveryTag = m.DeliveryTag
+		delivery.Redelivered = m.Redelivered
+		delivery.Exchange = m.Exchange
+		delivery.RoutingKey = m.RoutingKey
+
+	case *basicGetOk:
+		delivery.MessageCount = m.MessageCount
+		delivery.DeliveryTag = m.DeliveryTag
+		delivery.Redelivered = m.Redelivered
+		delivery.Exchange = m.Exchange
+		delivery.RoutingKey = m.RoutingKey
+	}
+
+	return &delivery
+}
+
+/*
+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 (d Delivery) Ack(multiple bool) error {
+	if d.Acknowledger == nil {
+		return errDeliveryNotInitialized
+	}
+	return d.Acknowledger.Ack(d.DeliveryTag, multiple)
+}
+
+/*
+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.
+*/
+func (d Delivery) Reject(requeue bool) error {
+	if d.Acknowledger == nil {
+		return errDeliveryNotInitialized
+	}
+	return d.Acknowledger.Reject(d.DeliveryTag, requeue)
+}
+
+/*
+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 (d Delivery) Nack(multiple, requeue bool) error {
+	if d.Acknowledger == nil {
+		return errDeliveryNotInitialized
+	}
+	return d.Acknowledger.Nack(d.DeliveryTag, multiple, requeue)
+}
diff --git a/src/dma/vendor/github.com/streadway/amqp/doc.go b/src/dma/vendor/github.com/streadway/amqp/doc.go
new file mode 100644
index 00000000..76bf3e58
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/doc.go
@@ -0,0 +1,108 @@
+// 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 http://github.com/streadway/amqp
+
+/*
+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 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
+
+*/
+package amqp
diff --git a/src/dma/vendor/github.com/streadway/amqp/fuzz.go b/src/dma/vendor/github.com/streadway/amqp/fuzz.go
new file mode 100644
index 00000000..16e626ce
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/fuzz.go
@@ -0,0 +1,17 @@
+// +build gofuzz
+
+package amqp
+
+import "bytes"
+
+func Fuzz(data []byte) int {
+	r := reader{bytes.NewReader(data)}
+	frame, err := r.ReadFrame()
+	if err != nil {
+		if frame != nil {
+			panic("frame is not nil")
+		}
+		return 0
+	}
+	return 1
+}
diff --git a/src/dma/vendor/github.com/streadway/amqp/gen.sh b/src/dma/vendor/github.com/streadway/amqp/gen.sh
new file mode 100755
index 00000000..d46e19bd
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/gen.sh
@@ -0,0 +1,2 @@
+#!/bin/sh
+go run spec/gen.go < spec/amqp0-9-1.stripped.extended.xml | gofmt > spec091.go
diff --git a/src/dma/vendor/github.com/streadway/amqp/pre-commit b/src/dma/vendor/github.com/streadway/amqp/pre-commit
new file mode 100755
index 00000000..7607f467
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/pre-commit
@@ -0,0 +1,29 @@
+#!/bin/sh
+
+GOFMT_FILES=$(gofmt -l .)
+if [ -n "${GOFMT_FILES}" ]; then
+  printf >&2 'gofmt failed for the following files:\n%s\n\nplease run "gofmt -w ." on your changes before committing.\n' "${GOFMT_FILES}"
+  exit 1
+fi
+
+GOLINT_ERRORS=$(golint ./... | grep -v "Id should be")
+if [ -n "${GOLINT_ERRORS}" ]; then
+  printf >&2 'golint failed for the following reasons:\n%s\n\nplease run 'golint ./...' on your changes before committing.\n' "${GOLINT_ERRORS}"
+  exit 1
+fi
+
+GOVET_ERRORS=$(go tool vet *.go 2>&1)
+if [ -n "${GOVET_ERRORS}" ]; then
+  printf >&2 'go vet failed for the following reasons:\n%s\n\nplease run "go tool vet *.go" on your changes before committing.\n' "${GOVET_ERRORS}"
+  exit 1
+fi
+
+if [ -z "${NOTEST}" ]; then
+  printf >&2 'Running short tests...\n'
+  env AMQP_URL= go test -short -v | egrep 'PASS|ok'
+
+  if [ $? -ne 0 ]; then
+    printf >&2 'go test failed, please fix before committing.\n'
+    exit 1
+  fi
+fi
diff --git a/src/dma/vendor/github.com/streadway/amqp/read.go b/src/dma/vendor/github.com/streadway/amqp/read.go
new file mode 100644
index 00000000..3aa0b338
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/read.go
@@ -0,0 +1,456 @@
+// 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 http://github.com/streadway/amqp
+
+package amqp
+
+import (
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"io"
+	"time"
+)
+
+/*
+Reads a frame from an input stream and returns an interface that can be cast into
+one of the following:
+
+   methodFrame
+   PropertiesFrame
+   bodyFrame
+   heartbeatFrame
+
+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.
+*/
+func (r *reader) ReadFrame() (frame frame, err error) {
+	var scratch [7]byte
+
+	if _, err = io.ReadFull(r.r, scratch[:7]); err != nil {
+		return
+	}
+
+	typ := uint8(scratch[0])
+	channel := binary.BigEndian.Uint16(scratch[1:3])
+	size := binary.BigEndian.Uint32(scratch[3:7])
+
+	switch typ {
+	case frameMethod:
+		if frame, err = r.parseMethodFrame(channel, size); err != nil {
+			return
+		}
+
+	case frameHeader:
+		if frame, err = r.parseHeaderFrame(channel, size); err != nil {
+			return
+		}
+
+	case frameBody:
+		if frame, err = r.parseBodyFrame(channel, size); err != nil {
+			return nil, err
+		}
+
+	case frameHeartbeat:
+		if frame, err = r.parseHeartbeatFrame(channel, size); err != nil {
+			return
+		}
+
+	default:
+		return nil, ErrFrame
+	}
+
+	if _, err = io.ReadFull(r.r, scratch[:1]); err != nil {
+		return nil, err
+	}
+
+	if scratch[0] != frameEnd {
+		return nil, ErrFrame
+	}
+
+	return
+}
+
+func readShortstr(r io.Reader) (v string, err error) {
+	var length uint8
+	if err = binary.Read(r, binary.BigEndian, &length); err != nil {
+		return
+	}
+
+	bytes := make([]byte, length)
+	if _, err = io.ReadFull(r, bytes); err != nil {
+		return
+	}
+	return string(bytes), nil
+}
+
+func readLongstr(r io.Reader) (v string, err error) {
+	var length uint32
+	if err = binary.Read(r, binary.BigEndian, &length); err != nil {
+		return
+	}
+
+	// slices can't be longer than max int32 value
+	if length > (^uint32(0) >> 1) {
+		return
+	}
+
+	bytes := make([]byte, length)
+	if _, err = io.ReadFull(r, bytes); err != nil {
+		return
+	}
+	return string(bytes), nil
+}
+
+func readDecimal(r io.Reader) (v Decimal, err error) {
+	if err = binary.Read(r, binary.BigEndian, &v.Scale); err != nil {
+		return
+	}
+	if err = binary.Read(r, binary.BigEndian, &v.Value); err != nil {
+		return
+	}
+	return
+}
+
+func readFloat32(r io.Reader) (v float32, err error) {
+	if err = binary.Read(r, binary.BigEndian, &v); err != nil {
+		return
+	}
+	return
+}
+
+func readFloat64(r io.Reader) (v float64, err error) {
+	if err = binary.Read(r, binary.BigEndian, &v); err != nil {
+		return
+	}
+	return
+}
+
+func readTimestamp(r io.Reader) (v time.Time, err error) {
+	var sec int64
+	if err = binary.Read(r, binary.BigEndian, &sec); err != nil {
+		return
+	}
+	return time.Unix(sec, 0), nil
+}
+
+/*
+'A': []interface{}
+'D': Decimal
+'F': Table
+'I': int32
+'S': string
+'T': time.Time
+'V': nil
+'b': byte
+'d': float64
+'f': float32
+'l': int64
+'s': int16
+'t': bool
+'x': []byte
+*/
+func readField(r io.Reader) (v interface{}, err error) {
+	var typ byte
+	if err = binary.Read(r, binary.BigEndian, &typ); err != nil {
+		return
+	}
+
+	switch typ {
+	case 't':
+		var value uint8
+		if err = binary.Read(r, binary.BigEndian, &value); err != nil {
+			return
+		}
+		return (value != 0), nil
+
+	case 'b':
+		var value [1]byte
+		if _, err = io.ReadFull(r, value[0:1]); err != nil {
+			return
+		}
+		return value[0], nil
+
+	case 's':
+		var value int16
+		if err = binary.Read(r, binary.BigEndian, &value); err != nil {
+			return
+		}
+		return value, nil
+
+	case 'I':
+		var value int32
+		if err = binary.Read(r, binary.BigEndian, &value); err != nil {
+			return
+		}
+		return value, nil
+
+	case 'l':
+		var value int64
+		if err = binary.Read(r, binary.BigEndian, &value); err != nil {
+			return
+		}
+		return value, nil
+
+	case 'f':
+		var value float32
+		if err = binary.Read(r, binary.BigEndian, &value); err != nil {
+			return
+		}
+		return value, nil
+
+	case 'd':
+		var value float64
+		if err = binary.Read(r, binary.BigEndian, &value); err != nil {
+			return
+		}
+		return value, nil
+
+	case 'D':
+		return readDecimal(r)
+
+	case 'S':
+		return readLongstr(r)
+
+	case 'A':
+		return readArray(r)
+
+	case 'T':
+		return readTimestamp(r)
+
+	case 'F':
+		return readTable(r)
+
+	case 'x':
+		var len int32
+		if err = binary.Read(r, binary.BigEndian, &len); err != nil {
+			return nil, err
+		}
+
+		value := make([]byte, len)
+		if _, err = io.ReadFull(r, value); err != nil {
+			return nil, err
+		}
+		return value, err
+
+	case 'V':
+		return nil, nil
+	}
+
+	return nil, ErrSyntax
+}
+
+/*
+	Field tables are long strings that contain packed name-value pairs.  The
+	name-value pairs are encoded as short string defining the name, and octet
+	defining the values type and then the value itself.   The valid field types for
+	tables are an extension of the native integer, bit, string, and timestamp
+	types, and are shown in the grammar.  Multi-octet integer fields are always
+	held in network byte order.
+*/
+func readTable(r io.Reader) (table Table, err error) {
+	var nested bytes.Buffer
+	var str string
+
+	if str, err = readLongstr(r); err != nil {
+		return
+	}
+
+	nested.Write([]byte(str))
+
+	table = make(Table)
+
+	for nested.Len() > 0 {
+		var key string
+		var value interface{}
+
+		if key, err = readShortstr(&nested); err != nil {
+			return
+		}
+
+		if value, err = readField(&nested); err != nil {
+			return
+		}
+
+		table[key] = value
+	}
+
+	return
+}
+
+func readArray(r io.Reader) ([]interface{}, error) {
+	var (
+		size uint32
+		err  error
+	)
+
+	if err = binary.Read(r, binary.BigEndian, &size); err != nil {
+		return nil, err
+	}
+
+	var (
+		lim   = &io.LimitedReader{R: r, N: int64(size)}
+		arr   = []interface{}{}
+		field interface{}
+	)
+
+	for {
+		if field, err = readField(lim); err != nil {
+			if err == io.EOF {
+				break
+			}
+			return nil, err
+		}
+		arr = append(arr, field)
+	}
+
+	return arr, nil
+}
+
+// Checks if this bit mask matches the flags bitset
+func hasProperty(mask uint16, prop int) bool {
+	return int(mask)&prop > 0
+}
+
+func (r *reader) parseHeaderFrame(channel uint16, size uint32) (frame frame, err error) {
+	hf := &headerFrame{
+		ChannelId: channel,
+	}
+
+	if err = binary.Read(r.r, binary.BigEndian, &hf.ClassId); err != nil {
+		return
+	}
+
+	if err = binary.Read(r.r, binary.BigEndian, &hf.weight); err != nil {
+		return
+	}
+
+	if err = binary.Read(r.r, binary.BigEndian, &hf.Size); err != nil {
+		return
+	}
+
+	var flags uint16
+
+	if err = binary.Read(r.r, binary.BigEndian, &flags); err != nil {
+		return
+	}
+
+	if hasProperty(flags, flagContentType) {
+		if hf.Properties.ContentType, err = readShortstr(r.r); err != nil {
+			return
+		}
+	}
+	if hasProperty(flags, flagContentEncoding) {
+		if hf.Properties.ContentEncoding, err = readShortstr(r.r); err != nil {
+			return
+		}
+	}
+	if hasProperty(flags, flagHeaders) {
+		if hf.Properties.Headers, err = readTable(r.r); err != nil {
+			return
+		}
+	}
+	if hasProperty(flags, flagDeliveryMode) {
+		if err = binary.Read(r.r, binary.BigEndian, &hf.Properties.DeliveryMode); err != nil {
+			return
+		}
+	}
+	if hasProperty(flags, flagPriority) {
+		if err = binary.Read(r.r, binary.BigEndian, &hf.Properties.Priority); err != nil {
+			return
+		}
+	}
+	if hasProperty(flags, flagCorrelationId) {
+		if hf.Properties.CorrelationId, err = readShortstr(r.r); err != nil {
+			return
+		}
+	}
+	if hasProperty(flags, flagReplyTo) {
+		if hf.Properties.ReplyTo, err = readShortstr(r.r); err != nil {
+			return
+		}
+	}
+	if hasProperty(flags, flagExpiration) {
+		if hf.Properties.Expiration, err = readShortstr(r.r); err != nil {
+			return
+		}
+	}
+	if hasProperty(flags, flagMessageId) {
+		if hf.Properties.MessageId, err = readShortstr(r.r); err != nil {
+			return
+		}
+	}
+	if hasProperty(flags, flagTimestamp) {
+		if hf.Properties.Timestamp, err = readTimestamp(r.r); err != nil {
+			return
+		}
+	}
+	if hasProperty(flags, flagType) {
+		if hf.Properties.Type, err = readShortstr(r.r); err != nil {
+			return
+		}
+	}
+	if hasProperty(flags, flagUserId) {
+		if hf.Properties.UserId, err = readShortstr(r.r); err != nil {
+			return
+		}
+	}
+	if hasProperty(flags, flagAppId) {
+		if hf.Properties.AppId, err = readShortstr(r.r); err != nil {
+			return
+		}
+	}
+	if hasProperty(flags, flagReserved1) {
+		if hf.Properties.reserved1, err = readShortstr(r.r); err != nil {
+			return
+		}
+	}
+
+	return hf, nil
+}
+
+func (r *reader) parseBodyFrame(channel uint16, size uint32) (frame frame, err error) {
+	bf := &bodyFrame{
+		ChannelId: channel,
+		Body:      make([]byte, size),
+	}
+
+	if _, err = io.ReadFull(r.r, bf.Body); err != nil {
+		return nil, err
+	}
+
+	return bf, nil
+}
+
+var errHeartbeatPayload = errors.New("Heartbeats should not have a payload")
+
+func (r *reader) parseHeartbeatFrame(channel uint16, size uint32) (frame frame, err error) {
+	hf := &heartbeatFrame{
+		ChannelId: channel,
+	}
+
+	if size > 0 {
+		return nil, errHeartbeatPayload
+	}
+
+	return hf, nil
+}
diff --git a/src/dma/vendor/github.com/streadway/amqp/return.go b/src/dma/vendor/github.com/streadway/amqp/return.go
new file mode 100644
index 00000000..10dcedb2
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/return.go
@@ -0,0 +1,64 @@
+// 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 http://github.com/streadway/amqp
+
+package amqp
+
+import (
+	"time"
+)
+
+// 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 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
+}
+
+func newReturn(msg basicReturn) *Return {
+	props, body := msg.getContent()
+
+	return &Return{
+		ReplyCode:  msg.ReplyCode,
+		ReplyText:  msg.ReplyText,
+		Exchange:   msg.Exchange,
+		RoutingKey: msg.RoutingKey,
+
+		Headers:         props.Headers,
+		ContentType:     props.ContentType,
+		ContentEncoding: props.ContentEncoding,
+		DeliveryMode:    props.DeliveryMode,
+		Priority:        props.Priority,
+		CorrelationId:   props.CorrelationId,
+		ReplyTo:         props.ReplyTo,
+		Expiration:      props.Expiration,
+		MessageId:       props.MessageId,
+		Timestamp:       props.Timestamp,
+		Type:            props.Type,
+		UserId:          props.UserId,
+		AppId:           props.AppId,
+
+		Body: body,
+	}
+}
diff --git a/src/dma/vendor/github.com/streadway/amqp/spec091.go b/src/dma/vendor/github.com/streadway/amqp/spec091.go
new file mode 100644
index 00000000..cd53ebe7
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/spec091.go
@@ -0,0 +1,3306 @@
+// 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 http://github.com/streadway/amqp
+
+/* GENERATED FILE - DO NOT EDIT */
+/* Rebuild from the spec/gen.go tool */
+
+package amqp
+
+import (
+	"encoding/binary"
+	"fmt"
+	"io"
+)
+
+// Error codes that can be sent from the server during a connection or
+// channel exception or used by the client to indicate a class of error like
+// ErrCredentials.  The text of the error is likely more interesting than
+// these constants.
+const (
+	frameMethod        = 1
+	frameHeader        = 2
+	frameBody          = 3
+	frameHeartbeat     = 8
+	frameMinSize       = 4096
+	frameEnd           = 206
+	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
+)
+
+func isSoftExceptionCode(code int) bool {
+	switch code {
+	case 311:
+		return true
+	case 312:
+		return true
+	case 313:
+		return true
+	case 403:
+		return true
+	case 404:
+		return true
+	case 405:
+		return true
+	case 406:
+		return true
+
+	}
+	return false
+}
+
+type connectionStart struct {
+	VersionMajor     byte
+	VersionMinor     byte
+	ServerProperties Table
+	Mechanisms       string
+	Locales          string
+}
+
+func (msg *connectionStart) id() (uint16, uint16) {
+	return 10, 10
+}
+
+func (msg *connectionStart) wait() bool {
+	return true
+}
+
+func (msg *connectionStart) write(w io.Writer) (err error) {
+
+	if err = binary.Write(w, binary.BigEndian, msg.VersionMajor); err != nil {
+		return
+	}
+	if err = binary.Write(w, binary.BigEndian, msg.VersionMinor); err != nil {
+		return
+	}
+
+	if err = writeTable(w, msg.ServerProperties); err != nil {
+		return
+	}
+
+	if err = writeLongstr(w, msg.Mechanisms); err != nil {
+		return
+	}
+	if err = writeLongstr(w, msg.Locales); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *connectionStart) read(r io.Reader) (err error) {
+
+	if err = binary.Read(r, binary.BigEndian, &msg.VersionMajor); err != nil {
+		return
+	}
+	if err = binary.Read(r, binary.BigEndian, &msg.VersionMinor); err != nil {
+		return
+	}
+
+	if msg.ServerProperties, err = readTable(r); err != nil {
+		return
+	}
+
+	if msg.Mechanisms, err = readLongstr(r); err != nil {
+		return
+	}
+	if msg.Locales, err = readLongstr(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type connectionStartOk struct {
+	ClientProperties Table
+	Mechanism        string
+	Response         string
+	Locale           string
+}
+
+func (msg *connectionStartOk) id() (uint16, uint16) {
+	return 10, 11
+}
+
+func (msg *connectionStartOk) wait() bool {
+	return true
+}
+
+func (msg *connectionStartOk) write(w io.Writer) (err error) {
+
+	if err = writeTable(w, msg.ClientProperties); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.Mechanism); err != nil {
+		return
+	}
+
+	if err = writeLongstr(w, msg.Response); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.Locale); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *connectionStartOk) read(r io.Reader) (err error) {
+
+	if msg.ClientProperties, err = readTable(r); err != nil {
+		return
+	}
+
+	if msg.Mechanism, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if msg.Response, err = readLongstr(r); err != nil {
+		return
+	}
+
+	if msg.Locale, err = readShortstr(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type connectionSecure struct {
+	Challenge string
+}
+
+func (msg *connectionSecure) id() (uint16, uint16) {
+	return 10, 20
+}
+
+func (msg *connectionSecure) wait() bool {
+	return true
+}
+
+func (msg *connectionSecure) write(w io.Writer) (err error) {
+
+	if err = writeLongstr(w, msg.Challenge); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *connectionSecure) read(r io.Reader) (err error) {
+
+	if msg.Challenge, err = readLongstr(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type connectionSecureOk struct {
+	Response string
+}
+
+func (msg *connectionSecureOk) id() (uint16, uint16) {
+	return 10, 21
+}
+
+func (msg *connectionSecureOk) wait() bool {
+	return true
+}
+
+func (msg *connectionSecureOk) write(w io.Writer) (err error) {
+
+	if err = writeLongstr(w, msg.Response); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *connectionSecureOk) read(r io.Reader) (err error) {
+
+	if msg.Response, err = readLongstr(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type connectionTune struct {
+	ChannelMax uint16
+	FrameMax   uint32
+	Heartbeat  uint16
+}
+
+func (msg *connectionTune) id() (uint16, uint16) {
+	return 10, 30
+}
+
+func (msg *connectionTune) wait() bool {
+	return true
+}
+
+func (msg *connectionTune) write(w io.Writer) (err error) {
+
+	if err = binary.Write(w, binary.BigEndian, msg.ChannelMax); err != nil {
+		return
+	}
+
+	if err = binary.Write(w, binary.BigEndian, msg.FrameMax); err != nil {
+		return
+	}
+
+	if err = binary.Write(w, binary.BigEndian, msg.Heartbeat); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *connectionTune) read(r io.Reader) (err error) {
+
+	if err = binary.Read(r, binary.BigEndian, &msg.ChannelMax); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &msg.FrameMax); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &msg.Heartbeat); err != nil {
+		return
+	}
+
+	return
+}
+
+type connectionTuneOk struct {
+	ChannelMax uint16
+	FrameMax   uint32
+	Heartbeat  uint16
+}
+
+func (msg *connectionTuneOk) id() (uint16, uint16) {
+	return 10, 31
+}
+
+func (msg *connectionTuneOk) wait() bool {
+	return true
+}
+
+func (msg *connectionTuneOk) write(w io.Writer) (err error) {
+
+	if err = binary.Write(w, binary.BigEndian, msg.ChannelMax); err != nil {
+		return
+	}
+
+	if err = binary.Write(w, binary.BigEndian, msg.FrameMax); err != nil {
+		return
+	}
+
+	if err = binary.Write(w, binary.BigEndian, msg.Heartbeat); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *connectionTuneOk) read(r io.Reader) (err error) {
+
+	if err = binary.Read(r, binary.BigEndian, &msg.ChannelMax); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &msg.FrameMax); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &msg.Heartbeat); err != nil {
+		return
+	}
+
+	return
+}
+
+type connectionOpen struct {
+	VirtualHost string
+	reserved1   string
+	reserved2   bool
+}
+
+func (msg *connectionOpen) id() (uint16, uint16) {
+	return 10, 40
+}
+
+func (msg *connectionOpen) wait() bool {
+	return true
+}
+
+func (msg *connectionOpen) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = writeShortstr(w, msg.VirtualHost); err != nil {
+		return
+	}
+	if err = writeShortstr(w, msg.reserved1); err != nil {
+		return
+	}
+
+	if msg.reserved2 {
+		bits |= 1 << 0
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *connectionOpen) read(r io.Reader) (err error) {
+	var bits byte
+
+	if msg.VirtualHost, err = readShortstr(r); err != nil {
+		return
+	}
+	if msg.reserved1, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.reserved2 = (bits&(1<<0) > 0)
+
+	return
+}
+
+type connectionOpenOk struct {
+	reserved1 string
+}
+
+func (msg *connectionOpenOk) id() (uint16, uint16) {
+	return 10, 41
+}
+
+func (msg *connectionOpenOk) wait() bool {
+	return true
+}
+
+func (msg *connectionOpenOk) write(w io.Writer) (err error) {
+
+	if err = writeShortstr(w, msg.reserved1); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *connectionOpenOk) read(r io.Reader) (err error) {
+
+	if msg.reserved1, err = readShortstr(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type connectionClose struct {
+	ReplyCode uint16
+	ReplyText string
+	ClassId   uint16
+	MethodId  uint16
+}
+
+func (msg *connectionClose) id() (uint16, uint16) {
+	return 10, 50
+}
+
+func (msg *connectionClose) wait() bool {
+	return true
+}
+
+func (msg *connectionClose) write(w io.Writer) (err error) {
+
+	if err = binary.Write(w, binary.BigEndian, msg.ReplyCode); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.ReplyText); err != nil {
+		return
+	}
+
+	if err = binary.Write(w, binary.BigEndian, msg.ClassId); err != nil {
+		return
+	}
+	if err = binary.Write(w, binary.BigEndian, msg.MethodId); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *connectionClose) read(r io.Reader) (err error) {
+
+	if err = binary.Read(r, binary.BigEndian, &msg.ReplyCode); err != nil {
+		return
+	}
+
+	if msg.ReplyText, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &msg.ClassId); err != nil {
+		return
+	}
+	if err = binary.Read(r, binary.BigEndian, &msg.MethodId); err != nil {
+		return
+	}
+
+	return
+}
+
+type connectionCloseOk struct {
+}
+
+func (msg *connectionCloseOk) id() (uint16, uint16) {
+	return 10, 51
+}
+
+func (msg *connectionCloseOk) wait() bool {
+	return true
+}
+
+func (msg *connectionCloseOk) write(w io.Writer) (err error) {
+
+	return
+}
+
+func (msg *connectionCloseOk) read(r io.Reader) (err error) {
+
+	return
+}
+
+type connectionBlocked struct {
+	Reason string
+}
+
+func (msg *connectionBlocked) id() (uint16, uint16) {
+	return 10, 60
+}
+
+func (msg *connectionBlocked) wait() bool {
+	return false
+}
+
+func (msg *connectionBlocked) write(w io.Writer) (err error) {
+
+	if err = writeShortstr(w, msg.Reason); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *connectionBlocked) read(r io.Reader) (err error) {
+
+	if msg.Reason, err = readShortstr(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type connectionUnblocked struct {
+}
+
+func (msg *connectionUnblocked) id() (uint16, uint16) {
+	return 10, 61
+}
+
+func (msg *connectionUnblocked) wait() bool {
+	return false
+}
+
+func (msg *connectionUnblocked) write(w io.Writer) (err error) {
+
+	return
+}
+
+func (msg *connectionUnblocked) read(r io.Reader) (err error) {
+
+	return
+}
+
+type channelOpen struct {
+	reserved1 string
+}
+
+func (msg *channelOpen) id() (uint16, uint16) {
+	return 20, 10
+}
+
+func (msg *channelOpen) wait() bool {
+	return true
+}
+
+func (msg *channelOpen) write(w io.Writer) (err error) {
+
+	if err = writeShortstr(w, msg.reserved1); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *channelOpen) read(r io.Reader) (err error) {
+
+	if msg.reserved1, err = readShortstr(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type channelOpenOk struct {
+	reserved1 string
+}
+
+func (msg *channelOpenOk) id() (uint16, uint16) {
+	return 20, 11
+}
+
+func (msg *channelOpenOk) wait() bool {
+	return true
+}
+
+func (msg *channelOpenOk) write(w io.Writer) (err error) {
+
+	if err = writeLongstr(w, msg.reserved1); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *channelOpenOk) read(r io.Reader) (err error) {
+
+	if msg.reserved1, err = readLongstr(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type channelFlow struct {
+	Active bool
+}
+
+func (msg *channelFlow) id() (uint16, uint16) {
+	return 20, 20
+}
+
+func (msg *channelFlow) wait() bool {
+	return true
+}
+
+func (msg *channelFlow) write(w io.Writer) (err error) {
+	var bits byte
+
+	if msg.Active {
+		bits |= 1 << 0
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *channelFlow) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.Active = (bits&(1<<0) > 0)
+
+	return
+}
+
+type channelFlowOk struct {
+	Active bool
+}
+
+func (msg *channelFlowOk) id() (uint16, uint16) {
+	return 20, 21
+}
+
+func (msg *channelFlowOk) wait() bool {
+	return false
+}
+
+func (msg *channelFlowOk) write(w io.Writer) (err error) {
+	var bits byte
+
+	if msg.Active {
+		bits |= 1 << 0
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *channelFlowOk) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.Active = (bits&(1<<0) > 0)
+
+	return
+}
+
+type channelClose struct {
+	ReplyCode uint16
+	ReplyText string
+	ClassId   uint16
+	MethodId  uint16
+}
+
+func (msg *channelClose) id() (uint16, uint16) {
+	return 20, 40
+}
+
+func (msg *channelClose) wait() bool {
+	return true
+}
+
+func (msg *channelClose) write(w io.Writer) (err error) {
+
+	if err = binary.Write(w, binary.BigEndian, msg.ReplyCode); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.ReplyText); err != nil {
+		return
+	}
+
+	if err = binary.Write(w, binary.BigEndian, msg.ClassId); err != nil {
+		return
+	}
+	if err = binary.Write(w, binary.BigEndian, msg.MethodId); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *channelClose) read(r io.Reader) (err error) {
+
+	if err = binary.Read(r, binary.BigEndian, &msg.ReplyCode); err != nil {
+		return
+	}
+
+	if msg.ReplyText, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &msg.ClassId); err != nil {
+		return
+	}
+	if err = binary.Read(r, binary.BigEndian, &msg.MethodId); err != nil {
+		return
+	}
+
+	return
+}
+
+type channelCloseOk struct {
+}
+
+func (msg *channelCloseOk) id() (uint16, uint16) {
+	return 20, 41
+}
+
+func (msg *channelCloseOk) wait() bool {
+	return true
+}
+
+func (msg *channelCloseOk) write(w io.Writer) (err error) {
+
+	return
+}
+
+func (msg *channelCloseOk) read(r io.Reader) (err error) {
+
+	return
+}
+
+type exchangeDeclare struct {
+	reserved1  uint16
+	Exchange   string
+	Type       string
+	Passive    bool
+	Durable    bool
+	AutoDelete bool
+	Internal   bool
+	NoWait     bool
+	Arguments  Table
+}
+
+func (msg *exchangeDeclare) id() (uint16, uint16) {
+	return 40, 10
+}
+
+func (msg *exchangeDeclare) wait() bool {
+	return true && !msg.NoWait
+}
+
+func (msg *exchangeDeclare) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = binary.Write(w, binary.BigEndian, msg.reserved1); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.Exchange); err != nil {
+		return
+	}
+	if err = writeShortstr(w, msg.Type); err != nil {
+		return
+	}
+
+	if msg.Passive {
+		bits |= 1 << 0
+	}
+
+	if msg.Durable {
+		bits |= 1 << 1
+	}
+
+	if msg.AutoDelete {
+		bits |= 1 << 2
+	}
+
+	if msg.Internal {
+		bits |= 1 << 3
+	}
+
+	if msg.NoWait {
+		bits |= 1 << 4
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	if err = writeTable(w, msg.Arguments); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *exchangeDeclare) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &msg.reserved1); err != nil {
+		return
+	}
+
+	if msg.Exchange, err = readShortstr(r); err != nil {
+		return
+	}
+	if msg.Type, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.Passive = (bits&(1<<0) > 0)
+	msg.Durable = (bits&(1<<1) > 0)
+	msg.AutoDelete = (bits&(1<<2) > 0)
+	msg.Internal = (bits&(1<<3) > 0)
+	msg.NoWait = (bits&(1<<4) > 0)
+
+	if msg.Arguments, err = readTable(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type exchangeDeclareOk struct {
+}
+
+func (msg *exchangeDeclareOk) id() (uint16, uint16) {
+	return 40, 11
+}
+
+func (msg *exchangeDeclareOk) wait() bool {
+	return true
+}
+
+func (msg *exchangeDeclareOk) write(w io.Writer) (err error) {
+
+	return
+}
+
+func (msg *exchangeDeclareOk) read(r io.Reader) (err error) {
+
+	return
+}
+
+type exchangeDelete struct {
+	reserved1 uint16
+	Exchange  string
+	IfUnused  bool
+	NoWait    bool
+}
+
+func (msg *exchangeDelete) id() (uint16, uint16) {
+	return 40, 20
+}
+
+func (msg *exchangeDelete) wait() bool {
+	return true && !msg.NoWait
+}
+
+func (msg *exchangeDelete) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = binary.Write(w, binary.BigEndian, msg.reserved1); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.Exchange); err != nil {
+		return
+	}
+
+	if msg.IfUnused {
+		bits |= 1 << 0
+	}
+
+	if msg.NoWait {
+		bits |= 1 << 1
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *exchangeDelete) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &msg.reserved1); err != nil {
+		return
+	}
+
+	if msg.Exchange, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.IfUnused = (bits&(1<<0) > 0)
+	msg.NoWait = (bits&(1<<1) > 0)
+
+	return
+}
+
+type exchangeDeleteOk struct {
+}
+
+func (msg *exchangeDeleteOk) id() (uint16, uint16) {
+	return 40, 21
+}
+
+func (msg *exchangeDeleteOk) wait() bool {
+	return true
+}
+
+func (msg *exchangeDeleteOk) write(w io.Writer) (err error) {
+
+	return
+}
+
+func (msg *exchangeDeleteOk) read(r io.Reader) (err error) {
+
+	return
+}
+
+type exchangeBind struct {
+	reserved1   uint16
+	Destination string
+	Source      string
+	RoutingKey  string
+	NoWait      bool
+	Arguments   Table
+}
+
+func (msg *exchangeBind) id() (uint16, uint16) {
+	return 40, 30
+}
+
+func (msg *exchangeBind) wait() bool {
+	return true && !msg.NoWait
+}
+
+func (msg *exchangeBind) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = binary.Write(w, binary.BigEndian, msg.reserved1); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.Destination); err != nil {
+		return
+	}
+	if err = writeShortstr(w, msg.Source); err != nil {
+		return
+	}
+	if err = writeShortstr(w, msg.RoutingKey); err != nil {
+		return
+	}
+
+	if msg.NoWait {
+		bits |= 1 << 0
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	if err = writeTable(w, msg.Arguments); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *exchangeBind) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &msg.reserved1); err != nil {
+		return
+	}
+
+	if msg.Destination, err = readShortstr(r); err != nil {
+		return
+	}
+	if msg.Source, err = readShortstr(r); err != nil {
+		return
+	}
+	if msg.RoutingKey, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.NoWait = (bits&(1<<0) > 0)
+
+	if msg.Arguments, err = readTable(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type exchangeBindOk struct {
+}
+
+func (msg *exchangeBindOk) id() (uint16, uint16) {
+	return 40, 31
+}
+
+func (msg *exchangeBindOk) wait() bool {
+	return true
+}
+
+func (msg *exchangeBindOk) write(w io.Writer) (err error) {
+
+	return
+}
+
+func (msg *exchangeBindOk) read(r io.Reader) (err error) {
+
+	return
+}
+
+type exchangeUnbind struct {
+	reserved1   uint16
+	Destination string
+	Source      string
+	RoutingKey  string
+	NoWait      bool
+	Arguments   Table
+}
+
+func (msg *exchangeUnbind) id() (uint16, uint16) {
+	return 40, 40
+}
+
+func (msg *exchangeUnbind) wait() bool {
+	return true && !msg.NoWait
+}
+
+func (msg *exchangeUnbind) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = binary.Write(w, binary.BigEndian, msg.reserved1); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.Destination); err != nil {
+		return
+	}
+	if err = writeShortstr(w, msg.Source); err != nil {
+		return
+	}
+	if err = writeShortstr(w, msg.RoutingKey); err != nil {
+		return
+	}
+
+	if msg.NoWait {
+		bits |= 1 << 0
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	if err = writeTable(w, msg.Arguments); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *exchangeUnbind) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &msg.reserved1); err != nil {
+		return
+	}
+
+	if msg.Destination, err = readShortstr(r); err != nil {
+		return
+	}
+	if msg.Source, err = readShortstr(r); err != nil {
+		return
+	}
+	if msg.RoutingKey, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.NoWait = (bits&(1<<0) > 0)
+
+	if msg.Arguments, err = readTable(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type exchangeUnbindOk struct {
+}
+
+func (msg *exchangeUnbindOk) id() (uint16, uint16) {
+	return 40, 51
+}
+
+func (msg *exchangeUnbindOk) wait() bool {
+	return true
+}
+
+func (msg *exchangeUnbindOk) write(w io.Writer) (err error) {
+
+	return
+}
+
+func (msg *exchangeUnbindOk) read(r io.Reader) (err error) {
+
+	return
+}
+
+type queueDeclare struct {
+	reserved1  uint16
+	Queue      string
+	Passive    bool
+	Durable    bool
+	Exclusive  bool
+	AutoDelete bool
+	NoWait     bool
+	Arguments  Table
+}
+
+func (msg *queueDeclare) id() (uint16, uint16) {
+	return 50, 10
+}
+
+func (msg *queueDeclare) wait() bool {
+	return true && !msg.NoWait
+}
+
+func (msg *queueDeclare) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = binary.Write(w, binary.BigEndian, msg.reserved1); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.Queue); err != nil {
+		return
+	}
+
+	if msg.Passive {
+		bits |= 1 << 0
+	}
+
+	if msg.Durable {
+		bits |= 1 << 1
+	}
+
+	if msg.Exclusive {
+		bits |= 1 << 2
+	}
+
+	if msg.AutoDelete {
+		bits |= 1 << 3
+	}
+
+	if msg.NoWait {
+		bits |= 1 << 4
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	if err = writeTable(w, msg.Arguments); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *queueDeclare) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &msg.reserved1); err != nil {
+		return
+	}
+
+	if msg.Queue, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.Passive = (bits&(1<<0) > 0)
+	msg.Durable = (bits&(1<<1) > 0)
+	msg.Exclusive = (bits&(1<<2) > 0)
+	msg.AutoDelete = (bits&(1<<3) > 0)
+	msg.NoWait = (bits&(1<<4) > 0)
+
+	if msg.Arguments, err = readTable(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type queueDeclareOk struct {
+	Queue         string
+	MessageCount  uint32
+	ConsumerCount uint32
+}
+
+func (msg *queueDeclareOk) id() (uint16, uint16) {
+	return 50, 11
+}
+
+func (msg *queueDeclareOk) wait() bool {
+	return true
+}
+
+func (msg *queueDeclareOk) write(w io.Writer) (err error) {
+
+	if err = writeShortstr(w, msg.Queue); err != nil {
+		return
+	}
+
+	if err = binary.Write(w, binary.BigEndian, msg.MessageCount); err != nil {
+		return
+	}
+	if err = binary.Write(w, binary.BigEndian, msg.ConsumerCount); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *queueDeclareOk) read(r io.Reader) (err error) {
+
+	if msg.Queue, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &msg.MessageCount); err != nil {
+		return
+	}
+	if err = binary.Read(r, binary.BigEndian, &msg.ConsumerCount); err != nil {
+		return
+	}
+
+	return
+}
+
+type queueBind struct {
+	reserved1  uint16
+	Queue      string
+	Exchange   string
+	RoutingKey string
+	NoWait     bool
+	Arguments  Table
+}
+
+func (msg *queueBind) id() (uint16, uint16) {
+	return 50, 20
+}
+
+func (msg *queueBind) wait() bool {
+	return true && !msg.NoWait
+}
+
+func (msg *queueBind) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = binary.Write(w, binary.BigEndian, msg.reserved1); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.Queue); err != nil {
+		return
+	}
+	if err = writeShortstr(w, msg.Exchange); err != nil {
+		return
+	}
+	if err = writeShortstr(w, msg.RoutingKey); err != nil {
+		return
+	}
+
+	if msg.NoWait {
+		bits |= 1 << 0
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	if err = writeTable(w, msg.Arguments); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *queueBind) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &msg.reserved1); err != nil {
+		return
+	}
+
+	if msg.Queue, err = readShortstr(r); err != nil {
+		return
+	}
+	if msg.Exchange, err = readShortstr(r); err != nil {
+		return
+	}
+	if msg.RoutingKey, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.NoWait = (bits&(1<<0) > 0)
+
+	if msg.Arguments, err = readTable(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type queueBindOk struct {
+}
+
+func (msg *queueBindOk) id() (uint16, uint16) {
+	return 50, 21
+}
+
+func (msg *queueBindOk) wait() bool {
+	return true
+}
+
+func (msg *queueBindOk) write(w io.Writer) (err error) {
+
+	return
+}
+
+func (msg *queueBindOk) read(r io.Reader) (err error) {
+
+	return
+}
+
+type queueUnbind struct {
+	reserved1  uint16
+	Queue      string
+	Exchange   string
+	RoutingKey string
+	Arguments  Table
+}
+
+func (msg *queueUnbind) id() (uint16, uint16) {
+	return 50, 50
+}
+
+func (msg *queueUnbind) wait() bool {
+	return true
+}
+
+func (msg *queueUnbind) write(w io.Writer) (err error) {
+
+	if err = binary.Write(w, binary.BigEndian, msg.reserved1); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.Queue); err != nil {
+		return
+	}
+	if err = writeShortstr(w, msg.Exchange); err != nil {
+		return
+	}
+	if err = writeShortstr(w, msg.RoutingKey); err != nil {
+		return
+	}
+
+	if err = writeTable(w, msg.Arguments); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *queueUnbind) read(r io.Reader) (err error) {
+
+	if err = binary.Read(r, binary.BigEndian, &msg.reserved1); err != nil {
+		return
+	}
+
+	if msg.Queue, err = readShortstr(r); err != nil {
+		return
+	}
+	if msg.Exchange, err = readShortstr(r); err != nil {
+		return
+	}
+	if msg.RoutingKey, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if msg.Arguments, err = readTable(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type queueUnbindOk struct {
+}
+
+func (msg *queueUnbindOk) id() (uint16, uint16) {
+	return 50, 51
+}
+
+func (msg *queueUnbindOk) wait() bool {
+	return true
+}
+
+func (msg *queueUnbindOk) write(w io.Writer) (err error) {
+
+	return
+}
+
+func (msg *queueUnbindOk) read(r io.Reader) (err error) {
+
+	return
+}
+
+type queuePurge struct {
+	reserved1 uint16
+	Queue     string
+	NoWait    bool
+}
+
+func (msg *queuePurge) id() (uint16, uint16) {
+	return 50, 30
+}
+
+func (msg *queuePurge) wait() bool {
+	return true && !msg.NoWait
+}
+
+func (msg *queuePurge) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = binary.Write(w, binary.BigEndian, msg.reserved1); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.Queue); err != nil {
+		return
+	}
+
+	if msg.NoWait {
+		bits |= 1 << 0
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *queuePurge) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &msg.reserved1); err != nil {
+		return
+	}
+
+	if msg.Queue, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.NoWait = (bits&(1<<0) > 0)
+
+	return
+}
+
+type queuePurgeOk struct {
+	MessageCount uint32
+}
+
+func (msg *queuePurgeOk) id() (uint16, uint16) {
+	return 50, 31
+}
+
+func (msg *queuePurgeOk) wait() bool {
+	return true
+}
+
+func (msg *queuePurgeOk) write(w io.Writer) (err error) {
+
+	if err = binary.Write(w, binary.BigEndian, msg.MessageCount); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *queuePurgeOk) read(r io.Reader) (err error) {
+
+	if err = binary.Read(r, binary.BigEndian, &msg.MessageCount); err != nil {
+		return
+	}
+
+	return
+}
+
+type queueDelete struct {
+	reserved1 uint16
+	Queue     string
+	IfUnused  bool
+	IfEmpty   bool
+	NoWait    bool
+}
+
+func (msg *queueDelete) id() (uint16, uint16) {
+	return 50, 40
+}
+
+func (msg *queueDelete) wait() bool {
+	return true && !msg.NoWait
+}
+
+func (msg *queueDelete) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = binary.Write(w, binary.BigEndian, msg.reserved1); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.Queue); err != nil {
+		return
+	}
+
+	if msg.IfUnused {
+		bits |= 1 << 0
+	}
+
+	if msg.IfEmpty {
+		bits |= 1 << 1
+	}
+
+	if msg.NoWait {
+		bits |= 1 << 2
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *queueDelete) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &msg.reserved1); err != nil {
+		return
+	}
+
+	if msg.Queue, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.IfUnused = (bits&(1<<0) > 0)
+	msg.IfEmpty = (bits&(1<<1) > 0)
+	msg.NoWait = (bits&(1<<2) > 0)
+
+	return
+}
+
+type queueDeleteOk struct {
+	MessageCount uint32
+}
+
+func (msg *queueDeleteOk) id() (uint16, uint16) {
+	return 50, 41
+}
+
+func (msg *queueDeleteOk) wait() bool {
+	return true
+}
+
+func (msg *queueDeleteOk) write(w io.Writer) (err error) {
+
+	if err = binary.Write(w, binary.BigEndian, msg.MessageCount); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *queueDeleteOk) read(r io.Reader) (err error) {
+
+	if err = binary.Read(r, binary.BigEndian, &msg.MessageCount); err != nil {
+		return
+	}
+
+	return
+}
+
+type basicQos struct {
+	PrefetchSize  uint32
+	PrefetchCount uint16
+	Global        bool
+}
+
+func (msg *basicQos) id() (uint16, uint16) {
+	return 60, 10
+}
+
+func (msg *basicQos) wait() bool {
+	return true
+}
+
+func (msg *basicQos) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = binary.Write(w, binary.BigEndian, msg.PrefetchSize); err != nil {
+		return
+	}
+
+	if err = binary.Write(w, binary.BigEndian, msg.PrefetchCount); err != nil {
+		return
+	}
+
+	if msg.Global {
+		bits |= 1 << 0
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *basicQos) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &msg.PrefetchSize); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &msg.PrefetchCount); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.Global = (bits&(1<<0) > 0)
+
+	return
+}
+
+type basicQosOk struct {
+}
+
+func (msg *basicQosOk) id() (uint16, uint16) {
+	return 60, 11
+}
+
+func (msg *basicQosOk) wait() bool {
+	return true
+}
+
+func (msg *basicQosOk) write(w io.Writer) (err error) {
+
+	return
+}
+
+func (msg *basicQosOk) read(r io.Reader) (err error) {
+
+	return
+}
+
+type basicConsume struct {
+	reserved1   uint16
+	Queue       string
+	ConsumerTag string
+	NoLocal     bool
+	NoAck       bool
+	Exclusive   bool
+	NoWait      bool
+	Arguments   Table
+}
+
+func (msg *basicConsume) id() (uint16, uint16) {
+	return 60, 20
+}
+
+func (msg *basicConsume) wait() bool {
+	return true && !msg.NoWait
+}
+
+func (msg *basicConsume) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = binary.Write(w, binary.BigEndian, msg.reserved1); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.Queue); err != nil {
+		return
+	}
+	if err = writeShortstr(w, msg.ConsumerTag); err != nil {
+		return
+	}
+
+	if msg.NoLocal {
+		bits |= 1 << 0
+	}
+
+	if msg.NoAck {
+		bits |= 1 << 1
+	}
+
+	if msg.Exclusive {
+		bits |= 1 << 2
+	}
+
+	if msg.NoWait {
+		bits |= 1 << 3
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	if err = writeTable(w, msg.Arguments); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *basicConsume) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &msg.reserved1); err != nil {
+		return
+	}
+
+	if msg.Queue, err = readShortstr(r); err != nil {
+		return
+	}
+	if msg.ConsumerTag, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.NoLocal = (bits&(1<<0) > 0)
+	msg.NoAck = (bits&(1<<1) > 0)
+	msg.Exclusive = (bits&(1<<2) > 0)
+	msg.NoWait = (bits&(1<<3) > 0)
+
+	if msg.Arguments, err = readTable(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type basicConsumeOk struct {
+	ConsumerTag string
+}
+
+func (msg *basicConsumeOk) id() (uint16, uint16) {
+	return 60, 21
+}
+
+func (msg *basicConsumeOk) wait() bool {
+	return true
+}
+
+func (msg *basicConsumeOk) write(w io.Writer) (err error) {
+
+	if err = writeShortstr(w, msg.ConsumerTag); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *basicConsumeOk) read(r io.Reader) (err error) {
+
+	if msg.ConsumerTag, err = readShortstr(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type basicCancel struct {
+	ConsumerTag string
+	NoWait      bool
+}
+
+func (msg *basicCancel) id() (uint16, uint16) {
+	return 60, 30
+}
+
+func (msg *basicCancel) wait() bool {
+	return true && !msg.NoWait
+}
+
+func (msg *basicCancel) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = writeShortstr(w, msg.ConsumerTag); err != nil {
+		return
+	}
+
+	if msg.NoWait {
+		bits |= 1 << 0
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *basicCancel) read(r io.Reader) (err error) {
+	var bits byte
+
+	if msg.ConsumerTag, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.NoWait = (bits&(1<<0) > 0)
+
+	return
+}
+
+type basicCancelOk struct {
+	ConsumerTag string
+}
+
+func (msg *basicCancelOk) id() (uint16, uint16) {
+	return 60, 31
+}
+
+func (msg *basicCancelOk) wait() bool {
+	return true
+}
+
+func (msg *basicCancelOk) write(w io.Writer) (err error) {
+
+	if err = writeShortstr(w, msg.ConsumerTag); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *basicCancelOk) read(r io.Reader) (err error) {
+
+	if msg.ConsumerTag, err = readShortstr(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type basicPublish struct {
+	reserved1  uint16
+	Exchange   string
+	RoutingKey string
+	Mandatory  bool
+	Immediate  bool
+	Properties properties
+	Body       []byte
+}
+
+func (msg *basicPublish) id() (uint16, uint16) {
+	return 60, 40
+}
+
+func (msg *basicPublish) wait() bool {
+	return false
+}
+
+func (msg *basicPublish) getContent() (properties, []byte) {
+	return msg.Properties, msg.Body
+}
+
+func (msg *basicPublish) setContent(props properties, body []byte) {
+	msg.Properties, msg.Body = props, body
+}
+
+func (msg *basicPublish) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = binary.Write(w, binary.BigEndian, msg.reserved1); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.Exchange); err != nil {
+		return
+	}
+	if err = writeShortstr(w, msg.RoutingKey); err != nil {
+		return
+	}
+
+	if msg.Mandatory {
+		bits |= 1 << 0
+	}
+
+	if msg.Immediate {
+		bits |= 1 << 1
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *basicPublish) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &msg.reserved1); err != nil {
+		return
+	}
+
+	if msg.Exchange, err = readShortstr(r); err != nil {
+		return
+	}
+	if msg.RoutingKey, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.Mandatory = (bits&(1<<0) > 0)
+	msg.Immediate = (bits&(1<<1) > 0)
+
+	return
+}
+
+type basicReturn struct {
+	ReplyCode  uint16
+	ReplyText  string
+	Exchange   string
+	RoutingKey string
+	Properties properties
+	Body       []byte
+}
+
+func (msg *basicReturn) id() (uint16, uint16) {
+	return 60, 50
+}
+
+func (msg *basicReturn) wait() bool {
+	return false
+}
+
+func (msg *basicReturn) getContent() (properties, []byte) {
+	return msg.Properties, msg.Body
+}
+
+func (msg *basicReturn) setContent(props properties, body []byte) {
+	msg.Properties, msg.Body = props, body
+}
+
+func (msg *basicReturn) write(w io.Writer) (err error) {
+
+	if err = binary.Write(w, binary.BigEndian, msg.ReplyCode); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.ReplyText); err != nil {
+		return
+	}
+	if err = writeShortstr(w, msg.Exchange); err != nil {
+		return
+	}
+	if err = writeShortstr(w, msg.RoutingKey); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *basicReturn) read(r io.Reader) (err error) {
+
+	if err = binary.Read(r, binary.BigEndian, &msg.ReplyCode); err != nil {
+		return
+	}
+
+	if msg.ReplyText, err = readShortstr(r); err != nil {
+		return
+	}
+	if msg.Exchange, err = readShortstr(r); err != nil {
+		return
+	}
+	if msg.RoutingKey, err = readShortstr(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type basicDeliver struct {
+	ConsumerTag string
+	DeliveryTag uint64
+	Redelivered bool
+	Exchange    string
+	RoutingKey  string
+	Properties  properties
+	Body        []byte
+}
+
+func (msg *basicDeliver) id() (uint16, uint16) {
+	return 60, 60
+}
+
+func (msg *basicDeliver) wait() bool {
+	return false
+}
+
+func (msg *basicDeliver) getContent() (properties, []byte) {
+	return msg.Properties, msg.Body
+}
+
+func (msg *basicDeliver) setContent(props properties, body []byte) {
+	msg.Properties, msg.Body = props, body
+}
+
+func (msg *basicDeliver) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = writeShortstr(w, msg.ConsumerTag); err != nil {
+		return
+	}
+
+	if err = binary.Write(w, binary.BigEndian, msg.DeliveryTag); err != nil {
+		return
+	}
+
+	if msg.Redelivered {
+		bits |= 1 << 0
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.Exchange); err != nil {
+		return
+	}
+	if err = writeShortstr(w, msg.RoutingKey); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *basicDeliver) read(r io.Reader) (err error) {
+	var bits byte
+
+	if msg.ConsumerTag, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &msg.DeliveryTag); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.Redelivered = (bits&(1<<0) > 0)
+
+	if msg.Exchange, err = readShortstr(r); err != nil {
+		return
+	}
+	if msg.RoutingKey, err = readShortstr(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type basicGet struct {
+	reserved1 uint16
+	Queue     string
+	NoAck     bool
+}
+
+func (msg *basicGet) id() (uint16, uint16) {
+	return 60, 70
+}
+
+func (msg *basicGet) wait() bool {
+	return true
+}
+
+func (msg *basicGet) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = binary.Write(w, binary.BigEndian, msg.reserved1); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.Queue); err != nil {
+		return
+	}
+
+	if msg.NoAck {
+		bits |= 1 << 0
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *basicGet) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &msg.reserved1); err != nil {
+		return
+	}
+
+	if msg.Queue, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.NoAck = (bits&(1<<0) > 0)
+
+	return
+}
+
+type basicGetOk struct {
+	DeliveryTag  uint64
+	Redelivered  bool
+	Exchange     string
+	RoutingKey   string
+	MessageCount uint32
+	Properties   properties
+	Body         []byte
+}
+
+func (msg *basicGetOk) id() (uint16, uint16) {
+	return 60, 71
+}
+
+func (msg *basicGetOk) wait() bool {
+	return true
+}
+
+func (msg *basicGetOk) getContent() (properties, []byte) {
+	return msg.Properties, msg.Body
+}
+
+func (msg *basicGetOk) setContent(props properties, body []byte) {
+	msg.Properties, msg.Body = props, body
+}
+
+func (msg *basicGetOk) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = binary.Write(w, binary.BigEndian, msg.DeliveryTag); err != nil {
+		return
+	}
+
+	if msg.Redelivered {
+		bits |= 1 << 0
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	if err = writeShortstr(w, msg.Exchange); err != nil {
+		return
+	}
+	if err = writeShortstr(w, msg.RoutingKey); err != nil {
+		return
+	}
+
+	if err = binary.Write(w, binary.BigEndian, msg.MessageCount); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *basicGetOk) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &msg.DeliveryTag); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.Redelivered = (bits&(1<<0) > 0)
+
+	if msg.Exchange, err = readShortstr(r); err != nil {
+		return
+	}
+	if msg.RoutingKey, err = readShortstr(r); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &msg.MessageCount); err != nil {
+		return
+	}
+
+	return
+}
+
+type basicGetEmpty struct {
+	reserved1 string
+}
+
+func (msg *basicGetEmpty) id() (uint16, uint16) {
+	return 60, 72
+}
+
+func (msg *basicGetEmpty) wait() bool {
+	return true
+}
+
+func (msg *basicGetEmpty) write(w io.Writer) (err error) {
+
+	if err = writeShortstr(w, msg.reserved1); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *basicGetEmpty) read(r io.Reader) (err error) {
+
+	if msg.reserved1, err = readShortstr(r); err != nil {
+		return
+	}
+
+	return
+}
+
+type basicAck struct {
+	DeliveryTag uint64
+	Multiple    bool
+}
+
+func (msg *basicAck) id() (uint16, uint16) {
+	return 60, 80
+}
+
+func (msg *basicAck) wait() bool {
+	return false
+}
+
+func (msg *basicAck) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = binary.Write(w, binary.BigEndian, msg.DeliveryTag); err != nil {
+		return
+	}
+
+	if msg.Multiple {
+		bits |= 1 << 0
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *basicAck) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &msg.DeliveryTag); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.Multiple = (bits&(1<<0) > 0)
+
+	return
+}
+
+type basicReject struct {
+	DeliveryTag uint64
+	Requeue     bool
+}
+
+func (msg *basicReject) id() (uint16, uint16) {
+	return 60, 90
+}
+
+func (msg *basicReject) wait() bool {
+	return false
+}
+
+func (msg *basicReject) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = binary.Write(w, binary.BigEndian, msg.DeliveryTag); err != nil {
+		return
+	}
+
+	if msg.Requeue {
+		bits |= 1 << 0
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *basicReject) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &msg.DeliveryTag); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.Requeue = (bits&(1<<0) > 0)
+
+	return
+}
+
+type basicRecoverAsync struct {
+	Requeue bool
+}
+
+func (msg *basicRecoverAsync) id() (uint16, uint16) {
+	return 60, 100
+}
+
+func (msg *basicRecoverAsync) wait() bool {
+	return false
+}
+
+func (msg *basicRecoverAsync) write(w io.Writer) (err error) {
+	var bits byte
+
+	if msg.Requeue {
+		bits |= 1 << 0
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *basicRecoverAsync) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.Requeue = (bits&(1<<0) > 0)
+
+	return
+}
+
+type basicRecover struct {
+	Requeue bool
+}
+
+func (msg *basicRecover) id() (uint16, uint16) {
+	return 60, 110
+}
+
+func (msg *basicRecover) wait() bool {
+	return true
+}
+
+func (msg *basicRecover) write(w io.Writer) (err error) {
+	var bits byte
+
+	if msg.Requeue {
+		bits |= 1 << 0
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *basicRecover) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.Requeue = (bits&(1<<0) > 0)
+
+	return
+}
+
+type basicRecoverOk struct {
+}
+
+func (msg *basicRecoverOk) id() (uint16, uint16) {
+	return 60, 111
+}
+
+func (msg *basicRecoverOk) wait() bool {
+	return true
+}
+
+func (msg *basicRecoverOk) write(w io.Writer) (err error) {
+
+	return
+}
+
+func (msg *basicRecoverOk) read(r io.Reader) (err error) {
+
+	return
+}
+
+type basicNack struct {
+	DeliveryTag uint64
+	Multiple    bool
+	Requeue     bool
+}
+
+func (msg *basicNack) id() (uint16, uint16) {
+	return 60, 120
+}
+
+func (msg *basicNack) wait() bool {
+	return false
+}
+
+func (msg *basicNack) write(w io.Writer) (err error) {
+	var bits byte
+
+	if err = binary.Write(w, binary.BigEndian, msg.DeliveryTag); err != nil {
+		return
+	}
+
+	if msg.Multiple {
+		bits |= 1 << 0
+	}
+
+	if msg.Requeue {
+		bits |= 1 << 1
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *basicNack) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &msg.DeliveryTag); err != nil {
+		return
+	}
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.Multiple = (bits&(1<<0) > 0)
+	msg.Requeue = (bits&(1<<1) > 0)
+
+	return
+}
+
+type txSelect struct {
+}
+
+func (msg *txSelect) id() (uint16, uint16) {
+	return 90, 10
+}
+
+func (msg *txSelect) wait() bool {
+	return true
+}
+
+func (msg *txSelect) write(w io.Writer) (err error) {
+
+	return
+}
+
+func (msg *txSelect) read(r io.Reader) (err error) {
+
+	return
+}
+
+type txSelectOk struct {
+}
+
+func (msg *txSelectOk) id() (uint16, uint16) {
+	return 90, 11
+}
+
+func (msg *txSelectOk) wait() bool {
+	return true
+}
+
+func (msg *txSelectOk) write(w io.Writer) (err error) {
+
+	return
+}
+
+func (msg *txSelectOk) read(r io.Reader) (err error) {
+
+	return
+}
+
+type txCommit struct {
+}
+
+func (msg *txCommit) id() (uint16, uint16) {
+	return 90, 20
+}
+
+func (msg *txCommit) wait() bool {
+	return true
+}
+
+func (msg *txCommit) write(w io.Writer) (err error) {
+
+	return
+}
+
+func (msg *txCommit) read(r io.Reader) (err error) {
+
+	return
+}
+
+type txCommitOk struct {
+}
+
+func (msg *txCommitOk) id() (uint16, uint16) {
+	return 90, 21
+}
+
+func (msg *txCommitOk) wait() bool {
+	return true
+}
+
+func (msg *txCommitOk) write(w io.Writer) (err error) {
+
+	return
+}
+
+func (msg *txCommitOk) read(r io.Reader) (err error) {
+
+	return
+}
+
+type txRollback struct {
+}
+
+func (msg *txRollback) id() (uint16, uint16) {
+	return 90, 30
+}
+
+func (msg *txRollback) wait() bool {
+	return true
+}
+
+func (msg *txRollback) write(w io.Writer) (err error) {
+
+	return
+}
+
+func (msg *txRollback) read(r io.Reader) (err error) {
+
+	return
+}
+
+type txRollbackOk struct {
+}
+
+func (msg *txRollbackOk) id() (uint16, uint16) {
+	return 90, 31
+}
+
+func (msg *txRollbackOk) wait() bool {
+	return true
+}
+
+func (msg *txRollbackOk) write(w io.Writer) (err error) {
+
+	return
+}
+
+func (msg *txRollbackOk) read(r io.Reader) (err error) {
+
+	return
+}
+
+type confirmSelect struct {
+	Nowait bool
+}
+
+func (msg *confirmSelect) id() (uint16, uint16) {
+	return 85, 10
+}
+
+func (msg *confirmSelect) wait() bool {
+	return true
+}
+
+func (msg *confirmSelect) write(w io.Writer) (err error) {
+	var bits byte
+
+	if msg.Nowait {
+		bits |= 1 << 0
+	}
+
+	if err = binary.Write(w, binary.BigEndian, bits); err != nil {
+		return
+	}
+
+	return
+}
+
+func (msg *confirmSelect) read(r io.Reader) (err error) {
+	var bits byte
+
+	if err = binary.Read(r, binary.BigEndian, &bits); err != nil {
+		return
+	}
+	msg.Nowait = (bits&(1<<0) > 0)
+
+	return
+}
+
+type confirmSelectOk struct {
+}
+
+func (msg *confirmSelectOk) id() (uint16, uint16) {
+	return 85, 11
+}
+
+func (msg *confirmSelectOk) wait() bool {
+	return true
+}
+
+func (msg *confirmSelectOk) write(w io.Writer) (err error) {
+
+	return
+}
+
+func (msg *confirmSelectOk) read(r io.Reader) (err error) {
+
+	return
+}
+
+func (r *reader) parseMethodFrame(channel uint16, size uint32) (f frame, err error) {
+	mf := &methodFrame{
+		ChannelId: channel,
+	}
+
+	if err = binary.Read(r.r, binary.BigEndian, &mf.ClassId); err != nil {
+		return
+	}
+
+	if err = binary.Read(r.r, binary.BigEndian, &mf.MethodId); err != nil {
+		return
+	}
+
+	switch mf.ClassId {
+
+	case 10: // connection
+		switch mf.MethodId {
+
+		case 10: // connection start
+			//fmt.Println("NextMethod: class:10 method:10")
+			method := &connectionStart{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 11: // connection start-ok
+			//fmt.Println("NextMethod: class:10 method:11")
+			method := &connectionStartOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 20: // connection secure
+			//fmt.Println("NextMethod: class:10 method:20")
+			method := &connectionSecure{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 21: // connection secure-ok
+			//fmt.Println("NextMethod: class:10 method:21")
+			method := &connectionSecureOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 30: // connection tune
+			//fmt.Println("NextMethod: class:10 method:30")
+			method := &connectionTune{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 31: // connection tune-ok
+			//fmt.Println("NextMethod: class:10 method:31")
+			method := &connectionTuneOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 40: // connection open
+			//fmt.Println("NextMethod: class:10 method:40")
+			method := &connectionOpen{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 41: // connection open-ok
+			//fmt.Println("NextMethod: class:10 method:41")
+			method := &connectionOpenOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 50: // connection close
+			//fmt.Println("NextMethod: class:10 method:50")
+			method := &connectionClose{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 51: // connection close-ok
+			//fmt.Println("NextMethod: class:10 method:51")
+			method := &connectionCloseOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 60: // connection blocked
+			//fmt.Println("NextMethod: class:10 method:60")
+			method := &connectionBlocked{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 61: // connection unblocked
+			//fmt.Println("NextMethod: class:10 method:61")
+			method := &connectionUnblocked{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		default:
+			return nil, fmt.Errorf("Bad method frame, unknown method %d for class %d", mf.MethodId, mf.ClassId)
+		}
+
+	case 20: // channel
+		switch mf.MethodId {
+
+		case 10: // channel open
+			//fmt.Println("NextMethod: class:20 method:10")
+			method := &channelOpen{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 11: // channel open-ok
+			//fmt.Println("NextMethod: class:20 method:11")
+			method := &channelOpenOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 20: // channel flow
+			//fmt.Println("NextMethod: class:20 method:20")
+			method := &channelFlow{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 21: // channel flow-ok
+			//fmt.Println("NextMethod: class:20 method:21")
+			method := &channelFlowOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 40: // channel close
+			//fmt.Println("NextMethod: class:20 method:40")
+			method := &channelClose{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 41: // channel close-ok
+			//fmt.Println("NextMethod: class:20 method:41")
+			method := &channelCloseOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		default:
+			return nil, fmt.Errorf("Bad method frame, unknown method %d for class %d", mf.MethodId, mf.ClassId)
+		}
+
+	case 40: // exchange
+		switch mf.MethodId {
+
+		case 10: // exchange declare
+			//fmt.Println("NextMethod: class:40 method:10")
+			method := &exchangeDeclare{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 11: // exchange declare-ok
+			//fmt.Println("NextMethod: class:40 method:11")
+			method := &exchangeDeclareOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 20: // exchange delete
+			//fmt.Println("NextMethod: class:40 method:20")
+			method := &exchangeDelete{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 21: // exchange delete-ok
+			//fmt.Println("NextMethod: class:40 method:21")
+			method := &exchangeDeleteOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 30: // exchange bind
+			//fmt.Println("NextMethod: class:40 method:30")
+			method := &exchangeBind{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 31: // exchange bind-ok
+			//fmt.Println("NextMethod: class:40 method:31")
+			method := &exchangeBindOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 40: // exchange unbind
+			//fmt.Println("NextMethod: class:40 method:40")
+			method := &exchangeUnbind{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 51: // exchange unbind-ok
+			//fmt.Println("NextMethod: class:40 method:51")
+			method := &exchangeUnbindOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		default:
+			return nil, fmt.Errorf("Bad method frame, unknown method %d for class %d", mf.MethodId, mf.ClassId)
+		}
+
+	case 50: // queue
+		switch mf.MethodId {
+
+		case 10: // queue declare
+			//fmt.Println("NextMethod: class:50 method:10")
+			method := &queueDeclare{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 11: // queue declare-ok
+			//fmt.Println("NextMethod: class:50 method:11")
+			method := &queueDeclareOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 20: // queue bind
+			//fmt.Println("NextMethod: class:50 method:20")
+			method := &queueBind{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 21: // queue bind-ok
+			//fmt.Println("NextMethod: class:50 method:21")
+			method := &queueBindOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 50: // queue unbind
+			//fmt.Println("NextMethod: class:50 method:50")
+			method := &queueUnbind{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 51: // queue unbind-ok
+			//fmt.Println("NextMethod: class:50 method:51")
+			method := &queueUnbindOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 30: // queue purge
+			//fmt.Println("NextMethod: class:50 method:30")
+			method := &queuePurge{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 31: // queue purge-ok
+			//fmt.Println("NextMethod: class:50 method:31")
+			method := &queuePurgeOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 40: // queue delete
+			//fmt.Println("NextMethod: class:50 method:40")
+			method := &queueDelete{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 41: // queue delete-ok
+			//fmt.Println("NextMethod: class:50 method:41")
+			method := &queueDeleteOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		default:
+			return nil, fmt.Errorf("Bad method frame, unknown method %d for class %d", mf.MethodId, mf.ClassId)
+		}
+
+	case 60: // basic
+		switch mf.MethodId {
+
+		case 10: // basic qos
+			//fmt.Println("NextMethod: class:60 method:10")
+			method := &basicQos{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 11: // basic qos-ok
+			//fmt.Println("NextMethod: class:60 method:11")
+			method := &basicQosOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 20: // basic consume
+			//fmt.Println("NextMethod: class:60 method:20")
+			method := &basicConsume{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 21: // basic consume-ok
+			//fmt.Println("NextMethod: class:60 method:21")
+			method := &basicConsumeOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 30: // basic cancel
+			//fmt.Println("NextMethod: class:60 method:30")
+			method := &basicCancel{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 31: // basic cancel-ok
+			//fmt.Println("NextMethod: class:60 method:31")
+			method := &basicCancelOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 40: // basic publish
+			//fmt.Println("NextMethod: class:60 method:40")
+			method := &basicPublish{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 50: // basic return
+			//fmt.Println("NextMethod: class:60 method:50")
+			method := &basicReturn{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 60: // basic deliver
+			//fmt.Println("NextMethod: class:60 method:60")
+			method := &basicDeliver{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 70: // basic get
+			//fmt.Println("NextMethod: class:60 method:70")
+			method := &basicGet{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 71: // basic get-ok
+			//fmt.Println("NextMethod: class:60 method:71")
+			method := &basicGetOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 72: // basic get-empty
+			//fmt.Println("NextMethod: class:60 method:72")
+			method := &basicGetEmpty{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 80: // basic ack
+			//fmt.Println("NextMethod: class:60 method:80")
+			method := &basicAck{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 90: // basic reject
+			//fmt.Println("NextMethod: class:60 method:90")
+			method := &basicReject{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 100: // basic recover-async
+			//fmt.Println("NextMethod: class:60 method:100")
+			method := &basicRecoverAsync{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 110: // basic recover
+			//fmt.Println("NextMethod: class:60 method:110")
+			method := &basicRecover{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 111: // basic recover-ok
+			//fmt.Println("NextMethod: class:60 method:111")
+			method := &basicRecoverOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 120: // basic nack
+			//fmt.Println("NextMethod: class:60 method:120")
+			method := &basicNack{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		default:
+			return nil, fmt.Errorf("Bad method frame, unknown method %d for class %d", mf.MethodId, mf.ClassId)
+		}
+
+	case 90: // tx
+		switch mf.MethodId {
+
+		case 10: // tx select
+			//fmt.Println("NextMethod: class:90 method:10")
+			method := &txSelect{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 11: // tx select-ok
+			//fmt.Println("NextMethod: class:90 method:11")
+			method := &txSelectOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 20: // tx commit
+			//fmt.Println("NextMethod: class:90 method:20")
+			method := &txCommit{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 21: // tx commit-ok
+			//fmt.Println("NextMethod: class:90 method:21")
+			method := &txCommitOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 30: // tx rollback
+			//fmt.Println("NextMethod: class:90 method:30")
+			method := &txRollback{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 31: // tx rollback-ok
+			//fmt.Println("NextMethod: class:90 method:31")
+			method := &txRollbackOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		default:
+			return nil, fmt.Errorf("Bad method frame, unknown method %d for class %d", mf.MethodId, mf.ClassId)
+		}
+
+	case 85: // confirm
+		switch mf.MethodId {
+
+		case 10: // confirm select
+			//fmt.Println("NextMethod: class:85 method:10")
+			method := &confirmSelect{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		case 11: // confirm select-ok
+			//fmt.Println("NextMethod: class:85 method:11")
+			method := &confirmSelectOk{}
+			if err = method.read(r.r); err != nil {
+				return
+			}
+			mf.Method = method
+
+		default:
+			return nil, fmt.Errorf("Bad method frame, unknown method %d for class %d", mf.MethodId, mf.ClassId)
+		}
+
+	default:
+		return nil, fmt.Errorf("Bad method frame, unknown class %d", mf.ClassId)
+	}
+
+	return mf, nil
+}
diff --git a/src/dma/vendor/github.com/streadway/amqp/types.go b/src/dma/vendor/github.com/streadway/amqp/types.go
new file mode 100644
index 00000000..ff5ea3cb
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/types.go
@@ -0,0 +1,427 @@
+// 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 http://github.com/streadway/amqp
+
+package amqp
+
+import (
+	"fmt"
+	"io"
+	"time"
+)
+
+// Constants for standard AMQP 0-9-1 exchange types.
+const (
+	ExchangeDirect  = "direct"
+	ExchangeFanout  = "fanout"
+	ExchangeTopic   = "topic"
+	ExchangeHeaders = "headers"
+)
+
+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"}
+)
+
+// Error captures 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 different parameters
+}
+
+func newError(code uint16, text string) *Error {
+	return &Error{
+		Code:    int(code),
+		Reason:  text,
+		Recover: isSoftExceptionCode(int(code)),
+		Server:  true,
+	}
+}
+
+func (e Error) Error() string {
+	return fmt.Sprintf("Exception (%d) Reason: %q", e.Code, e.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 implementation 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
+)
+
+// Queue captures the current server 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
+}
+
+// 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 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
+}
+
+// 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 Blocking struct {
+	Active bool   // TCP pushback active/inactive on server
+	Reason string // Server reason for activation
+}
+
+// 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 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
+}
+
+// Decimal matches the AMQP decimal type.  Scale is the number of decimal
+// digits Scale == 2, Value == 12345, Decimal == 123.45
+type Decimal struct {
+	Scale uint8
+	Value int32
+}
+
+// Table stores user supplied fields of the following types:
+//
+//   bool
+//   byte
+//   float32
+//   float64
+//   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.
+//
+type Table map[string]interface{}
+
+func validateField(f interface{}) error {
+	switch fv := f.(type) {
+	case nil, bool, byte, int16, int32, int64, float32, float64, string, []byte, Decimal, time.Time:
+		return nil
+
+	case []interface{}:
+		for _, v := range fv {
+			if err := validateField(v); err != nil {
+				return fmt.Errorf("in array %s", err)
+			}
+		}
+		return nil
+
+	case Table:
+		for k, v := range fv {
+			if err := validateField(v); err != nil {
+				return fmt.Errorf("table field %q %s", k, err)
+			}
+		}
+		return nil
+	}
+
+	return fmt.Errorf("value %t not supported", f)
+}
+
+// Validate returns and error if any Go types in the table are incompatible with AMQP types.
+func (t Table) Validate() error {
+	return validateField(t)
+}
+
+// Heap interface for maintaining delivery tags
+type tagSet []uint64
+
+func (set tagSet) Len() int              { return len(set) }
+func (set tagSet) Less(i, j int) bool    { return (set)[i] < (set)[j] }
+func (set tagSet) Swap(i, j int)         { (set)[i], (set)[j] = (set)[j], (set)[i] }
+func (set *tagSet) Push(tag interface{}) { *set = append(*set, tag.(uint64)) }
+func (set *tagSet) Pop() interface{} {
+	val := (*set)[len(*set)-1]
+	*set = (*set)[:len(*set)-1]
+	return val
+}
+
+type message interface {
+	id() (uint16, uint16)
+	wait() bool
+	read(io.Reader) error
+	write(io.Writer) error
+}
+
+type messageWithContent interface {
+	message
+	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 (f *methodFrame) channel() uint16 { return f.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 (f *heartbeatFrame) channel() uint16 { return f.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
+process
+*/
+type headerFrame struct {
+	ChannelId  uint16
+	ClassId    uint16
+	weight     uint16
+	Size       uint64
+	Properties properties
+}
+
+func (f *headerFrame) channel() uint16 { return f.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]
+		[method] [header] [body] [body]
+		[method]
+		...
+*/
+type bodyFrame struct {
+	ChannelId uint16
+	Body      []byte
+}
+
+func (f *bodyFrame) channel() uint16 { return f.ChannelId }
diff --git a/src/dma/vendor/github.com/streadway/amqp/uri.go b/src/dma/vendor/github.com/streadway/amqp/uri.go
new file mode 100644
index 00000000..35fefdc2
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/uri.go
@@ -0,0 +1,167 @@
+// 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 http://github.com/streadway/amqp
+
+package amqp
+
+import (
+	"errors"
+	"net"
+	"net/url"
+	"strconv"
+	"strings"
+)
+
+var errURIScheme = errors.New("AMQP scheme must be either 'amqp://' or 'amqps://'")
+var errURIWhitespace = errors.New("URI must not contain whitespace")
+
+var schemePorts = map[string]int{
+	"amqp":  5672,
+	"amqps": 5671,
+}
+
+var defaultURI = URI{
+	Scheme:   "amqp",
+	Host:     "localhost",
+	Port:     5672,
+	Username: "guest",
+	Password: "guest",
+	Vhost:    "/",
+}
+
+// URI represents a parsed AMQP URI string.
+type URI struct {
+	Scheme   string
+	Host     string
+	Port     int
+	Username string
+	Password string
+	Vhost    string
+}
+
+// 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 ParseURI(uri string) (URI, error) {
+	builder := defaultURI
+
+	if strings.Contains(uri, " ") == true {
+		return builder, errURIWhitespace
+	}
+
+	u, err := url.Parse(uri)
+	if err != nil {
+		return builder, err
+	}
+
+	defaultPort, okScheme := schemePorts[u.Scheme]
+
+	if okScheme {
+		builder.Scheme = u.Scheme
+	} else {
+		return builder, errURIScheme
+	}
+
+	host := u.Hostname()
+	port := u.Port()
+
+	if host != "" {
+		builder.Host = host
+	}
+
+	if port != "" {
+		port32, err := strconv.ParseInt(port, 10, 32)
+		if err != nil {
+			return builder, err
+		}
+		builder.Port = int(port32)
+	} else {
+		builder.Port = defaultPort
+	}
+
+	if u.User != nil {
+		builder.Username = u.User.Username()
+		if password, ok := u.User.Password(); ok {
+			builder.Password = password
+		}
+	}
+
+	if u.Path != "" {
+		if strings.HasPrefix(u.Path, "/") {
+			if u.Host == "" && strings.HasPrefix(u.Path, "///") {
+				// net/url doesn't handle local context authorities and leaves that up
+				// to the scheme handler.  In our case, we translate amqp:/// into the
+				// default host and whatever the vhost should be
+				if len(u.Path) > 3 {
+					builder.Vhost = u.Path[3:]
+				}
+			} else if len(u.Path) > 1 {
+				builder.Vhost = u.Path[1:]
+			}
+		} else {
+			builder.Vhost = u.Path
+		}
+	}
+
+	return builder, nil
+}
+
+// PlainAuth returns a PlainAuth structure based on the parsed URI's
+// Username and Password fields.
+func (uri URI) PlainAuth() *PlainAuth {
+	return &PlainAuth{
+		Username: uri.Username,
+		Password: uri.Password,
+	}
+}
+
+func (uri URI) String() string {
+	authority, err := url.Parse("")
+	if err != nil {
+		return err.Error()
+	}
+
+	authority.Scheme = uri.Scheme
+
+	if uri.Username != defaultURI.Username || uri.Password != defaultURI.Password {
+		authority.User = url.User(uri.Username)
+
+		if uri.Password != defaultURI.Password {
+			authority.User = url.UserPassword(uri.Username, uri.Password)
+		}
+	}
+
+	authority.Host = net.JoinHostPort(uri.Host, strconv.Itoa(uri.Port))
+
+	if defaultPort, found := schemePorts[uri.Scheme]; !found || defaultPort != uri.Port {
+		authority.Host = net.JoinHostPort(uri.Host, strconv.Itoa(uri.Port))
+	} else {
+		// JoinHostPort() automatically add brackets to the host if it's
+		// an IPv6 address.
+		//
+		// If not port is specified, JoinHostPort() return an IP address in the
+		// form of "[::1]:", so we use TrimSuffix() to remove the extra ":".
+		authority.Host = strings.TrimSuffix(net.JoinHostPort(uri.Host, ""), ":")
+	}
+
+	if uri.Vhost != defaultURI.Vhost {
+		// Make sure net/url does not double escape, e.g.
+		// "%2F" does not become "%252F".
+		authority.Path = uri.Vhost
+		authority.RawPath = url.QueryEscape(uri.Vhost)
+	} else {
+		authority.Path = "/"
+	}
+
+	return authority.String()
+}
diff --git a/src/dma/vendor/github.com/streadway/amqp/write.go b/src/dma/vendor/github.com/streadway/amqp/write.go
new file mode 100644
index 00000000..58ed20d6
--- /dev/null
+++ b/src/dma/vendor/github.com/streadway/amqp/write.go
@@ -0,0 +1,411 @@
+// 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 http://github.com/streadway/amqp
+
+package amqp
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"io"
+	"math"
+	"time"
+)
+
+func (w *writer) WriteFrame(frame frame) (err error) {
+	if err = frame.write(w.w); err != nil {
+		return
+	}
+
+	if buf, ok := w.w.(*bufio.Writer); ok {
+		err = buf.Flush()
+	}
+
+	return
+}
+
+func (f *methodFrame) write(w io.Writer) (err error) {
+	var payload bytes.Buffer
+
+	if f.Method == nil {
+		return errors.New("malformed frame: missing method")
+	}
+
+	class, method := f.Method.id()
+
+	if err = binary.Write(&payload, binary.BigEndian, class); err != nil {
+		return
+	}
+
+	if err = binary.Write(&payload, binary.BigEndian, method); err != nil {
+		return
+	}
+
+	if err = f.Method.write(&payload); err != nil {
+		return
+	}
+
+	return writeFrame(w, frameMethod, f.ChannelId, payload.Bytes())
+}
+
+// Heartbeat
+//
+// Payload is empty
+func (f *heartbeatFrame) write(w io.Writer) (err error) {
+	return writeFrame(w, frameHeartbeat, f.ChannelId, []byte{})
+}
+
+// CONTENT HEADER
+// 0          2        4           12               14
+// +----------+--------+-----------+----------------+------------- - -
+// | class-id | weight | body size | property flags | property list...
+// +----------+--------+-----------+----------------+------------- - -
+//    short     short    long long       short        remainder...
+//
+func (f *headerFrame) write(w io.Writer) (err error) {
+	var payload bytes.Buffer
+	var zeroTime time.Time
+
+	if err = binary.Write(&payload, binary.BigEndian, f.ClassId); err != nil {
+		return
+	}
+
+	if err = binary.Write(&payload, binary.BigEndian, f.weight); err != nil {
+		return
+	}
+
+	if err = binary.Write(&payload, binary.BigEndian, f.Size); err != nil {
+		return
+	}
+
+	// First pass will build the mask to be serialized, second pass will serialize
+	// each of the fields that appear in the mask.
+
+	var mask uint16
+
+	if len(f.Properties.ContentType) > 0 {
+		mask = mask | flagContentType
+	}
+	if len(f.Properties.ContentEncoding) > 0 {
+		mask = mask | flagContentEncoding
+	}
+	if f.Properties.Headers != nil && len(f.Properties.Headers) > 0 {
+		mask = mask | flagHeaders
+	}
+	if f.Properties.DeliveryMode > 0 {
+		mask = mask | flagDeliveryMode
+	}
+	if f.Properties.Priority > 0 {
+		mask = mask | flagPriority
+	}
+	if len(f.Properties.CorrelationId) > 0 {
+		mask = mask | flagCorrelationId
+	}
+	if len(f.Properties.ReplyTo) > 0 {
+		mask = mask | flagReplyTo
+	}
+	if len(f.Properties.Expiration) > 0 {
+		mask = mask | flagExpiration
+	}
+	if len(f.Properties.MessageId) > 0 {
+		mask = mask | flagMessageId
+	}
+	if f.Properties.Timestamp != zeroTime {
+		mask = mask | flagTimestamp
+	}
+	if len(f.Properties.Type) > 0 {
+		mask = mask | flagType
+	}
+	if len(f.Properties.UserId) > 0 {
+		mask = mask | flagUserId
+	}
+	if len(f.Properties.AppId) > 0 {
+		mask = mask | flagAppId
+	}
+
+	if err = binary.Write(&payload, binary.BigEndian, mask); err != nil {
+		return
+	}
+
+	if hasProperty(mask, flagContentType) {
+		if err = writeShortstr(&payload, f.Properties.ContentType); err != nil {
+			return
+		}
+	}
+	if hasProperty(mask, flagContentEncoding) {
+		if err = writeShortstr(&payload, f.Properties.ContentEncoding); err != nil {
+			return
+		}
+	}
+	if hasProperty(mask, flagHeaders) {
+		if err = writeTable(&payload, f.Properties.Headers); err != nil {
+			return
+		}
+	}
+	if hasProperty(mask, flagDeliveryMode) {
+		if err = binary.Write(&payload, binary.BigEndian, f.Properties.DeliveryMode); err != nil {
+			return
+		}
+	}
+	if hasProperty(mask, flagPriority) {
+		if err = binary.Write(&payload, binary.BigEndian, f.Properties.Priority); err != nil {
+			return
+		}
+	}
+	if hasProperty(mask, flagCorrelationId) {
+		if err = writeShortstr(&payload, f.Properties.CorrelationId); err != nil {
+			return
+		}
+	}
+	if hasProperty(mask, flagReplyTo) {
+		if err = writeShortstr(&payload, f.Properties.ReplyTo); err != nil {
+			return
+		}
+	}
+	if hasProperty(mask, flagExpiration) {
+		if err = writeShortstr(&payload, f.Properties.Expiration); err != nil {
+			return
+		}
+	}
+	if hasProperty(mask, flagMessageId) {
+		if err = writeShortstr(&payload, f.Properties.MessageId); err != nil {
+			return
+		}
+	}
+	if hasProperty(mask, flagTimestamp) {
+		if err = binary.Write(&payload, binary.BigEndian, uint64(f.Properties.Timestamp.Unix())); err != nil {
+			return
+		}
+	}
+	if hasProperty(mask, flagType) {
+		if err = writeShortstr(&payload, f.Properties.Type); err != nil {
+			return
+		}
+	}
+	if hasProperty(mask, flagUserId) {
+		if err = writeShortstr(&payload, f.Properties.UserId); err != nil {
+			return
+		}
+	}
+	if hasProperty(mask, flagAppId) {
+		if err = writeShortstr(&payload, f.Properties.AppId); err != nil {
+			return
+		}
+	}
+
+	return writeFrame(w, frameHeader, f.ChannelId, payload.Bytes())
+}
+
+// Body
+//
+// Payload is one byterange from the full body who's size is declared in the
+// Header frame
+func (f *bodyFrame) write(w io.Writer) (err error) {
+	return writeFrame(w, frameBody, f.ChannelId, f.Body)
+}
+
+func writeFrame(w io.Writer, typ uint8, channel uint16, payload []byte) (err error) {
+	end := []byte{frameEnd}
+	size := uint(len(payload))
+
+	_, err = w.Write([]byte{
+		byte(typ),
+		byte((channel & 0xff00) >> 8),
+		byte((channel & 0x00ff) >> 0),
+		byte((size & 0xff000000) >> 24),
+		byte((size & 0x00ff0000) >> 16),
+		byte((size & 0x0000ff00) >> 8),
+		byte((size & 0x000000ff) >> 0),
+	})
+
+	if err != nil {
+		return
+	}
+
+	if _, err = w.Write(payload); err != nil {
+		return
+	}
+
+	if _, err = w.Write(end); err != nil {
+		return
+	}
+
+	return
+}
+
+func writeShortstr(w io.Writer, s string) (err error) {
+	b := []byte(s)
+
+	var length = uint8(len(b))
+
+	if err = binary.Write(w, binary.BigEndian, length); err != nil {
+		return
+	}
+
+	if _, err = w.Write(b[:length]); err != nil {
+		return
+	}
+
+	return
+}
+
+func writeLongstr(w io.Writer, s string) (err error) {
+	b := []byte(s)
+
+	var length = uint32(len(b))
+
+	if err = binary.Write(w, binary.BigEndian, length); err != nil {
+		return
+	}
+
+	if _, err = w.Write(b[:length]); err != nil {
+		return
+	}
+
+	return
+}
+
+/*
+'A': []interface{}
+'D': Decimal
+'F': Table
+'I': int32
+'S': string
+'T': time.Time
+'V': nil
+'b': byte
+'d': float64
+'f': float32
+'l': int64
+'s': int16
+'t': bool
+'x': []byte
+*/
+func writeField(w io.Writer, value interface{}) (err error) {
+	var buf [9]byte
+	var enc []byte
+
+	switch v := value.(type) {
+	case bool:
+		buf[0] = 't'
+		if v {
+			buf[1] = byte(1)
+		} else {
+			buf[1] = byte(0)
+		}
+		enc = buf[:2]
+
+	case byte:
+		buf[0] = 'b'
+		buf[1] = byte(v)
+		enc = buf[:2]
+
+	case int16:
+		buf[0] = 's'
+		binary.BigEndian.PutUint16(buf[1:3], uint16(v))
+		enc = buf[:3]
+
+	case int32:
+		buf[0] = 'I'
+		binary.BigEndian.PutUint32(buf[1:5], uint32(v))
+		enc = buf[:5]
+
+	case int64:
+		buf[0] = 'l'
+		binary.BigEndian.PutUint64(buf[1:9], uint64(v))
+		enc = buf[:9]
+
+	case float32:
+		buf[0] = 'f'
+		binary.BigEndian.PutUint32(buf[1:5], math.Float32bits(v))
+		enc = buf[:5]
+
+	case float64:
+		buf[0] = 'd'
+		binary.BigEndian.PutUint64(buf[1:9], math.Float64bits(v))
+		enc = buf[:9]
+
+	case Decimal:
+		buf[0] = 'D'
+		buf[1] = byte(v.Scale)
+		binary.BigEndian.PutUint32(buf[2:6], uint32(v.Value))
+		enc = buf[:6]
+
+	case string:
+		buf[0] = 'S'
+		binary.BigEndian.PutUint32(buf[1:5], uint32(len(v)))
+		enc = append(buf[:5], []byte(v)...)
+
+	case []interface{}: // field-array
+		buf[0] = 'A'
+
+		sec := new(bytes.Buffer)
+		for _, val := range v {
+			if err = writeField(sec, val); err != nil {
+				return
+			}
+		}
+
+		binary.BigEndian.PutUint32(buf[1:5], uint32(sec.Len()))
+		if _, err = w.Write(buf[:5]); err != nil {
+			return
+		}
+
+		if _, err = w.Write(sec.Bytes()); err != nil {
+			return
+		}
+
+		return
+
+	case time.Time:
+		buf[0] = 'T'
+		binary.BigEndian.PutUint64(buf[1:9], uint64(v.Unix()))
+		enc = buf[:9]
+
+	case Table:
+		if _, err = w.Write([]byte{'F'}); err != nil {
+			return
+		}
+		return writeTable(w, v)
+
+	case []byte:
+		buf[0] = 'x'
+		binary.BigEndian.PutUint32(buf[1:5], uint32(len(v)))
+		if _, err = w.Write(buf[0:5]); err != nil {
+			return
+		}
+		if _, err = w.Write(v); err != nil {
+			return
+		}
+		return
+
+	case nil:
+		buf[0] = 'V'
+		enc = buf[:1]
+
+	default:
+		return ErrFieldType
+	}
+
+	_, err = w.Write(enc)
+
+	return
+}
+
+func writeTable(w io.Writer, table Table) (err error) {
+	var buf bytes.Buffer
+
+	for key, val := range table {
+		if err = writeShortstr(&buf, key); err != nil {
+			return
+		}
+		if err = writeField(&buf, val); err != nil {
+			return
+		}
+	}
+
+	return writeLongstr(w, string(buf.Bytes()))
+}