"""Using Pika with a Twisted reactor.

The interfaces in this module are Deferred-based when possible. This means that
the connection.channel() method and most of the channel methods return
Deferreds instead of taking a callback argument and that basic_consume()
returns a Twisted DeferredQueue where messages from the server will be
stored. Refer to the docstrings for TwistedProtocolConnection.channel() and the
TwistedChannel class for details.

"""

import functools
import logging
from collections import namedtuple

from twisted.internet import (defer, error as twisted_error, reactor, protocol)
from twisted.python.failure import Failure

import pika.connection
from pika import exceptions, spec
from pika.adapters.utils import nbio_interface
from pika.adapters.utils.io_services_utils import check_callback_arg
from pika.exchange_type import ExchangeType

# Twistisms
# pylint: disable=C0111,C0103
# Other
# pylint: disable=too-many-lines

LOGGER = logging.getLogger(__name__)


class ClosableDeferredQueue(defer.DeferredQueue):
    """
    Like the normal Twisted DeferredQueue, but after close() is called with an
    exception instance all pending Deferreds are errbacked and further attempts
    to call get() or put() return a Failure wrapping that exception.
    """

    def __init__(self, size=None, backlog=None):
        self.closed = None
        super(ClosableDeferredQueue, self).__init__(size, backlog)

    def put(self, obj):
        """
        Like the original :meth:`DeferredQueue.put` method, but returns an
        errback if the queue is closed.

        """
        if self.closed:
            LOGGER.error('Impossible to put to the queue, it is closed.')
            return defer.fail(self.closed)
        return defer.DeferredQueue.put(self, obj)

    def get(self):
        """
        Returns a Deferred that will fire with the next item in the queue, when
        it's available.

        The Deferred will errback if the queue is closed.

        :returns: Deferred that fires with the next item.
        :rtype: Deferred

        """
        if self.closed:
            LOGGER.error('Impossible to get from the queue, it is closed.')
            return defer.fail(self.closed)
        return defer.DeferredQueue.get(self)

    def close(self, reason):
        """Closes the queue.

        Errback the pending calls to :meth:`get()`.

        """
        if self.closed:
            LOGGER.warning('Queue was already closed with reason: %s.',
                           self.closed)
        self.closed = reason
        while self.waiting:
            self.waiting.pop().errback(reason)
        self.pending = []


ReceivedMessage = namedtuple("ReceivedMessage",
                             ["channel", "method", "properties", "body"])


class TwistedChannel(object):
    """A wrapper around Pika's Channel.

    Channel methods that normally take a callback argument are wrapped to
    return a Deferred that fires with whatever would be passed to the callback.
    If the channel gets closed, all pending Deferreds are errbacked with a
    ChannelClosed exception. The returned Deferreds fire with whatever
    arguments the callback to the original method would receive.

    Some methods like basic_consume and basic_get are wrapped in a special way,
    see their docstrings for details.
    """

    def __init__(self, channel):
        self._channel = channel
        self._closed = None
        self._calls = set()
        self._consumers = {}
        # Store Basic.Get calls so we can handle GetEmpty replies
        self._basic_get_deferred = None
        self._channel.add_callback(self._on_getempty, [spec.Basic.GetEmpty],
                                   False)
        # We need this mapping to close the ClosableDeferredQueue when a queue
        # is deleted.
        self._queue_name_to_consumer_tags = {}
        # Whether RabbitMQ delivery confirmation has been enabled
        self._delivery_confirmation = False
        self._delivery_message_id = None
        self._deliveries = {}
        # Holds a ReceivedMessage object representing a message received via
        # Basic.Return in publisher-acknowledgments mode.
        self._puback_return = None

        self.on_closed = defer.Deferred()
        self._channel.add_on_close_callback(self._on_channel_closed)
        self._channel.add_on_cancel_callback(
            self._on_consumer_cancelled_by_broker)

    def __repr__(self):
        return '<{cls} channel={chan!r}>'.format(
            cls=self.__class__.__name__, chan=self._channel)

    def _on_channel_closed(self, _channel, reason):
        # enter the closed state
        self._closed = reason
        # errback all pending calls
        for d in self._calls:
            d.errback(self._closed)
        # errback all pending deliveries
        for d in self._deliveries.values():
            d.errback(self._closed)
        # close all open queues
        for consumer in self._consumers.values():
            consumer.close(self._closed)
        # release references to stored objects
        self._calls = set()
        self._deliveries = {}
        self._consumers = {}
        self.on_closed.callback(self._closed)

    def _on_consumer_cancelled_by_broker(self, method_frame):
        """Called by impl when broker cancels consumer via Basic.Cancel.

        This is a RabbitMQ-specific feature. The circumstances include deletion
        of queue being consumed as well as failure of a HA node responsible for
        the queue being consumed.

        :param pika.frame.Method method_frame: method frame with the
            `spec.Basic.Cancel` method

        """
        return self._on_consumer_cancelled(method_frame)

    def _on_consumer_cancelled(self, frame):
        """Called when the broker cancels a consumer via Basic.Cancel or when
        the broker responds to a Basic.Cancel request by Basic.CancelOk.

        :param pika.frame.Method frame: method frame with the
            `spec.Basic.Cancel` or `spec.Basic.CancelOk` method

        """
        consumer_tag = frame.method.consumer_tag
        if consumer_tag not in self._consumers:
            # Could be cancelled by user or broker earlier
            LOGGER.warning('basic_cancel - consumer not found: %s',
                           consumer_tag)
            return frame
        self._consumers[consumer_tag].close(exceptions.ConsumerCancelled())
        del self._consumers[consumer_tag]
        # Remove from the queue-to-ctags index:
        for ctags in self._queue_name_to_consumer_tags.values():
            try:
                ctags.remove(consumer_tag)
            except KeyError:
                continue
        return frame

    def _on_getempty(self, _method_frame):
        """Callback the Basic.Get deferred with None.
        """
        if self._basic_get_deferred is None:
            LOGGER.warning("Got Basic.GetEmpty but no Basic.Get calls "
                           "were pending.")
            return
        self._basic_get_deferred.callback(None)

    def _wrap_channel_method(self, name):
        """Wrap Pika's Channel method to make it return a Deferred that fires
        when the method completes and errbacks if the channel gets closed. If
        the original method's callback would receive more than one argument,
        the Deferred fires with a tuple of argument values.

        """
        method = getattr(self._channel, name)

        @functools.wraps(method)
        def wrapped(*args, **kwargs):
            if self._closed:
                return defer.fail(self._closed)

            d = defer.Deferred()
            self._calls.add(d)
            d.addCallback(self._clear_call, d)

            def single_argument(*args):
                """
                Make sure that the deferred is called with a single argument.
                In case the original callback fires with more than one, convert
                to a tuple.
                """
                if len(args) > 1:
                    d.callback(tuple(args))
                else:
                    d.callback(*args)

            kwargs['callback'] = single_argument

            try:
                method(*args, **kwargs)
            except Exception:  # pylint: disable=W0703
                return defer.fail()
            return d

        return wrapped

    def _clear_call(self, ret, d):
        self._calls.discard(d)
        return ret

    # Public Channel attributes

    @property
    def channel_number(self):
        return self._channel.channel_number

    @property
    def connection(self):
        return self._channel.connection

    @property
    def is_closed(self):
        """Returns True if the channel is closed.

        :rtype: bool

        """
        return self._channel.is_closed

    @property
    def is_closing(self):
        """Returns True if client-initiated closing of the channel is in
        progress.

        :rtype: bool

        """
        return self._channel.is_closing

    @property
    def is_open(self):
        """Returns True if the channel is open.

        :rtype: bool

        """
        return self._channel.is_open

    @property
    def flow_active(self):
        return self._channel.flow_active

    @property
    def consumer_tags(self):
        return self._channel.consumer_tags

    # Deferred-equivalents of public Channel methods

    def callback_deferred(self, deferred, replies):
        """Pass in a Deferred and a list replies from the RabbitMQ broker which
        you'd like the Deferred to be callbacked on with the frame as callback
        value.

        :param Deferred deferred: The Deferred to callback
        :param list replies: The replies to callback on

        """
        self._channel.add_callback(deferred.callback, replies)

    # Public Channel methods

    def add_on_return_callback(self, callback):
        """Pass a callback function that will be called when a published
        message is rejected and returned by the server via `Basic.Return`.

        :param callable callback: The method to call on callback with the
            message as only argument. The message is a named tuple with
            the following attributes
            - channel: this TwistedChannel
            - method: pika.spec.Basic.Return
            - properties: pika.spec.BasicProperties
            - body: bytes
        """
        self._channel.add_on_return_callback(
            lambda _channel, method, properties, body: callback(
                ReceivedMessage(
                    channel=self,
                    method=method,
                    properties=properties,
                    body=body,
                )
            )
        )

    def basic_ack(self, delivery_tag=0, multiple=False):
        """Acknowledge one or more messages. When sent by the client, this
        method acknowledges one or more messages delivered via the Deliver or
        Get-Ok methods. When sent by server, this method acknowledges one or
        more messages published with the Publish method on a channel in
        confirm mode. The acknowledgement can be for a single message or a
        set of messages up to and including a specific message.

        :param integer delivery_tag: int/long The server-assigned delivery tag
        :param bool multiple: If set to True, the delivery tag is treated as
                              "up to and including", so that multiple messages
                              can be acknowledged with a single method. If set
                              to False, the delivery tag refers to a single
                              message. If the multiple field is 1, and the
                              delivery tag is zero, this indicates
                              acknowledgement of all outstanding messages.

        """
        return self._channel.basic_ack(
            delivery_tag=delivery_tag, multiple=multiple)

    def basic_cancel(self, consumer_tag=''):
        """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. It may also be sent from the server to the client in
        the event of the consumer being unexpectedly cancelled (i.e. cancelled
        for any reason other than the server receiving the corresponding
        basic.cancel from the client). This allows clients to be notified of
        the loss of consumers due to events such as queue deletion.

        This method wraps :meth:`Channel.basic_cancel
        <pika.channel.Channel.basic_cancel>` and closes any deferred queue
        associated with that consumer.

        :param str consumer_tag: Identifier for the consumer
        :returns: Deferred that fires on the Basic.CancelOk response
        :rtype: Deferred
        :raises ValueError:

        """
        wrapped = self._wrap_channel_method('basic_cancel')
        d = wrapped(consumer_tag=consumer_tag)
        return d.addCallback(self._on_consumer_cancelled)

    def basic_consume(self,
                      queue,
                      auto_ack=False,
                      exclusive=False,
                      consumer_tag=None,
                      arguments=None):
        """Consume from a server queue.

        Sends the AMQP 0-9-1 command Basic.Consume to the broker and binds
        messages for the consumer_tag to a
        :class:`ClosableDeferredQueue`. If you do not pass in a
        consumer_tag, one will be automatically generated for you.

        For more information on basic_consume, see:
        Tutorial 2 at http://www.rabbitmq.com/getstarted.html
        http://www.rabbitmq.com/confirms.html
        http://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.consume

        :param str queue: The queue to consume from. Use the empty string to
            specify the most recent server-named queue for this channel.
        :param bool auto_ack: if set to True, automatic acknowledgement mode
            will be used (see http://www.rabbitmq.com/confirms.html). This
            corresponds with the 'no_ack' parameter in the basic.consume AMQP
            0.9.1 method
        :param bool exclusive: Don't allow other consumers on the queue
        :param str consumer_tag: Specify your own consumer tag
        :param dict arguments: Custom key/value pair arguments for the consumer
        :returns: Deferred that fires with a tuple
            ``(queue_object, consumer_tag)``. The Deferred will errback with an
            instance of :class:`exceptions.ChannelClosed` if the call fails.
            The queue object is an instance of :class:`ClosableDeferredQueue`,
            where data received from the queue will be stored. Clients should
            use its :meth:`get() <ClosableDeferredQueue.get>` method to fetch
            an individual message, which will return a Deferred firing with a
            namedtuple whose attributes are:
            - channel: this TwistedChannel
            - method: pika.spec.Basic.Deliver
            - properties: pika.spec.BasicProperties
            - body: bytes
        :rtype: Deferred

        """
        if self._closed:
            return defer.fail(self._closed)

        queue_obj = ClosableDeferredQueue()
        d = defer.Deferred()
        self._calls.add(d)

        def on_consume_ok(frame):
            consumer_tag = frame.method.consumer_tag
            self._queue_name_to_consumer_tags.setdefault(
                queue, set()).add(consumer_tag)
            self._consumers[consumer_tag] = queue_obj
            self._calls.discard(d)
            d.callback((queue_obj, consumer_tag))

        def on_message_callback(_channel, method, properties, body):
            """Add the ReceivedMessage to the queue, while replacing the
            channel implementation.
            """
            queue_obj.put(
                ReceivedMessage(
                    channel=self,
                    method=method,
                    properties=properties,
                    body=body,
                ))

        try:
            self._channel.basic_consume(
                queue=queue,
                on_message_callback=on_message_callback,
                auto_ack=auto_ack,
                exclusive=exclusive,
                consumer_tag=consumer_tag,
                arguments=arguments,
                callback=on_consume_ok,
            )
        except Exception:  # pylint: disable=W0703
            return defer.fail()

        return d

    def basic_get(self, queue, auto_ack=False):
        """Get a single message from the AMQP broker.

        Will return If the queue is empty, it will return None.
        If you want to
        be notified of Basic.GetEmpty, use the Channel.add_callback method
        adding your Basic.GetEmpty callback which should expect only one
        parameter, frame. Due to implementation details, this cannot be called
        a second time until the callback is executed.  For more information on
        basic_get and its parameters, see:

        http://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.get

        This method wraps :meth:`Channel.basic_get
        <pika.channel.Channel.basic_get>`.

        :param str queue: The queue from which to get a message. Use the empty
                      string to specify the most recent server-named queue
                      for this channel.
        :param bool auto_ack: Tell the broker to not expect a reply
        :returns: Deferred that fires with a namedtuple whose attributes are:
             - channel: this TwistedChannel
             - method: pika.spec.Basic.GetOk
             - properties: pika.spec.BasicProperties
             - body: bytes
            If the queue is empty, None will be returned.
        :rtype: Deferred
        :raises pika.exceptions.DuplicateGetOkCallback:

        """
        if self._basic_get_deferred is not None:
            raise exceptions.DuplicateGetOkCallback()

        def create_namedtuple(result):
            if result is None:
                return None
            _channel, method, properties, body = result
            return ReceivedMessage(
                channel=self,
                method=method,
                properties=properties,
                body=body,
            )

        def cleanup_attribute(result):
            self._basic_get_deferred = None
            return result

        d = self._wrap_channel_method("basic_get")(
            queue=queue, auto_ack=auto_ack)
        d.addCallback(create_namedtuple)
        d.addBoth(cleanup_attribute)
        self._basic_get_deferred = d
        return d

    def basic_nack(self, delivery_tag=None, multiple=False, requeue=True):
        """This method allows a client to reject one or more incoming messages.
        It can be used to interrupt and cancel large incoming messages, or
        return untreatable messages to their original queue.

        :param integer delivery_tag: int/long The server-assigned delivery tag
        :param bool multiple: If set to True, the delivery tag is treated as
                              "up to and including", so that multiple messages
                              can be acknowledged with a single method. If set
                              to False, the delivery tag refers to a single
                              message. If the multiple field is 1, and the
                              delivery tag is zero, this indicates
                              acknowledgement of all outstanding messages.
        :param bool requeue: If requeue is true, the server will attempt to
                             requeue the message. If requeue is false or the
                             requeue attempt fails the messages are discarded
                             or dead-lettered.

        """
        return self._channel.basic_nack(
            delivery_tag=delivery_tag,
            multiple=multiple,
            requeue=requeue,
        )

    def basic_publish(self,
                      exchange,
                      routing_key,
                      body,
                      properties=None,
                      mandatory=False):
        """Publish to the channel with the given exchange, routing key and body.

        This method wraps :meth:`Channel.basic_publish
        <pika.channel.Channel.basic_publish>`, but makes sure the channel is
        not closed before publishing.

        For more information on basic_publish and what the parameters do, see:

        http://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.publish

        :param str exchange: The exchange to publish to
        :param str routing_key: The routing key to bind on
        :param bytes body: The message body
        :param pika.spec.BasicProperties properties: Basic.properties
        :param bool mandatory: The mandatory flag
        :returns: A Deferred that fires with the result of the channel's
            basic_publish.
        :rtype: Deferred
        :raises UnroutableError: raised when a message published in
            publisher-acknowledgments mode (see
            `BlockingChannel.confirm_delivery`) is returned via `Basic.Return`
            followed by `Basic.Ack`.
        :raises NackError: raised when a message published in
            publisher-acknowledgements mode is Nack'ed by the broker. See
            `BlockingChannel.confirm_delivery`.

        """
        if self._closed:
            return defer.fail(self._closed)
        result = self._channel.basic_publish(
            exchange=exchange,
            routing_key=routing_key,
            body=body,
            properties=properties,
            mandatory=mandatory)
        if not self._delivery_confirmation:
            return defer.succeed(result)
        else:
            # See http://www.rabbitmq.com/confirms.html#publisher-confirms
            self._delivery_message_id += 1
            self._deliveries[self._delivery_message_id] = defer.Deferred()
            return self._deliveries[self._delivery_message_id]

    def basic_qos(self, prefetch_size=0, prefetch_count=0, global_qos=False):
        """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 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.

        :param int prefetch_size:  This field specifies the prefetch window
                                   size. 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 by
                                   consumers who have enabled the no-ack
                                   option.
        :param int prefetch_count: 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 by consumers who
                                   have enabled the no-ack option.
        :param bool global_qos:    Should the QoS apply to all channels on the
                                   connection.
        :returns: Deferred that fires on the Basic.QosOk response
        :rtype: Deferred

        """
        return self._wrap_channel_method("basic_qos")(
            prefetch_size=prefetch_size,
            prefetch_count=prefetch_count,
            global_qos=global_qos,
        )

    def basic_reject(self, delivery_tag, requeue=True):
        """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.

        :param integer delivery_tag: int/long The server-assigned delivery tag
        :param bool requeue: If requeue is true, the server will attempt to
                             requeue the message. If requeue is false or the
                             requeue attempt fails the messages are discarded
                             or dead-lettered.
        :raises: TypeError

        """
        return self._channel.basic_reject(
            delivery_tag=delivery_tag, requeue=requeue)

    def basic_recover(self, requeue=False):
        """This method asks the server to redeliver all unacknowledged messages
        on a specified channel. Zero or more messages may be redelivered. This
        method replaces the asynchronous Recover.

        :param bool requeue: If False, the message will be redelivered to the
                             original recipient. If True, the server will
                             attempt to requeue the message, potentially then
                             delivering it to an alternative subscriber.
        :returns: Deferred that fires on the Basic.RecoverOk response
        :rtype: Deferred

        """
        return self._wrap_channel_method("basic_recover")(requeue=requeue)

    def close(self, reply_code=0, reply_text="Normal shutdown"):
        """Invoke a graceful shutdown of the channel with the AMQP Broker.

        If channel is OPENING, transition to CLOSING and suppress the incoming
        Channel.OpenOk, if any.

        :param int reply_code: The reason code to send to broker
        :param str reply_text: The reason text to send to broker

        :raises ChannelWrongStateError: if channel is closed or closing

        """
        return self._channel.close(reply_code=reply_code, reply_text=reply_text)

    def confirm_delivery(self):
        """Turn on Confirm mode in the channel. Pass in a callback to be
        notified by the Broker when a message has been confirmed as received or
        rejected (Basic.Ack, Basic.Nack) from the broker to the publisher.

        For more information see:
            http://www.rabbitmq.com/confirms.html#publisher-confirms

        :returns: Deferred that fires on the Confirm.SelectOk response
        :rtype: Deferred

        """
        if self._delivery_confirmation:
            LOGGER.error('confirm_delivery: confirmation was already enabled.')
            return defer.succeed(None)
        wrapped = self._wrap_channel_method('confirm_delivery')
        d = wrapped(ack_nack_callback=self._on_delivery_confirmation)

        def set_delivery_confirmation(result):
            self._delivery_confirmation = True
            self._delivery_message_id = 0
            LOGGER.debug("Delivery confirmation enabled.")
            return result

        d.addCallback(set_delivery_confirmation)
        # Unroutable messages returned after this point will be in the context
        # of publisher acknowledgments
        self._channel.add_on_return_callback(self._on_puback_message_returned)
        return d

    def _on_delivery_confirmation(self, method_frame):
        """Invoked by pika when RabbitMQ responds to a Basic.Publish RPC
        command, passing in either a Basic.Ack or Basic.Nack frame with
        the delivery tag of the message that was published. The delivery tag
        is an integer counter indicating the message number that was sent
        on the channel via Basic.Publish. Here we're just doing house keeping
        to keep track of stats and remove message numbers that we expect
        a delivery confirmation of from the list used to keep track of messages
        that are pending confirmation.

        :param pika.frame.Method method_frame: Basic.Ack or Basic.Nack frame

        """
        delivery_tag = method_frame.method.delivery_tag
        if delivery_tag not in self._deliveries:
            LOGGER.error("Delivery tag %s not found in the pending deliveries",
                         delivery_tag)
            return
        if method_frame.method.multiple:
            tags = [tag for tag in self._deliveries if tag <= delivery_tag]
            tags.sort()
        else:
            tags = [delivery_tag]
        for tag in tags:
            d = self._deliveries[tag]
            del self._deliveries[tag]
            if isinstance(method_frame.method, pika.spec.Basic.Nack):
                # Broker was unable to process the message due to internal
                # error
                LOGGER.warning(
                    "Message was Nack'ed by broker: nack=%r; channel=%s;",
                    method_frame.method, self.channel_number)
                if self._puback_return is not None:
                    returned_messages = [self._puback_return]
                    self._puback_return = None
                else:
                    returned_messages = []
                d.errback(exceptions.NackError(returned_messages))
            else:
                assert isinstance(method_frame.method, pika.spec.Basic.Ack)
                if self._puback_return is not None:
                    # Unroutable message was returned
                    returned_messages = [self._puback_return]
                    self._puback_return = None
                    d.errback(exceptions.UnroutableError(returned_messages))
                else:
                    d.callback(method_frame.method)

    def _on_puback_message_returned(self, channel, method, properties, body):
        """Called as the result of Basic.Return from broker in
        publisher-acknowledgements mode. Saves the info as a ReturnedMessage
        instance in self._puback_return.

        :param pika.Channel channel: our self._impl channel
        :param pika.spec.Basic.Return method:
        :param pika.spec.BasicProperties properties: message properties
        :param bytes body: returned message body; empty string if no body

        """
        assert isinstance(method, spec.Basic.Return), method
        assert isinstance(properties, spec.BasicProperties), properties

        LOGGER.warning(
            "Published message was returned: _delivery_confirmation=%s; "
            "channel=%s; method=%r; properties=%r; body_size=%d; "
            "body_prefix=%.255r", self._delivery_confirmation,
            channel.channel_number, method, properties,
            len(body) if body is not None else None, body)

        self._puback_return = ReceivedMessage(channel=self,
                method=method, properties=properties, body=body)

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

        :param str destination: The destination exchange to bind
        :param str source: The source exchange to bind to
        :param str routing_key: The routing key to bind on
        :param dict arguments: Custom key/value pair arguments for the binding
        :raises ValueError:
        :returns: Deferred that fires on the Exchange.BindOk response
        :rtype: Deferred

        """
        return self._wrap_channel_method("exchange_bind")(
            destination=destination,
            source=source,
            routing_key=routing_key,
            arguments=arguments,
        )

    def exchange_declare(self,
                         exchange,
                         exchange_type=ExchangeType.direct,
                         passive=False,
                         durable=False,
                         auto_delete=False,
                         internal=False,
                         arguments=None):
        """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.

        If passive set, the server will reply with Declare-Ok if the exchange
        already exists with the same name, and raise an error if not and if the
        exchange does not already exist, the server MUST raise a channel
        exception with reply code 404 (not found).

        :param str exchange: The exchange name consists of a non-empty sequence
            of these characters: letters, digits, hyphen, underscore, period,
            or colon
        :param str exchange_type: The exchange type to use
        :param bool passive: Perform a declare or just check to see if it
            exists
        :param bool durable: Survive a reboot of RabbitMQ
        :param bool auto_delete: Remove when no more queues are bound to it
        :param bool internal: Can only be published to by other exchanges
        :param dict arguments: Custom key/value pair arguments for the exchange
        :returns: Deferred that fires on the Exchange.DeclareOk response
        :rtype: Deferred
        :raises ValueError:

        """
        return self._wrap_channel_method("exchange_declare")(
            exchange=exchange,
            exchange_type=exchange_type,
            passive=passive,
            durable=durable,
            auto_delete=auto_delete,
            internal=internal,
            arguments=arguments,
        )

    def exchange_delete(self, exchange=None, if_unused=False):
        """Delete the exchange.

        :param str exchange: The exchange name
        :param bool if_unused: only delete if the exchange is unused
        :returns: Deferred that fires on the Exchange.DeleteOk response
        :rtype: Deferred
        :raises ValueError:

        """
        return self._wrap_channel_method("exchange_delete")(
            exchange=exchange,
            if_unused=if_unused,
        )

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

        :param str destination: The destination exchange to unbind
        :param str source: The source exchange to unbind from
        :param str routing_key: The routing key to unbind
        :param dict arguments: Custom key/value pair arguments for the binding
        :returns: Deferred that fires on the Exchange.UnbindOk response
        :rtype: Deferred
        :raises ValueError:

        """
        return self._wrap_channel_method("exchange_unbind")(
            destination=destination,
            source=source,
            routing_key=routing_key,
            arguments=arguments,
        )

    def flow(self, active):
        """Turn Channel flow control off and on.

        Returns a Deferred that will fire with a bool indicating the channel
        flow state. For more information, please reference:

        http://www.rabbitmq.com/amqp-0-9-1-reference.html#channel.flow

        :param bool active: Turn flow on or off
        :returns: Deferred that fires with the channel flow state
        :rtype: Deferred
        :raises ValueError:

        """
        return self._wrap_channel_method("flow")(active=active)

    def open(self):
        """Open the channel"""
        return self._channel.open()

    def queue_bind(self, queue, exchange, routing_key=None, arguments=None):
        """Bind the queue to the specified exchange

        :param str queue: The queue to bind to the exchange
        :param str exchange: The source exchange to bind to
        :param str routing_key: The routing key to bind on
        :param dict arguments: Custom key/value pair arguments for the binding
        :returns: Deferred that fires on the Queue.BindOk response
        :rtype: Deferred
        :raises ValueError:

        """
        return self._wrap_channel_method("queue_bind")(
            queue=queue,
            exchange=exchange,
            routing_key=routing_key,
            arguments=arguments,
        )

    def queue_declare(self,
                      queue,
                      passive=False,
                      durable=False,
                      exclusive=False,
                      auto_delete=False,
                      arguments=None):
        """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.

        Use an empty string as the queue name for the broker to auto-generate
        one

        :param str queue: The queue name; if empty string, the broker will
            create a unique queue name
        :param bool passive: Only check to see if the queue exists
        :param bool durable: Survive reboots of the broker
        :param bool exclusive: Only allow access by the current connection
        :param bool auto_delete: Delete after consumer cancels or disconnects
        :param dict arguments: Custom key/value arguments for the queue
        :returns: Deferred that fires on the Queue.DeclareOk response
        :rtype: Deferred
        :raises ValueError:

        """
        return self._wrap_channel_method("queue_declare")(
            queue=queue,
            passive=passive,
            durable=durable,
            exclusive=exclusive,
            auto_delete=auto_delete,
            arguments=arguments,
        )

    def queue_delete(self, queue, if_unused=False, if_empty=False):
        """Delete a queue from the broker.


        This method wraps :meth:`Channel.queue_delete
        <pika.channel.Channel.queue_delete>`, and removes the reference to the
        queue object after it gets deleted on the server.

        :param str queue: The queue to delete
        :param bool if_unused: only delete if it's unused
        :param bool if_empty: only delete if the queue is empty
        :returns: Deferred that fires on the Queue.DeleteOk response
        :rtype: Deferred
        :raises ValueError:

        """
        wrapped = self._wrap_channel_method('queue_delete')
        d = wrapped(queue=queue, if_unused=if_unused, if_empty=if_empty)

        def _clear_consumer(ret, queue_name):
            for consumer_tag in list(
                    self._queue_name_to_consumer_tags.get(queue_name, set())):
                self._consumers[consumer_tag].close(
                    exceptions.ConsumerCancelled(
                        "Queue %s was deleted." % queue_name))
                del self._consumers[consumer_tag]
                self._queue_name_to_consumer_tags[queue_name].remove(
                    consumer_tag)
            return ret

        return d.addCallback(_clear_consumer, queue)

    def queue_purge(self, queue):
        """Purge all of the messages from the specified queue

        :param str queue: The queue to purge
        :returns: Deferred that fires on the Queue.PurgeOk response
        :rtype: Deferred
        :raises ValueError:

        """
        return self._wrap_channel_method("queue_purge")(queue=queue)

    def queue_unbind(self,
                     queue,
                     exchange=None,
                     routing_key=None,
                     arguments=None):
        """Unbind a queue from an exchange.

        :param str queue: The queue to unbind from the exchange
        :param str exchange: The source exchange to bind from
        :param str routing_key: The routing key to unbind
        :param dict arguments: Custom key/value pair arguments for the binding
        :returns: Deferred that fires on the Queue.UnbindOk response
        :rtype: Deferred
        :raises ValueError:

        """
        return self._wrap_channel_method("queue_unbind")(
            queue=queue,
            exchange=exchange,
            routing_key=routing_key,
            arguments=arguments,
        )

    def tx_commit(self):
        """Commit a transaction.

        :returns: Deferred that fires on the Tx.CommitOk response
        :rtype: Deferred
        :raises ValueError:

        """
        return self._wrap_channel_method("tx_commit")()

    def tx_rollback(self):
        """Rollback a transaction.

        :returns: Deferred that fires on the Tx.RollbackOk response
        :rtype: Deferred
        :raises ValueError:

        """
        return self._wrap_channel_method("tx_rollback")()

    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.

        :returns: Deferred that fires on the Tx.SelectOk response
        :rtype: Deferred
        :raises ValueError:

        """
        return self._wrap_channel_method("tx_select")()


class _TwistedConnectionAdapter(pika.connection.Connection):
    """A Twisted-specific implementation of a Pika Connection.

    NOTE: since `base_connection.BaseConnection`'s primary responsibility is
    management of the transport, we use `pika.connection.Connection` directly
    as our base class because this adapter uses a different transport
    management strategy.

    """

    def __init__(self, parameters, on_open_callback, on_open_error_callback,
                 on_close_callback, custom_reactor):
        super(_TwistedConnectionAdapter, self).__init__(
            parameters=parameters,
            on_open_callback=on_open_callback,
            on_open_error_callback=on_open_error_callback,
            on_close_callback=on_close_callback,
            internal_connection_workflow=False)

        self._reactor = custom_reactor or reactor
        self._transport = None  # to be provided by `connection_made()`

    def _adapter_call_later(self, delay, callback):
        """Implement
        :py:meth:`pika.connection.Connection._adapter_call_later()`.

        """
        check_callback_arg(callback, 'callback')
        return _TimerHandle(self._reactor.callLater(delay, callback))

    def _adapter_remove_timeout(self, timeout_id):
        """Implement
        :py:meth:`pika.connection.Connection._adapter_remove_timeout()`.

        """
        timeout_id.cancel()

    def _adapter_add_callback_threadsafe(self, callback):
        """Implement
        :py:meth:`pika.connection.Connection._adapter_add_callback_threadsafe()`.

        """
        check_callback_arg(callback, 'callback')
        self._reactor.callFromThread(callback)

    def _adapter_connect_stream(self):
        """Implement pure virtual
        :py:ref:meth:`pika.connection.Connection._adapter_connect_stream()`
         method.

        NOTE: This should not be called due to our initialization of Connection
        via `internal_connection_workflow=False`
        """
        raise NotImplementedError

    def _adapter_disconnect_stream(self):
        """Implement pure virtual
        :py:ref:meth:`pika.connection.Connection._adapter_disconnect_stream()`
         method.

        """
        self._transport.loseConnection()

    def _adapter_emit_data(self, data):
        """Implement pure virtual
        :py:ref:meth:`pika.connection.Connection._adapter_emit_data()` method.

        """
        self._transport.write(data)

    def connection_made(self, transport):
        """Introduces transport to protocol after transport is connected.

        :param twisted.internet.interfaces.ITransport transport:
        :raises Exception: Exception-based exception on error

        """
        self._transport = transport
        # Let connection know that stream is available
        self._on_stream_connected()

    def connection_lost(self, error):
        """Called upon loss or closing of TCP connection.

        NOTE: `connection_made()` and `connection_lost()` are each called just
        once and in that order. All other callbacks are called between them.

        :param Failure: A Twisted Failure instance wrapping an exception.

        """
        self._transport = None
        error = error.value  # drop the Failure wrapper
        if isinstance(error, twisted_error.ConnectionDone):
            self._error = error
            error = None
        LOGGER.log(logging.DEBUG if error is None else logging.ERROR,
                   'connection_lost: %r', error)
        self._on_stream_terminated(error)

    def data_received(self, data):
        """Called to deliver incoming data from the server to the protocol.

        :param data: Non-empty data bytes.
        :raises Exception: Exception-based exception on error

        """
        self._on_data_available(data)


class TwistedProtocolConnection(protocol.Protocol):
    """A Pika-specific implementation of a Twisted Protocol. Allows using
    Twisted's non-blocking connectTCP/connectSSL methods for connecting to the
    server.

    TwistedProtocolConnection objects have a `ready` instance variable that's a
    Deferred which fires when the connection is ready to be used (the initial
    AMQP handshaking has been done). You *have* to wait for this Deferred to
    fire before requesting a channel.

    Once the connection is ready, you will be able to use the `closed` instance
    variable: a Deferred which fires when the connection is closed.

    Since it's Twisted handling connection establishing it does not accept
    connect callbacks, you have to implement that within Twisted. Also remember
    that the host, port and ssl values of the connection parameters are ignored
    because, yet again, it's Twisted who manages the connection.

    """

    def __init__(self, parameters=None, custom_reactor=None):
        self.ready = defer.Deferred()
        self.ready.addCallback(lambda _: self.connectionReady())
        self.closed = None
        self._impl = _TwistedConnectionAdapter(
            parameters=parameters,
            on_open_callback=self._on_connection_ready,
            on_open_error_callback=self._on_connection_failed,
            on_close_callback=self._on_connection_closed,
            custom_reactor=custom_reactor,
        )
        self._calls = set()

    def channel(self, channel_number=None):  # pylint: disable=W0221
        """Create a new channel with the next available channel number or pass
        in a channel number to use. Must be non-zero if you would like to
        specify but it is recommended that you let Pika manage the channel
        numbers.

        :param int channel_number: The channel number to use, defaults to the
                                   next available.
        :returns: a Deferred that fires with an instance of a wrapper around
            the Pika Channel class.
        :rtype: Deferred

        """
        d = defer.Deferred()
        self._impl.channel(channel_number, d.callback)
        self._calls.add(d)
        d.addCallback(self._clear_call, d)
        return d.addCallback(TwistedChannel)

    @property
    def is_open(self):
        # For compatibility with previous releases.
        return self._impl.is_open

    @property
    def is_closed(self):
        # For compatibility with previous releases.
        return self._impl.is_closed

    def close(self, reply_code=200, reply_text='Normal shutdown'):
        if not self._impl.is_closed:
            self._impl.close(reply_code, reply_text)
        return self.closed

    # IProtocol methods

    def dataReceived(self, data):
        # Pass the bytes to Pika for parsing
        self._impl.data_received(data)

    def connectionLost(self, reason=protocol.connectionDone):
        self._impl.connection_lost(reason)
        # Let the caller know there's been an error
        d, self.ready = self.ready, None
        if d:
            d.errback(reason)

    def makeConnection(self, transport):
        self._impl.connection_made(transport)
        protocol.Protocol.makeConnection(self, transport)

    # Our own methods

    def connectionReady(self):
        """This method will be called when the underlying connection is ready.
        """
        return self

    def _on_connection_ready(self, _connection):
        d, self.ready = self.ready, None
        if d:
            self.closed = defer.Deferred()
            d.callback(None)

    def _on_connection_failed(self, _connection, _error_message=None):
        d, self.ready = self.ready, None
        if d:
            attempts = self._impl.params.connection_attempts
            exc = exceptions.AMQPConnectionError(attempts)
            d.errback(exc)

    def _on_connection_closed(self, _connection, exception):
        # errback all pending calls
        for d in self._calls:
            d.errback(exception)
        self._calls = set()

        d, self.closed = self.closed, None
        if d:
            if isinstance(exception, Failure):
                # Calling `callback` with a Failure instance will trigger the
                # errback path.
                exception = exception.value
            d.callback(exception)

    def _clear_call(self, ret, d):
        self._calls.discard(d)
        return ret


class _TimerHandle(nbio_interface.AbstractTimerReference):
    """This module's adaptation of `nbio_interface.AbstractTimerReference`.

    """

    def __init__(self, handle):
        """

        :param twisted.internet.base.DelayedCall handle:
        """
        self._handle = handle

    def cancel(self):
        if self._handle is not None:
            try:
                self._handle.cancel()
            except (twisted_error.AlreadyCalled,
                    twisted_error.AlreadyCancelled):
                pass

            self._handle = None