This document is for py-amqp's development version, which can be significantly different from previous releases. Get the stable docs here: 5.3.

# Source code for amqp.channel

    """AMQP Channels."""
    # Copyright (C) 2007-2008 Barry Pederson <bp@barryp.org>

    import logging
    import socket
    from collections import defaultdict
    from queue import Queue

    from vine import ensure_promise

    from . import spec
    from .abstract_channel import AbstractChannel
    from .exceptions import (ChannelError, ConsumerCancelled, MessageNacked,
                             RecoverableChannelError, RecoverableConnectionError,
                             error_for_code)
    from .protocol import queue_declare_ok_t

    __all__ = ('Channel',)

    AMQP_LOGGER = logging.getLogger('amqp')

    REJECTED_MESSAGE_WITHOUT_CALLBACK = """\
    Rejecting message with delivery tag %r for reason of having no callbacks.
    consumer_tag=%r exchange=%r routing_key=%r.\
    """

    class VDeprecationWarning(DeprecationWarning):
        pass

    class Channel(AbstractChannel):
        """AMQP Channel.

        The channel class provides methods for a client to establish a
        virtual connection - a channel - to a server and for both peers to
        operate the virtual connection thereafter.

        GRAMMAR::

            channel             = open-channel *use-channel close-channel
            open-channel        = C:OPEN S:OPEN-OK
            use-channel         = C:FLOW S:FLOW-OK
                                / S:FLOW C:FLOW-OK
                                / functional-class
            close-channel       = C:CLOSE S:CLOSE-OK
                                / S:CLOSE C:CLOSE-OK

        Create a channel bound to a connection and using the specified
        numeric channel_id, and open on the server.

        The 'auto_decode' parameter (defaults to True), indicates
        whether the library should attempt to decode the body
        of Messages to a Unicode string if there's a 'content_encoding'
        property for the message.  If there's no 'content_encoding'
        property, or the decode raises an Exception, the message body
        is left as plain bytes.
        """

        _METHODS = {
            spec.method(spec.Channel.Close, 'BsBB'),
            spec.method(spec.Channel.CloseOk),
            spec.method(spec.Channel.Flow, 'b'),
            spec.method(spec.Channel.FlowOk, 'b'),
            spec.method(spec.Channel.OpenOk),
            spec.method(spec.Exchange.DeclareOk),
            spec.method(spec.Exchange.DeleteOk),
            spec.method(spec.Exchange.BindOk),
            spec.method(spec.Exchange.UnbindOk),
            spec.method(spec.Queue.BindOk),
            spec.method(spec.Queue.UnbindOk),
            spec.method(spec.Queue.DeclareOk, 'sll'),
            spec.method(spec.Queue.DeleteOk, 'l'),
            spec.method(spec.Queue.PurgeOk, 'l'),
            spec.method(spec.Basic.Cancel, 's'),
            spec.method(spec.Basic.CancelOk, 's'),
            spec.method(spec.Basic.ConsumeOk, 's'),
            spec.method(spec.Basic.Deliver, 'sLbss', content=True),
            spec.method(spec.Basic.GetEmpty, 's'),
            spec.method(spec.Basic.GetOk, 'Lbssl', content=True),
            spec.method(spec.Basic.QosOk),
            spec.method(spec.Basic.RecoverOk),
            spec.method(spec.Basic.Return, 'Bsss', content=True),
            spec.method(spec.Tx.CommitOk),
            spec.method(spec.Tx.RollbackOk),
            spec.method(spec.Tx.SelectOk),
            spec.method(spec.Confirm.SelectOk),
            spec.method(spec.Basic.Ack, 'Lb'),
            spec.method(spec.Basic.Nack, 'Lb'),
        }
        _METHODS = {m.method_sig: m for m in _METHODS}

        _ALLOWED_METHODS_WHEN_CLOSING = (
            spec.Channel.Close, spec.Channel.CloseOk
        )

        def __init__(self, connection,
                     channel_id=None, auto_decode=True, on_open=None):
            if channel_id:
                connection._claim_channel_id(channel_id)
            else:
                channel_id = connection._get_free_channel_id()

            AMQP_LOGGER.debug('using channel_id: %s', channel_id)

            super().__init__(connection, channel_id)

            self.is_open = False
            self.active = True  # Flow control
            self.returned_messages = Queue()
            self.callbacks = {}
            self.cancel_callbacks = {}
            self.auto_decode = auto_decode
            self.events = defaultdict(set)
            self.no_ack_consumers = set()

            self.on_open = ensure_promise(on_open)

            # set first time basic_publish_confirm is called
            # and publisher confirms are enabled for this channel.
            self._confirm_selected = False
            if self.connection.confirm_publish:
                self.basic_publish = self.basic_publish_confirm

            __slots__ = (
            "is_open",
            "active",
            "returned_messages",
            "callbacks",
            "cancel_callbacks",
            "events",
            "no_ack_consumers",
            "on_open",
            "_confirm_selected",
            )

        def then(self, on_success, on_error=None):
            return self.on_open.then(on_success, on_error)

        def _setup_listeners(self):
            self._callbacks.update({
                spec.Channel.Close: self._on_close,
                spec.Channel.CloseOk: self._on_close_ok,
                spec.Channel.Flow: self._on_flow,
                spec.Channel.OpenOk: self._on_open_ok,
                spec.Basic.Cancel: self._on_basic_cancel,
                spec.Basic.CancelOk: self._on_basic_cancel_ok,
                spec.Basic.Deliver: self._on_basic_deliver,
                spec.Basic.Return: self._on_basic_return,
                spec.Basic.Ack: self._on_basic_ack,
                spec.Basic.Nack: self._on_basic_nack,
            })

        def collect(self):
            """Tear down this object.

            Best called after we've agreed to close with the server.
            """
            AMQP_LOGGER.debug('Closed channel #%s', self.channel_id)
            self.is_open = False
            channel_id, self.channel_id = self.channel_id, None
            connection, self.connection = self.connection, None
            if connection:
                connection.channels.pop(channel_id, None)
                try:
                    connection._used_channel_ids.remove(channel_id)
                except ValueError:
                    # channel id already removed
                    pass
            self.callbacks.clear()
            self.cancel_callbacks.clear()
            self.events.clear()
            self.no_ack_consumers.clear()

        def _do_revive(self):
            self.is_open = False
            self.open()

        def close(self, reply_code=0, reply_text='', method_sig=(0, 0),
                  argsig='BsBB'):
            """Request a channel close.

            This method indicates that the sender wants to close the
            channel. This may be due to internal conditions (e.g. a forced
            shut-down) or due to an error handling a specific method, i.e.
            an exception.  When a close is due to an exception, the sender
            provides the class and method id of the method which caused
            the exception.

            RULE:

                After sending this method any received method except
                Channel.Close-OK MUST be discarded.

            RULE:

                The peer sending this method MAY use a counter or timeout
                to detect failure of the other peer to respond correctly
                with Channel.Close-OK..

            PARAMETERS:
                reply_code: short

                    The reply code. The AMQ reply codes are defined in AMQ
                    RFC 011.

                reply_text: shortstr

                    The localised reply text.  This text can be logged as an
                    aid to resolving issues.

                class_id: short

                    failing method class

                    When the close is provoked by a method exception, this
                    is the class of the method.

                method_id: short

                    failing method ID

                    When the close is provoked by a method exception, this
                    is the ID of the method.
            """
            try:
                if self.connection is None:
                    return
                if self.connection.channels is None:
                    return
                if not self.is_open:
                    return

                self.is_closing = True
                return self.send_method(
                    spec.Channel.Close, argsig,
                    (reply_code, reply_text, method_sig[0], method_sig[1]),
                    wait=spec.Channel.CloseOk,
                )
            finally:
                self.is_closing = False
                self.connection = None

        def _on_close(self, reply_code, reply_text, class_id, method_id):
            """Request a channel close.

            This method indicates that the sender wants to close the
            channel. This may be due to internal conditions (e.g. a forced
            shut-down) or due to an error handling a specific method, i.e.
            an exception.  When a close is due to an exception, the sender
            provides the class and method id of the method which caused
            the exception.

            RULE:

                After sending this method any received method except
                Channel.Close-OK MUST be discarded.

            RULE:

                The peer sending this method MAY use a counter or timeout
                to detect failure of the other peer to respond correctly
                with Channel.Close-OK..

            PARAMETERS:
                reply_code: short

                    The reply code. The AMQ reply codes are defined in AMQ
                    RFC 011.

                reply_text: shortstr

                    The localised reply text.  This text can be logged as an
                    aid to resolving issues.

                class_id: short

                    failing method class

                    When the close is provoked by a method exception, this
                    is the class of the method.

                method_id: short

                    failing method ID

                    When the close is provoked by a method exception, this
                    is the ID of the method.
            """
            self.send_method(spec.Channel.CloseOk)
            if not self.connection.is_closing:
                self._do_revive()
                raise error_for_code(
                    reply_code, reply_text, (class_id, method_id), ChannelError,
                )

        def _on_close_ok(self):
            """Confirm a channel close.

            This method confirms a Channel.Close method and tells the
            recipient that it is safe to release resources for the channel
            and close the socket.

            RULE:

                A peer that detects a socket closure without having
                received a Channel.Close-Ok handshake method SHOULD log
                the error.
            """
            self.collect()

        def flow(self, active):
            """Enable/disable flow from peer.

            This method asks the peer to pause or restart the flow of
            content data. This is a simple flow-control mechanism that a
            peer 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.  The
            peer that receives a request to stop sending content should
            finish sending the current content, if any, and then wait
            until it receives a Flow restart method.

            RULE:

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

            RULE:

                When sending content data in multiple frames, a peer
                SHOULD monitor the channel for incoming methods and
                respond to a Channel.Flow as rapidly as possible.

            RULE:

                A peer MAY use the Channel.Flow method to throttle
                incoming content data for internal reasons, for example,
                when exchanging data over a slower connection.

            RULE:

                The peer that requests a Channel.Flow method MAY
                disconnect and/or ban a peer that does not respect the
                request.

            PARAMETERS:
                active: boolean

                    start/stop content frames

                    If True, the peer starts sending content frames.  If
                    False, the peer stops sending content frames.
            """
            return self.send_method(
                spec.Channel.Flow, 'b', (active,), wait=spec.Channel.FlowOk,
            )

        def _on_flow(self, active):
            """Enable/disable flow from peer.

            This method asks the peer to pause or restart the flow of
            content data. This is a simple flow-control mechanism that a
            peer 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.  The
            peer that receives a request to stop sending content should
            finish sending the current content, if any, and then wait
            until it receives a Flow restart method.

            RULE:

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

            RULE:

                When sending content data in multiple frames, a peer
                SHOULD monitor the channel for incoming methods and
                respond to a Channel.Flow as rapidly as possible.

            RULE:

                A peer MAY use the Channel.Flow method to throttle
                incoming content data for internal reasons, for example,
                when exchanging data over a slower connection.

            RULE:

                The peer that requests a Channel.Flow method MAY
                disconnect and/or ban a peer that does not respect the
                request.

            PARAMETERS:
                active: boolean

                    start/stop content frames

                    If True, the peer starts sending content frames.  If
                    False, the peer stops sending content frames.
            """
            self.active = active
            self._x_flow_ok(self.active)

        def _x_flow_ok(self, active):
            """Confirm a flow method.

            Confirms to the peer that a flow command was received and
            processed.

            PARAMETERS:
                active: boolean

                    current flow setting

                    Confirms the setting of the processed flow method:
                    True means the peer will start sending or continue
                    to send content frames; False means it will not.
            """
            return self.send_method(spec.Channel.FlowOk, 'b', (active,))

        def open(self):
            """Open a channel for use.

            This method opens a virtual connection (a channel).

            RULE:

                This method MUST NOT be called when the channel is already
                open.

            PARAMETERS:
                out_of_band: shortstr (DEPRECATED)

                    out-of-band settings

                    Configures out-of-band transfers on this channel.  The
                    syntax and meaning of this field will be formally
                    defined at a later date.
            """
            if self.is_open:
                return

            return self.send_method(
                spec.Channel.Open, 's', ('',), wait=spec.Channel.OpenOk,
            )

        def _on_open_ok(self):
            """Signal that the channel is ready.

            This method signals to the client that the channel is ready
            for use.
            """
            self.is_open = True
            self.on_open(self)
            AMQP_LOGGER.debug('Channel open')

        #############
        #
        #  Exchange
        #
        #
        # work with exchanges
        #
        # Exchanges match and distribute messages across queues.
        # Exchanges can be configured in the server or created at runtime.
        #
        # GRAMMAR::
        #
        #     exchange            = C:DECLARE  S:DECLARE-OK
        #                         / C:DELETE   S:DELETE-OK
        #
        # RULE:
        #
        #     The server MUST implement the direct and fanout exchange
        #     types, and predeclare the corresponding exchanges named
        #     amq.direct and amq.fanout in each virtual host. The server
        #     MUST also predeclare a direct exchange to act as the default
        #     exchange for content Publish methods and for default queue
        #     bindings.
        #
        # RULE:
        #
        #     The server SHOULD implement the topic exchange type, and
        #     predeclare the corresponding exchange named amq.topic in
        #     each virtual host.
        #
        # RULE:
        #
        #     The server MAY implement the system exchange type, and
        #     predeclare the corresponding exchanges named amq.system in
        #     each virtual host. If the client attempts to bind a queue to
        #     the system exchange, the server MUST raise a connection
        #     exception with reply code 507 (not allowed).
        #

        def exchange_declare(self, exchange, type, passive=False, durable=False,
                             auto_delete=True, nowait=False, arguments=None,
                             argsig='BssbbbbbF'):
            """Declare exchange, create if needed.

            This method creates an exchange if it does not already exist,
            and if the exchange exists, verifies that it is of the correct
            and expected class.

            RULE:

                The server SHOULD support a minimum of 16 exchanges per
                virtual host and ideally, impose no limit except as
                defined by available resources.

            PARAMETERS:
                exchange: shortstr

                    RULE:

                        Exchange names starting with "amq." are reserved
                        for predeclared and standardised exchanges.  If
                        the client attempts to create an exchange starting
                        with "amq.", the server MUST raise a channel
                        exception with reply code 403 (access refused).

                type: shortstr

                    exchange type

                    Each exchange belongs to one of a set of exchange
                    types implemented by the server.  The exchange types
                    define the functionality of the exchange - i.e. how
                    messages are routed through it.  It is not valid or
                    meaningful to attempt to change the type of an
                    existing exchange.

                    RULE:

                        If the exchange already exists with a different
                        type, the server MUST raise a connection exception
                        with a reply code 507 (not allowed).

                    RULE:

                        If the server does not support the requested
                        exchange type it MUST raise a connection exception
                        with a reply code 503 (command invalid).

                passive: boolean

                    do not create exchange

                    If set, the server will not create the exchange.  The
                    client can use this to check whether an exchange
                    exists without modifying the server state.

                    RULE:

                        If set, and the exchange does not already exist,
                        the server MUST raise a channel exception with
                        reply code 404 (not found).

                durable: boolean

                    request a durable exchange

                    If set when creating a new exchange, the exchange will
                    be marked as durable.  Durable exchanges remain active
                    when a server restarts. Non-durable exchanges
                    (transient exchanges) are purged if/when a server
                    restarts.

                    RULE:

                        The server MUST support both durable and transient
                        exchanges.

                    RULE:

                        The server MUST ignore the durable field if the
                        exchange already exists.

                auto_delete: boolean

                    auto-delete when unused

                    If set, the exchange is deleted when all queues have
                    finished using it.

                    RULE:

                        The server SHOULD allow for a reasonable delay
                        between the point when it determines that an
                        exchange is not being used (or no longer used),
                        and the point when it deletes the exchange.  At
                        the least it must allow a client to create an
                        exchange and then bind a queue to it, with a small
                        but non-zero delay between these two actions.

                    RULE:

                        The server MUST ignore the auto-delete field if
                        the exchange already exists.

                nowait: boolean

                    do not send a reply method

                    If set, the server will not respond to the method. The
                    client should not wait for a reply method.  If the
                    server could not complete the method it will raise a
                    channel or connection exception.

                arguments: table

                    arguments for declaration

                    A set of arguments for the declaration. The syntax and
                    semantics of these arguments depends on the server
                    implementation.  This field is ignored if passive is
                    True.
            """
            self.send_method(
                spec.Exchange.Declare, argsig,
                (0, exchange, type, passive, durable, auto_delete,
                 False, nowait, arguments),
                wait=None if nowait else spec.Exchange.DeclareOk,
            )

        def exchange_delete(self, exchange, if_unused=False, nowait=False,
                            argsig='Bsbb'):
            """Delete an exchange.

            This method deletes an exchange.  When an exchange is deleted
            all queue bindings on the exchange are cancelled.

            PARAMETERS:
                exchange: shortstr

                    RULE:

                        The exchange MUST exist. Attempting to delete a
                        non-existing exchange causes a channel exception.

                if_unused: boolean

                    delete only if unused

                    If set, 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 raises a
                    channel exception instead.

                    RULE:

                        If set, the server SHOULD delete the exchange but
                        only if it has no queue bindings.

                    RULE:

                        If set, the server SHOULD raise a channel
                        exception if the exchange is in use.

                nowait: boolean

                    do not send a reply method

                    If set, the server will not respond to the method. The
                    client should not wait for a reply method.  If the
                    server could not complete the method it will raise a
                    channel or connection exception.
            """
            return self.send_method(
                spec.Exchange.Delete, argsig, (0, exchange, if_unused, nowait),
                wait=None if nowait else spec.Exchange.DeleteOk,
            )

        def exchange_bind(self, destination, source='', routing_key='',
                          nowait=False, arguments=None, argsig='BsssbF'):
            """Bind an exchange to an exchange.

            RULE:

                A server MUST allow and ignore duplicate bindings - that
                is, two or more bind methods for a specific exchanges,
                with identical arguments - without treating these as an
                error.

            RULE:

                A server MUST allow cycles of exchange bindings to be
                created including allowing an exchange to be bound to
                itself.

            RULE:

                A server MUST not deliver the same message more than once
                to a destination exchange, even if the topology of
                exchanges and bindings results in multiple (even infinite)
                routes to that exchange.

            PARAMETERS:
                reserved-1: short

                destination: shortstr

                    Specifies the name of the destination exchange to
                    bind.

                    RULE:

                        A client MUST NOT be allowed to bind a non-
                        existent destination exchange.

                    RULE:

                        The server MUST accept a blank exchange name to
                        mean the default exchange.

                source: shortstr

                    Specifies the name of the source exchange to bind.

                    RULE:

                        A client MUST NOT be allowed to bind a non-
                        existent source exchange.

                    RULE:

                        The server MUST accept a blank exchange name to
                        mean the default exchange.

                routing-key: shortstr

                    Specifies the routing key for the binding. The routing
                    key is used for routing messages depending on the
                    exchange configuration. Not all exchanges use a
                    routing key - refer to the specific exchange
                    documentation.

                no-wait: bit

                arguments: table

                    A set of arguments for the binding. The syntax and
                    semantics of these arguments depends on the exchange
                    class.
            """
            return self.send_method(
                spec.Exchange.Bind, argsig,
                (0, destination, source, routing_key, nowait, arguments),
                wait=None if nowait else spec.Exchange.BindOk,
            )

        def exchange_unbind(self, destination, source='', routing_key='',
                            nowait=False, arguments=None, argsig='BsssbF'):
            """Unbind an exchange from an exchange.

            RULE:

                If a unbind fails, the server MUST raise a connection
                exception.

            PARAMETERS:
                reserved-1: short

                destination: shortstr

                    Specifies the name of the destination exchange to
                    unbind.

                    RULE:

                        The client MUST NOT attempt to unbind an exchange
                        that does not exist from an exchange.

                    RULE:

                        The server MUST accept a blank exchange name to
                        mean the default exchange.

                source: shortstr

                    Specifies the name of the source exchange to unbind.

                    RULE:

                        The client MUST NOT attempt to unbind an exchange
                        from an exchange that does not exist.

                    RULE:

                        The server MUST accept a blank exchange name to
                        mean the default exchange.

                routing-key: shortstr

                    Specifies the routing key of the binding to unbind.

                no-wait: bit

                arguments: table

                    Specifies the arguments of the binding to unbind.
            """
            return self.send_method(
                spec.Exchange.Unbind, argsig,
                (0, destination, source, routing_key, nowait, arguments),
                wait=None if nowait else spec.Exchange.UnbindOk,
            )

        #############
        #
        #  Queue
        #
        #
        # work with queues
        #
        # Queues store and forward messages.  Queues can be configured in
        # the server or created at runtime.  Queues must be attached to at
        # least one exchange in order to receive messages from publishers.
        #
        # GRAMMAR::
        #
        #     queue               = C:DECLARE  S:DECLARE-OK
        #                         / C:BIND     S:BIND-OK
        #                         / C:PURGE    S:PURGE-OK
        #                         / C:DELETE   S:DELETE-OK
        #
        # RULE:
        #
        #     A server MUST allow any content class to be sent to any
        #     queue, in any mix, and queue and delivery these content
        #     classes independently. Note that all methods that fetch
        #     content off queues are specific to a given content class.
        #

        def queue_bind(self, queue, exchange='', routing_key='',
                       nowait=False, arguments=None, argsig='BsssbF'):
            """Bind queue to an exchange.

            This method binds a queue to an exchange.  Until a queue is
            bound it will not receive any messages.  In a classic
            messaging model, store-and-forward queues are bound to a dest
            exchange and subscription queues are bound to a dest_wild
            exchange.

            RULE:

                A server MUST allow ignore duplicate bindings - that is,
                two or more bind methods for a specific queue, with
                identical arguments - without treating these as an error.

            RULE:

                If a bind fails, the server MUST raise a connection
                exception.

            RULE:

                The server MUST NOT allow a durable queue to bind to a
                transient exchange. If the client attempts this the server
                MUST raise a channel exception.

            RULE:

                Bindings for durable queues are automatically durable and
                the server SHOULD restore such bindings after a server
                restart.

            RULE:

                The server SHOULD support at least 4 bindings per queue,
                and ideally, impose no limit except as defined by
                available resources.

            PARAMETERS:
                queue: shortstr

                    Specifies the name of the queue to bind.  If the queue
                    name is empty, refers to the current queue for the
                    channel, which is the last declared queue.

                    RULE:

                        If the client did not previously declare a queue,
                        and the queue name in this method is empty, the
                        server MUST raise a connection exception with
                        reply code 530 (not allowed).

                    RULE:

                        If the queue does not exist the server MUST raise
                        a channel exception with reply code 404 (not
                        found).

                exchange: shortstr

                    The name of the exchange to bind to.

                    RULE:

                        If the exchange does not exist the server MUST
                        raise a channel exception with reply code 404 (not
                        found).

                routing_key: shortstr

                    message routing key

                    Specifies the routing key for the binding.  The
                    routing key is used for routing messages depending on
                    the exchange configuration. Not all exchanges use a
                    routing key - refer to the specific exchange
                    documentation.  If the routing key is empty and the
                    queue name is empty, the routing key will be the
                    current queue for the channel, which is the last
                    declared queue.

                nowait: boolean

                    do not send a reply method

                    If set, the server will not respond to the method. The
                    client should not wait for a reply method.  If the
                    server could not complete the method it will raise a
                    channel or connection exception.

                arguments: table

                    arguments for binding

                    A set of arguments for the binding.  The syntax and
                    semantics of these arguments depends on the exchange
                    class.
            """
            return self.send_method(
                spec.Queue.Bind, argsig,
                (0, queue, exchange, routing_key, nowait, arguments),
                wait=None if nowait else spec.Queue.BindOk,
            )

        def queue_unbind(self, queue, exchange, routing_key='',
                         nowait=False, arguments=None, argsig='BsssF'):
            """Unbind a queue from an exchange.

            This method unbinds a queue from an exchange.

            RULE:

                If a unbind fails, the server MUST raise a connection exception.

            PARAMETERS:
                queue: shortstr

                    Specifies the name of the queue to unbind.

                    RULE:

                        The client MUST either specify a queue name or have
                        previously declared a queue on the same channel

                    RULE:

                        The client MUST NOT attempt to unbind a queue that
                        does not exist.

                exchange: shortstr

                    The name of the exchange to unbind from.

                    RULE:

                        The client MUST NOT attempt to unbind a queue from an
                        exchange that does not exist.

                    RULE:

                        The server MUST accept a blank exchange name to mean
                        the default exchange.

                routing_key: shortstr

                    routing key of binding

                    Specifies the routing key of the binding to unbind.

                arguments: table

                    arguments of binding

                    Specifies the arguments of the binding to unbind.
            """
            return self.send_method(
                spec.Queue.Unbind, argsig,
                (0, queue, exchange, routing_key, arguments),
                wait=None if nowait else spec.Queue.UnbindOk,
            )

        def queue_declare(self, queue='', passive=False, durable=False,
                          exclusive=False, auto_delete=True, nowait=False,
                          arguments=None, argsig='BsbbbbbF'):
            """Declare queue, create if needed.

            This method creates or checks a queue.  When creating a new
            queue the client can specify various properties that control
            the durability of the queue and its contents, and the level of
            sharing for the queue.

            RULE:

                The server MUST create a default binding for a newly-
                created queue to the default exchange, which is an
                exchange of type 'direct'.

            RULE:

                The server SHOULD support a minimum of 256 queues per
                virtual host and ideally, impose no limit except as
                defined by available resources.

            PARAMETERS:
                queue: shortstr

                    RULE:

                        The queue name MAY be empty, in which case the
                        server MUST create a new queue with a unique
                        generated name and return this to the client in
                        the Declare-Ok method.

                    RULE:

                        Queue names starting with "amq." are reserved for
                        predeclared and standardised server queues.  If
                        the queue name starts with "amq." and the passive
                        option is False, the server MUST raise a connection
                        exception with reply code 403 (access refused).

                passive: boolean

                    do not create queue

                    If set, the server will not create the queue.  The
                    client can use this to check whether a queue exists
                    without modifying the server state.

                    RULE:

                        If set, and the queue does not already exist, the
                        server MUST respond with a reply code 404 (not
                        found) and raise a channel exception.

                durable: boolean

                    request a durable queue

                    If set when creating a new queue, the queue will be
                    marked as durable.  Durable queues remain active when
                    a server restarts. Non-durable queues (transient
                    queues) are purged if/when a server restarts.  Note
                    that durable queues do not necessarily hold persistent
                    messages, although it does not make sense to send
                    persistent messages to a transient queue.

                    RULE:

                        The server MUST recreate the durable queue after a
                        restart.

                    RULE:

                        The server MUST support both durable and transient
                        queues.

                    RULE:

                        The server MUST ignore the durable field if the
                        queue already exists.

                exclusive: boolean

                    request an exclusive queue

                    Exclusive queues may only be consumed from by the
                    current connection. Setting the 'exclusive' flag
                    always implies 'auto-delete'.

                    RULE:

                        The server MUST support both exclusive (private)
                        and non-exclusive (shared) queues.

                    RULE:

                        The server MUST raise a channel exception if
                        'exclusive' is specified and the queue already
                        exists and is owned by a different connection.

                auto_delete: boolean

                    auto-delete queue when unused

                    If set, the queue is deleted when all consumers have
                    finished using it. Last consumer can be cancelled
                    either explicitly or because its channel is closed. If
                    there was no consumer ever on the queue, it won't be
                    deleted.

                    RULE:

                        The server SHOULD allow for a reasonable delay
                        between the point when it determines that a queue
                        is not being used (or no longer used), and the
                        point when it deletes the queue.  At the least it
                        must allow a client to create a queue and then
                        create a consumer to read from it, with a small
                        but non-zero delay between these two actions.  The
                        server should equally allow for clients that may
                        be disconnected prematurely, and wish to re-
                        consume from the same queue without losing
                        messages.  We would recommend a configurable
                        timeout, with a suitable default value being one
                        minute.

                    RULE:

                        The server MUST ignore the auto-delete field if
                        the queue already exists.

                nowait: boolean

                    do not send a reply method

                    If set, the server will not respond to the method. The
                    client should not wait for a reply method.  If the
                    server could not complete the method it will raise a
                    channel or connection exception.

                arguments: table

                    arguments for declaration

                    A set of arguments for the declaration. The syntax and
                    semantics of these arguments depends on the server
                    implementation.  This field is ignored if passive is
                    True.

            Returns a tuple containing 3 items:
                the name of the queue (essential for automatically-named queues),
                message count and
                consumer count
            """
            self.send_method(
                spec.Queue.Declare, argsig,
                (0, queue, passive, durable, exclusive, auto_delete,
                 nowait, arguments),
            )
            if not nowait:
                return queue_declare_ok_t(*self.wait(
                    spec.Queue.DeclareOk, returns_tuple=True,
                ))

        def queue_delete(self, queue='',
                         if_unused=False, if_empty=False, nowait=False,
                         argsig='Bsbbb'):
            """Delete a queue.

            This method deletes a queue.  When a queue is deleted any
            pending messages are sent to a dead-letter queue if this is
            defined in the server configuration, and all consumers on the
            queue are cancelled.

            RULE:

                The server SHOULD use a dead-letter queue to hold messages
                that were pending on a deleted queue, and MAY provide
                facilities for a system administrator to move these
                messages back to an active queue.

            PARAMETERS:
                queue: shortstr

                    Specifies the name of the queue to delete. If the
                    queue name is empty, refers to the current queue for
                    the channel, which is the last declared queue.

                    RULE:

                        If the client did not previously declare a queue,
                        and the queue name in this method is empty, the
                        server MUST raise a connection exception with
                        reply code 530 (not allowed).

                    RULE:

                        The queue must exist. Attempting to delete a non-
                        existing queue causes a channel exception.

                if_unused: boolean

                    delete only if unused

                    If set, the server will only delete the queue if it
                    has no consumers. If the queue has consumers the
                    server does does not delete it but raises a channel
                    exception instead.

                    RULE:

                        The server MUST respect the if-unused flag when
                        deleting a queue.

                if_empty: boolean

                    delete only if empty

                    If set, the server will only delete the queue if it
                    has no messages. If the queue is not empty the server
                    raises a channel exception.

                nowait: boolean

                    do not send a reply method

                    If set, the server will not respond to the method. The
                    client should not wait for a reply method.  If the
                    server could not complete the method it will raise a
                    channel or connection exception.

            If nowait is False, returns the number of deleted messages.
            """
            return self.send_method(
                spec.Queue.Delete, argsig,
                (0, queue, if_unused, if_empty, nowait),
                wait=None if nowait else spec.Queue.DeleteOk,
            )

        def queue_purge(self, queue='', nowait=False, argsig='Bsb'):
            """Purge a queue.

            This method removes all messages from a queue.  It does not
            cancel consumers.  Purged messages are deleted without any
            formal "undo" mechanism.

            RULE:

                A call to purge MUST result in an empty queue.

            RULE:

                On transacted channels the server MUST not purge messages
                that have already been sent to a client but not yet
                acknowledged.

            RULE:

                The server MAY implement a purge queue or log that allows
                system administrators to recover accidentally-purged
                messages.  The server SHOULD NOT keep purged messages in
                the same storage spaces as the live messages since the
                volumes of purged messages may get very large.

            PARAMETERS:
                queue: shortstr

                    Specifies the name of the queue to purge.  If the
                    queue name is empty, refers to the current queue for
                    the channel, which is the last declared queue.

                    RULE:

                        If the client did not previously declare a queue,
                        and the queue name in this method is empty, the
                        server MUST raise a connection exception with
                        reply code 530 (not allowed).

                    RULE:

                        The queue must exist. Attempting to purge a non-
                        existing queue causes a channel exception.

                nowait: boolean

                    do not send a reply method

                    If set, the server will not respond to the method. The
                    client should not wait for a reply method.  If the
                    server could not complete the method it will raise a
                    channel or connection exception.

            If nowait is False, returns a number of purged messages.
            """
            return self.send_method(
                spec.Queue.Purge, argsig, (0, queue, nowait),
                wait=None if nowait else spec.Queue.PurgeOk,
            )

        #############
        #
        #  Basic
        #
        #
        # work with basic content
        #
        # The Basic class provides methods that support an industry-
        # standard messaging model.
        #
        # GRAMMAR::
        #
        #     basic               = C:QOS S:QOS-OK
        #                         / C:CONSUME S:CONSUME-OK
        #                         / C:CANCEL S:CANCEL-OK
        #                         / C:PUBLISH content
        #                         / S:RETURN content
        #                         / S:DELIVER content
        #                         / C:GET ( S:GET-OK content / S:GET-EMPTY )
        #                         / C:ACK
        #                         / C:REJECT
        #
        # RULE:
        #
        #     The server SHOULD respect the persistent property of basic
        #     messages and SHOULD make a best-effort to hold persistent
        #     basic messages on a reliable storage mechanism.
        #
        # RULE:
        #
        #     The server MUST NOT discard a persistent basic message in
        #     case of a queue overflow. The server MAY use the
        #     Channel.Flow method to slow or stop a basic message
        #     publisher when necessary.
        #
        # RULE:
        #
        #     The server MAY overflow non-persistent basic messages to
        #     persistent storage and MAY discard or dead-letter non-
        #     persistent basic messages on a priority basis if the queue
        #     size exceeds some configured limit.
        #
        # RULE:
        #
        #     The server MUST implement at least 2 priority levels for
        #     basic messages, where priorities 0-4 and 5-9 are treated as
        #     two distinct levels. The server MAY implement up to 10
        #     priority levels.
        #
        # RULE:
        #
        #     The server MUST deliver messages of the same priority in
        #     order irrespective of their individual persistence.
        #
        # RULE:
        #
        #     The server MUST support both automatic and explicit
        #     acknowledgments on Basic content.
        #

        def basic_ack(self, delivery_tag, multiple=False, argsig='Lb'):
            """Acknowledge one or more messages.

            This method acknowledges one or more messages delivered via
            the Deliver or Get-Ok methods.  The client can ask to confirm
            a single message or a set of messages up to and including a
            specific message.

            PARAMETERS:
                delivery_tag: longlong

                    server-assigned delivery tag

                    The server-assigned and channel-specific delivery tag

                    RULE:

                        The delivery tag is valid only within the channel
                        from which the message was received.  I.e. a client
                        MUST NOT receive a message on one channel and then
                        acknowledge it on another.

                    RULE:

                        The server MUST NOT use a zero value for delivery
                        tags.  Zero is reserved for client use, meaning "all
                        messages so far received".

                multiple: boolean

                    acknowledge multiple messages

                    If set to True, the delivery tag is treated as "up to
                    and including", so that the client can acknowledge
                    multiple messages with a single method.  If set to
                    False, the delivery tag refers to a single message.
                    If the multiple field is True, and the delivery tag
                    is zero, tells the server to acknowledge all
                    outstanding messages.

                    RULE:

                        The server MUST validate that a non-zero delivery-
                        tag refers to an delivered message, and raise a
                        channel exception if this is not the case.
            """
            return self.send_method(
                spec.Basic.Ack, argsig, (delivery_tag, multiple),
            )

        def basic_cancel(self, consumer_tag, nowait=False, argsig='sb'):
            """End a queue consumer.

            This method cancels a consumer. This does not affect already
            delivered messages, but it does mean the server will not send
            any more messages for that consumer.  The client may receive
            an arbitrary number of messages in between sending the cancel
            method and receiving the cancel-ok reply.

            RULE:

                If the queue no longer exists when the client sends a
                cancel command, or the consumer has been cancelled for
                other reasons, this command has no effect.

            PARAMETERS:
                consumer_tag: shortstr

                    consumer tag

                    Identifier for the consumer, valid within the current
                    connection.

                    RULE:

                        The consumer tag is valid only within the channel
                        from which the consumer was created. I.e. a client
                        MUST NOT create a consumer in one channel and then
                        use it in another.

                nowait: boolean

                    do not send a reply method

                    If set, the server will not respond to the method. The
                    client should not wait for a reply method.  If the
                    server could not complete the method it will raise a
                    channel or connection exception.
            """
            if self.connection is not None:
                self.no_ack_consumers.discard(consumer_tag)
                return self.send_method(
                    spec.Basic.Cancel, argsig, (consumer_tag, nowait),
                    wait=None if nowait else spec.Basic.CancelOk,
                )

        def _on_basic_cancel(self, consumer_tag):
            """Consumer cancelled by server.

            Most likely the queue was deleted.

            """
            callback = self._remove_tag(consumer_tag)
            if callback:
                callback(consumer_tag)
            else:
                raise ConsumerCancelled(consumer_tag, spec.Basic.Cancel)

        def _on_basic_cancel_ok(self, consumer_tag):
            self._remove_tag(consumer_tag)

        def _remove_tag(self, consumer_tag):
            self.callbacks.pop(consumer_tag, None)
            return self.cancel_callbacks.pop(consumer_tag, None)

        def basic_consume(self, queue='', consumer_tag='', no_local=False,
                          no_ack=False, exclusive=False, nowait=False,
                          callback=None, arguments=None, on_cancel=None,
                          argsig='BssbbbbF'):
            """Start a queue consumer.

            This method asks the server to start a "consumer", which is a
            transient request for messages from a specific queue.
            Consumers last as long as the channel they were created on, or
            until the client cancels them.

            RULE:

                The server SHOULD support at least 16 consumers per queue,
                unless the queue was declared as private, and ideally,
                impose no limit except as defined by available resources.

            PARAMETERS:
                queue: shortstr

                    Specifies the name of the queue to consume from.  If
                    the queue name is null, refers to the current queue
                    for the channel, which is the last declared queue.

                    RULE:

                        If the client did not previously declare a queue,
                        and the queue name in this method is empty, the
                        server MUST raise a connection exception with
                        reply code 530 (not allowed).

                consumer_tag: shortstr

                    Specifies the identifier for the consumer. The
                    consumer tag is local to a connection, so two clients
                    can use the same consumer tags. If this field is empty
                    the server will generate a unique tag.

                    RULE:

                        The tag MUST NOT refer to an existing consumer. If
                        the client attempts to create two consumers with
                        the same non-empty tag the server MUST raise a
                        connection exception with reply code 530 (not
                        allowed).

                no_local: boolean

                    do not deliver own messages

                    If the no-local field is set the server will not send
                    messages to the client that published them.

                no_ack: boolean

                    no acknowledgment needed

                    If this field is set the server does not expect
                    acknowledgments for messages.  That is, when a message
                    is delivered to the client the server automatically and
                    silently acknowledges it on behalf of the client.  This
                    functionality increases performance but at the cost of
                    reliability.  Messages can get lost if a client dies
                    before it can deliver them to the application.

                exclusive: boolean

                    request exclusive access

                    Request exclusive consumer access, meaning only this
                    consumer can access the queue.

                    RULE:

                        If the server cannot grant exclusive access to the
                        queue when asked, - because there are other
                        consumers active - it MUST raise a channel
                        exception with return code 403 (access refused).

                nowait: boolean

                    do not send a reply method

                    If set, the server will not respond to the method. The
                    client should not wait for a reply method.  If the
                    server could not complete the method it will raise a
                    channel or connection exception.

                callback: Python callable

                    function/method called with each delivered message

                    For each message delivered by the broker, the
                    callable will be called with a Message object
                    as the single argument.  If no callable is specified,
                    messages are quietly discarded, no_ack should probably
                    be set to True in that case.
            """
            p = self.send_method(
                spec.Basic.Consume, argsig,
                (
                    0, queue, consumer_tag, no_local, no_ack, exclusive,
                    nowait, arguments
                ),
                wait=None if nowait else spec.Basic.ConsumeOk,
                returns_tuple=True
            )

            if not nowait:
                # send_method() returns (consumer_tag,) tuple.
                # consumer_tag is returned by broker using following rules:
                # * consumer_tag is not specified by client, random one
                #   is generated by Broker
                # * consumer_tag is provided by client, the same one
                #   is returned by broker
                consumer_tag = p[0]
            elif nowait and not consumer_tag:
                raise ValueError(
                    'Consumer tag must be specified when nowait is True'
                )

            self.callbacks[consumer_tag] = callback

            if on_cancel:
                self.cancel_callbacks[consumer_tag] = on_cancel
            if no_ack:
                self.no_ack_consumers.add(consumer_tag)

            if not nowait:
                return consumer_tag
            else:
                return p

        def _on_basic_deliver(self, consumer_tag, delivery_tag, redelivered,
                              exchange, routing_key, msg):
            msg.channel = self
            msg.delivery_info = {
                'consumer_tag': consumer_tag,
                'delivery_tag': delivery_tag,
                'redelivered': redelivered,
                'exchange': exchange,
                'routing_key': routing_key,
            }

            try:
                fun = self.callbacks[consumer_tag]
            except KeyError:
                AMQP_LOGGER.warning(
                    REJECTED_MESSAGE_WITHOUT_CALLBACK,
                    delivery_tag, consumer_tag, exchange, routing_key,
                )
                self.basic_reject(delivery_tag, requeue=True)
            else:
                fun(msg)

        def basic_get(self, queue='', no_ack=False, argsig='Bsb'):
            """Direct access to a queue.

            This method provides a direct access to the messages in a
            queue using a synchronous dialogue that is designed for
            specific types of application where synchronous functionality
            is more important than performance.

            PARAMETERS:
                queue: shortstr

                    Specifies the name of the queue to consume from.  If
                    the queue name is null, refers to the current queue
                    for the channel, which is the last declared queue.

                    RULE:

                        If the client did not previously declare a queue,
                        and the queue name in this method is empty, the
                        server MUST raise a connection exception with
                        reply code 530 (not allowed).

                no_ack: boolean

                    no acknowledgment needed

                    If this field is set the server does not expect
                    acknowledgments for messages.  That is, when a message
                    is delivered to the client the server automatically and
                    silently acknowledges it on behalf of the client.  This
                    functionality increases performance but at the cost of
                    reliability.  Messages can get lost if a client dies
                    before it can deliver them to the application.

            Non-blocking, returns a amqp.basic_message.Message object,
            or None if queue is empty.
            """
            ret = self.send_method(
                spec.Basic.Get, argsig, (0, queue, no_ack),
                wait=[spec.Basic.GetOk, spec.Basic.GetEmpty], returns_tuple=True,
            )
            if not ret or len(ret) < 2:
                return self._on_get_empty(*ret)
            return self._on_get_ok(*ret)

        def _on_get_empty(self, cluster_id=None):
            pass

        def _on_get_ok(self, delivery_tag, redelivered, exchange, routing_key,
                       message_count, msg):
            msg.channel = self
            msg.delivery_info = {
                'delivery_tag': delivery_tag,
                'redelivered': redelivered,
                'exchange': exchange,
                'routing_key': routing_key,
                'message_count': message_count
            }
            return msg

        def _basic_publish(self, msg, exchange='', routing_key='',
                           mandatory=False, immediate=False, timeout=None,
                           confirm_timeout=None,
                           argsig='Bssbb'):
            """Publish a message.

            This method publishes a message to a specific exchange. The
            message will be routed to queues as defined by the exchange
            configuration and distributed to any active consumers when the
            transaction, if any, is committed.

            When channel is in confirm mode (when Connection parameter
            confirm_publish is set to True), each message is confirmed.
            When broker rejects published message (e.g. due internal broker
            constrains), MessageNacked exception is raised and
            set confirm_timeout to wait maximum confirm_timeout second
            for message to confirm.

            PARAMETERS:
                exchange: shortstr

                    Specifies the name of the exchange to publish to.  The
                    exchange name can be empty, meaning the default
                    exchange.  If the exchange name is specified, and that
                    exchange does not exist, the server will raise a
                    channel exception.

                    RULE:

                        The server MUST accept a blank exchange name to
                        mean the default exchange.

                    RULE:

                        The exchange MAY refuse basic content in which
                        case it MUST raise a channel exception with reply
                        code 540 (not implemented).

                routing_key: shortstr

                    Message routing key

                    Specifies the routing key for the message.  The
                    routing key is used for routing messages depending on
                    the exchange configuration.

                mandatory: boolean

                    indicate mandatory routing

                    This flag tells the server how to react if the message
                    cannot be routed to a queue.  If this flag is True, the
                    server will return an unroutable message with a Return
                    method.  If this flag is False, the server silently
                    drops the message.

                    RULE:

                        The server SHOULD implement the mandatory flag.

                immediate: boolean

                    request immediate delivery

                    This flag tells the server how to react if the message
                    cannot be routed to a queue consumer immediately.  If
                    this flag is set, the server will return an
                    undeliverable message with a Return method. If this
                    flag is zero, the server will queue the message, but
                    with no guarantee that it will ever be consumed.

                    RULE:

                        The server SHOULD implement the immediate flag.

                timeout: short

                    timeout for publish

                    Set timeout to wait maximum timeout second
                    for message to publish.

                confirm_timeout: short

                    confirm_timeout for publish in confirm mode

                    When the channel is in confirm mode set
                    confirm_timeout to wait maximum confirm_timeout
                    second for message to confirm.

            """
            if not self.connection:
                raise RecoverableConnectionError(
                    'basic_publish: connection closed')

            capabilities = self.connection. \
                client_properties.get('capabilities', {})
            if capabilities.get('connection.blocked', False):
                try:
                    # Check if an event was sent, such as the out of memory message
                    self.connection.drain_events(timeout=0)
                except socket.timeout:
                    pass

            try:
                with self.connection.transport.having_timeout(timeout):
                    return self.send_method(
                        spec.Basic.Publish, argsig,
                        (0, exchange, routing_key, mandatory, immediate), msg
                    )
            except socket.timeout:
                raise RecoverableChannelError('basic_publish: timed out')

        basic_publish = _basic_publish

        def basic_publish_confirm(self, *args, **kwargs):
            confirm_timeout = kwargs.pop('confirm_timeout', None)

            def confirm_handler(method, *args):
                # When RMQ nacks message we are raising MessageNacked exception
                if method == spec.Basic.Nack:
                    raise MessageNacked()

            if not self._confirm_selected:
                self._confirm_selected = True
                self.confirm_select()
            ret = self._basic_publish(*args, **kwargs)
            # Waiting for confirmation of message.
            timeout = confirm_timeout or kwargs.get('timeout', None)
            self.wait([spec.Basic.Ack, spec.Basic.Nack],
                      callback=confirm_handler,
                      timeout=timeout)
            return ret

        def basic_qos(self, prefetch_size, prefetch_count, a_global,
                      argsig='lBb'):
            """Specify quality of service.

            This method requests a specific quality of service.  The QoS
            can be specified for the current channel or for all channels
            on the connection.  The particular properties and semantics of
            a qos method always depend on the content class semantics.
            Though the qos method could in principle apply to both peers,
            it is currently meaningful only for the server.

            PARAMETERS:
                prefetch_size: long

                    prefetch window in octets

                    The client can request that messages be sent in
                    advance so that when the client finishes processing a
                    message, the following message is already held
                    locally, rather than needing to be sent down the
                    channel.  Prefetching gives a performance improvement.
                    This field specifies the prefetch window size in
                    octets.  The server will send a message in advance if
                    it is equal to or smaller in size than the available
                    prefetch size (and also falls into other prefetch
                    limits). May be set to zero, meaning "no specific
                    limit", although other prefetch limits may still
                    apply. The prefetch-size is ignored if the no-ack
                    option is set.

                    RULE:

                        The server MUST ignore this setting when the
                        client is not processing any messages - i.e. the
                        prefetch size does not limit the transfer of
                        single messages to a client, only the sending in
                        advance of more messages while the client still
                        has one or more unacknowledged messages.

                prefetch_count: short

                    prefetch window in messages

                    Specifies a prefetch window in terms of whole
                    messages.  This field may be used in combination with
                    the prefetch-size field; a message will only be sent
                    in advance if both prefetch windows (and those at the
                    channel and connection level) allow it. The prefetch-
                    count is ignored if the no-ack option is set.

                    RULE:

                        The server MAY send less data in advance than
                        allowed by the client's specified prefetch windows
                        but it MUST NOT send more.

                a_global: boolean

                    Defines a scope of QoS. Semantics of this parameter differs
                    between AMQP 0-9-1 standard and RabbitMQ broker:

                    MEANING IN AMQP 0-9-1:
                        False:  shared across all consumers on the channel
                        True:   shared across all consumers on the connection
                    MEANING IN RABBITMQ:
                        False:  applied separately to each new consumer
                                on the channel
                        True:   shared across all consumers on the channel
            """
            return self.send_method(
                spec.Basic.Qos, argsig, (prefetch_size, prefetch_count, a_global),
                wait=spec.Basic.QosOk,
            )

        def basic_recover(self, requeue=False):
            """Redeliver unacknowledged messages.

            This method asks the broker to redeliver all unacknowledged
            messages on a specified channel. Zero or more messages may be
            redelivered.  This method is only allowed on non-transacted
            channels.

            RULE:

                The server MUST set the redelivered flag on all messages
                that are resent.

            RULE:

                The server MUST raise a channel exception if this is
                called on a transacted channel.

            PARAMETERS:
                requeue: boolean

                    requeue the message

                    If this field is False, the message will be redelivered
                    to the original recipient.  If this field is True, the
                    server will attempt to requeue the message,
                    potentially then delivering it to an alternative
                    subscriber.
            """
            return self.send_method(spec.Basic.Recover, 'b', (requeue,))

        def basic_recover_async(self, requeue=False):
            return self.send_method(spec.Basic.RecoverAsync, 'b', (requeue,))

        def basic_reject(self, delivery_tag, requeue, argsig='Lb'):
            """Reject an incoming message.

            This method allows a client to reject a message.  It can be
            used to interrupt and cancel large incoming messages, or
            return untreatable messages to their original queue.

            RULE:

                The server SHOULD be capable of accepting and process the
                Reject method while sending message content with a Deliver
                or Get-Ok method.  I.e. the server should read and process
                incoming methods while sending output frames.  To cancel a
                partially-send content, the server sends a content body
                frame of size 1 (i.e. with no data except the frame-end
                octet).

            RULE:

                The server SHOULD interpret this method as meaning that
                the client is unable to process the message at this time.

            RULE:

                A client MUST NOT use this method as a means of selecting
                messages to process.  A rejected message MAY be discarded
                or dead-lettered, not necessarily passed to another
                client.

            PARAMETERS:
                delivery_tag: longlong

                    server-assigned delivery tag

                    The server-assigned and channel-specific delivery tag

                    RULE:

                        The delivery tag is valid only within the channel
                        from which the message was received.  I.e. a client
                        MUST NOT receive a message on one channel and then
                        acknowledge it on another.

                    RULE:

                        The server MUST NOT use a zero value for delivery
                        tags.  Zero is reserved for client use, meaning "all
                        messages so far received".

                requeue: boolean

                    requeue the message

                    If this field is False, the message will be discarded.
                    If this field is True, the server will attempt to
                    requeue the message.

                    RULE:

                        The server MUST NOT deliver the message to the
                        same client within the context of the current
                        channel.  The recommended strategy is to attempt
                        to deliver the message to an alternative consumer,
                        and if that is not possible, to move the message
                        to a dead-letter queue.  The server MAY use more
                        sophisticated tracking to hold the message on the
                        queue and redeliver it to the same client at a
                        later stage.
            """
            return self.send_method(
                spec.Basic.Reject, argsig, (delivery_tag, requeue),
            )

        def _on_basic_return(self, reply_code, reply_text,
                             exchange, routing_key, message):
            """Return a failed message.

            This method returns an undeliverable message that was
            published with the "immediate" flag set, or an unroutable
            message published with the "mandatory" flag set. The reply
            code and text provide information about the reason that the
            message was undeliverable.

            PARAMETERS:
                reply_code: short

                    The reply code. The AMQ reply codes are defined in AMQ
                    RFC 011.

                reply_text: shortstr

                    The localised reply text.  This text can be logged as an
                    aid to resolving issues.

                exchange: shortstr

                    Specifies the name of the exchange that the message
                    was originally published to.

                routing_key: shortstr

                    Message routing key

                    Specifies the routing key name specified when the
                    message was published.
            """
            exc = error_for_code(
                reply_code, reply_text, spec.Basic.Return, ChannelError,
            )
            handlers = self.events.get('basic_return')
            if not handlers:
                raise exc
            for callback in handlers:
                callback(exc, exchange, routing_key, message)

        #############
        #
        #  Tx
        #
        #
        # work with standard transactions
        #
        # Standard transactions provide so-called "1.5 phase commit".  We
        # can ensure that work is never lost, but there is a chance of
        # confirmations being lost, so that messages may be resent.
        # Applications that use standard transactions must be able to
        # detect and ignore duplicate messages.
        #
        # GRAMMAR::
        #
        #     tx                  = C:SELECT S:SELECT-OK
        #                         / C:COMMIT S:COMMIT-OK
        #                         / C:ROLLBACK S:ROLLBACK-OK
        #
        # RULE:
        #
        #     An client using standard transactions SHOULD be able to
        #     track all messages received within a reasonable period, and
        #     thus detect and reject duplicates of the same message. It
        #     SHOULD NOT pass these to the application layer.
        #
        #

        def tx_commit(self):
            """Commit the current transaction.

            This method commits all messages published and acknowledged in
            the current transaction.  A new transaction starts immediately
            after a commit.
            """
            return self.send_method(spec.Tx.Commit, wait=spec.Tx.CommitOk)

        def tx_rollback(self):
            """Abandon the current transaction.

            This method abandons all messages published and acknowledged
            in the current transaction.  A new transaction starts
            immediately after a rollback.
            """
            return self.send_method(spec.Tx.Rollback, wait=spec.Tx.RollbackOk)

        def tx_select(self):
            """Select standard transaction mode.

            This method sets the channel to use standard transactions.
            The client must use this method at least once on a channel
            before using the Commit or Rollback methods.
            """
            return self.send_method(spec.Tx.Select, wait=spec.Tx.SelectOk)

        def confirm_select(self, nowait=False):
            """Enable publisher confirms for this channel.

            Note: This is an RabbitMQ extension.

            Can now be used if the channel is in transactional mode.

            :param nowait:
                If set, the server will not respond to the method.
                The client should not wait for a reply method. If the
                server could not complete the method it will raise a channel
                or connection exception.
            """
            return self.send_method(
                spec.Confirm.Select, 'b', (nowait,),
                wait=None if nowait else spec.Confirm.SelectOk,
            )

        def _on_basic_ack(self, delivery_tag, multiple):
            for callback in self.events['basic_ack']:
                callback(delivery_tag, multiple)

        def _on_basic_nack(self, delivery_tag, multiple):
            for callback in self.events['basic_nack']:
                callback(delivery_tag, multiple)