blocking_connection.py

"""The blocking connection adapter module implements blocking semantics on top
of Pika's core AMQP driver. While most of the asynchronous expectations are
removed when using the blocking connection adapter, it attempts to remain true
to the asynchronous RPC nature of the AMQP protocol, supporting server sent
RPC commands.

The user facing classes in the module consist of the
:py:class:`~pika.adapters.blocking_connection.BlockingConnection`
and the :class:`~pika.adapters.blocking_connection.BlockingChannel`
classes.

"""
import os
import logging
import select
import socket
import time
import warnings
import errno
from functools import wraps

from pika import frame
from pika import callback
from pika import channel
from pika import exceptions
from pika import spec
from pika import utils
from pika.adapters import base_connection

if os.name == 'java':
    from select import cpython_compatible_select as select_function
else:
    from select import select as select_function

LOGGER = logging.getLogger(__name__)


def retry_on_eintr(f):
    @wraps(f)
    def inner(*args, **kwargs):
        while True:
            try:
                return f(*args, **kwargs)
            except select.error as e:
                if e[0] != errno.EINTR:
                    raise
    return inner


class ReadPoller(object):
    """A poller that will check to see if data is ready on the socket using
    very short timeouts to avoid having to read on the socket, speeding up the
    BlockingConnection._handle_read() method.

    """
    POLL_TIMEOUT = 10

    @retry_on_eintr
    def __init__(self, fd, poll_timeout=POLL_TIMEOUT):
        """Create a new instance of the ReadPoller which wraps poll and select
        to determine if the socket has data to read on it.

        :param int fd: The file descriptor for the socket
        :param float poll_timeout: How long to wait for events (milliseconds)

        """
        self.fd = fd
        self.poll_timeout = poll_timeout
        if hasattr(select, 'poll') and os.name != 'java':
            self.poller = select.poll()
            self.poll_events = select.POLLIN | select.POLLPRI
            self.poller.register(self.fd, self.poll_events)
        else:
            self.poller = None
            self.poll_timeout = float(poll_timeout) / 1000

    @retry_on_eintr
    def ready(self):
        """Check to see if the socket has data to read.

        :rtype: bool

        """
        if self.poller:
            events = self.poller.poll(self.poll_timeout)
            return bool(events)
        else:
            ready, unused_wri, unused_err = select_function([self.fd], [], [],
                                                            self.poll_timeout)
            return bool(ready)


class BlockingConnection(base_connection.BaseConnection):
    """The BlockingConnection creates a layer on top of Pika's asynchronous core
    providing methods that will block until their expected response has returned.
    Due to the asynchronous nature of the `Basic.Deliver` and `Basic.Return` calls
    from RabbitMQ to your application, you can still implement
    continuation-passing style asynchronous methods if you'd like to receive
    messages from RabbitMQ using
    :meth:`basic_consume <BlockingChannel.basic_consume>` or if you want  to be
    notified of a delivery failure when using
    :meth:`basic_publish <BlockingChannel.basic_publish>` .

    `Basic.Get` is a blocking call which will either return the Method Frame,
    Header Frame and Body of a message, or it will return a `Basic.GetEmpty`
    frame as the method frame.

    For more information about communicating with the blocking_connection
    adapter, be sure to check out the
    :class:`BlockingChannel <BlockingChannel>` class which implements the
    :class:`Channel <pika.channel.Channel>` based communication for the
    blocking_connection adapter.

    """
    WRITE_TO_READ_RATIO = 10
    DO_HANDSHAKE = True
    SLEEP_DURATION = 0.1
    SOCKET_CONNECT_TIMEOUT = 0.25
    SOCKET_TIMEOUT_THRESHOLD = 12
    SOCKET_TIMEOUT_CLOSE_THRESHOLD = 3
    SOCKET_TIMEOUT_MESSAGE = "Timeout exceeded, disconnected"

    def __init__(self, parameters=None):
        """Create a new instance of the Connection object.

        :param pika.connection.Parameters parameters: Connection parameters
        :raises: RuntimeError

        """
        super(BlockingConnection, self).__init__(parameters, None, False)

    def add_on_close_callback(self, callback_method_unused):
        """This is not supported in BlockingConnection. When a connection is
        closed in BlockingConnection, a pika.exceptions.ConnectionClosed
        exception will raised instead.

        :param method callback_method_unused: Unused
        :raises: NotImplementedError

        """
        raise NotImplementedError('Blocking connection will raise '
                                  'ConnectionClosed exception')

    def add_on_open_callback(self, callback_method_unused):
        """This method is not supported in BlockingConnection.

        :param method callback_method_unused: Unused
        :raises: NotImplementedError

        """
        raise NotImplementedError('Connection callbacks not supported in '
                                  'BlockingConnection')

    def add_on_open_error_callback(self, callback_method_unused,
                                   remove_default=False):
        """This method is not supported in BlockingConnection.

        A pika.exceptions.AMQPConnectionError will be raised instead.

        :param method callback_method_unused: Unused
        :raises: NotImplementedError

        """
        raise NotImplementedError('Connection callbacks not supported in '
                                  'BlockingConnection')

    def add_timeout(self, deadline, callback_method):
        """Add the callback_method to the IOLoop timer to fire after deadline
        seconds. Returns a handle to the timeout. Do not confuse with
        Tornado's timeout where you pass in the time you want to have your
        callback called. Only pass in the seconds until it's to be called.

        :param int deadline: The number of seconds to wait to call callback
        :param method callback_method: The callback method
        :rtype: str

        """

        value = {'deadline': time.time() + deadline,
                 'callback': callback_method}
        timeout_id = hash(frozenset(value.items()))
        self._timeouts[timeout_id] = value
        return timeout_id

    def channel(self, channel_number=None):
        """Create a new channel with the next available or specified channel #.

        :param int channel_number: Specify the channel number

        """
        self._channel_open = False
        if not channel_number:
            channel_number = self._next_channel_number()
        LOGGER.debug('Opening channel %i', channel_number)
        self._channels[channel_number] = BlockingChannel(self, channel_number)
        return self._channels[channel_number]

    def close(self, reply_code=200, reply_text='Normal shutdown'):
        """Disconnect from RabbitMQ. If there are any open channels, it will
        attempt to close them prior to fully disconnecting. Channels which
        have active consumers will attempt to send a Basic.Cancel to RabbitMQ
        to cleanly stop the delivery of messages prior to closing the channel.

        :param int reply_code: The code number for the close
        :param str reply_text: The text reason for the close

        """
        LOGGER.info("Closing connection (%s): %s", reply_code, reply_text)
        self._set_connection_state(self.CONNECTION_CLOSING)
        self._remove_connection_callbacks()
        if self._has_open_channels:
            self._close_channels(reply_code, reply_text)
        while self._has_open_channels:
            self.process_data_events()
        if self.socket:
            self._send_connection_close(reply_code, reply_text)
        while self.is_closing:
            self.process_data_events()
        if self.heartbeat:
            self.heartbeat.stop()
        self._remove_connection_callbacks()
        self._adapter_disconnect()

    def connect(self):
        """Invoke if trying to reconnect to a RabbitMQ server. Constructing the
        Connection object should connect on its own.

        """
        self._set_connection_state(self.CONNECTION_INIT)
        error = self._adapter_connect()
        if error:
            raise exceptions.AMQPConnectionError(error)

    def process_data_events(self):
        """Will make sure that data events are processed. Your app can
        block on this method.

        """
        try:
            if self._handle_read():
                self._socket_timeouts = 0
        except AttributeError:
            raise exceptions.ConnectionClosed()
        except socket.timeout:
            self._handle_timeout()
        self._flush_outbound()
        self.process_timeouts()

    def process_timeouts(self):
        """Process the self._timeouts event stack"""
        for timeout_id in self._timeouts.keys():
            if self._deadline_passed(timeout_id):
                self._call_timeout_method(self._timeouts.pop(timeout_id))

    def remove_timeout(self, timeout_id):
        """Remove the timeout from the IOLoop by the ID returned from
        add_timeout.

        :param str timeout_id: The id of the timeout to remove

        """
        if timeout_id in self._timeouts:
            del self._timeouts[timeout_id]

    def send_method(self, channel_number, method_frame, content=None):
        """Constructs a RPC method frame and then sends it to the broker.

        :param int channel_number: The channel number for the frame
        :param pika.object.Method method_frame: The method frame to send
        :param tuple content: If set, is a content frame, is tuple of
                              properties and body.

        """
        self._send_method(channel_number, method_frame, content)

    def sleep(self, duration):
        """A safer way to sleep than calling time.sleep() directly which will
        keep the adapter from ignoring frames sent from RabbitMQ. The
        connection will "sleep" or block the number of seconds specified in
        duration in small intervals.

        :param int duration: The time to sleep

        """
        deadline = time.time() + duration
        while time.time() < deadline:
            time.sleep(self.SLEEP_DURATION)
            self.process_data_events()

    def _adapter_connect(self):
        """Connect to the RabbitMQ broker

        :rtype: bool
        :raises: pika.Exceptions.AMQPConnectionError

        """
        # Remove the default behavior for connection errors
        self.callbacks.remove(0, self.ON_CONNECTION_ERROR)
        error = super(BlockingConnection, self)._adapter_connect()
        if error:
            raise exceptions.AMQPConnectionError(error)
        self.socket.settimeout(self.SOCKET_CONNECT_TIMEOUT)
        self._frames_written_without_read = 0
        self._socket_timeouts = 0
        self._timeouts = dict()
        self._read_poller = ReadPoller(self.socket.fileno())
        self._on_connected()
        while not self.is_open:
            self.process_data_events()
        self.socket.settimeout(self.params.socket_timeout)
        self._set_connection_state(self.CONNECTION_OPEN)

    def _adapter_disconnect(self):
        """Called if the connection is being requested to disconnect."""
        if self.socket:
            self.socket.close()
        self.socket = None
        self._check_state_on_disconnect()
        self._init_connection_state()

    def _call_timeout_method(self, timeout_value):
        """Execute the method that was scheduled to be called.

        :param dict timeout_value: The configuration for the timeout

        """
        LOGGER.debug('Invoking scheduled call of %s', timeout_value['callback'])
        timeout_value['callback']()

    def _deadline_passed(self, timeout_id):
        """Returns True if the deadline has passed for the specified timeout_id.

        :param str timeout_id: The id of the timeout to check
        :rtype: bool

        """
        if timeout_id not in self._timeouts.keys():
            return False
        return self._timeouts[timeout_id]['deadline'] <= time.time()

    def _handle_read(self):
        """If the ReadPoller says there is data to read, try adn read it in the
        _handle_read of the parent class. Once read, reset the counter that
        keeps track of how many frames have been written since the last read.

        """
        if self._read_poller.ready():
            super(BlockingConnection, self)._handle_read()
            self._frames_written_without_read = 0

    def _handle_timeout(self):
        """Invoked whenever the socket times out"""
        self._socket_timeouts += 1
        threshold = (self.SOCKET_TIMEOUT_THRESHOLD if not self.is_closing else
                     self.SOCKET_TIMEOUT_CLOSE_THRESHOLD)

        LOGGER.debug('Handling timeout %i with a threshold of %i',
                     self._socket_timeouts, threshold)
        if self.is_closing and self._socket_timeouts > threshold:
            if not self.is_closing:
                LOGGER.critical('Closing connection due to timeout')
            self._on_connection_closed(None, True)

    def _flush_outbound(self):
        """Flush the outbound socket buffer."""
        if self.outbound_buffer:
            try:
                if self._handle_write():
                    self._socket_timeouts = 0
            except socket.timeout:
                return self._handle_timeout()

    def _on_connection_closed(self, method_frame, from_adapter=False):
        """Called when the connection is closed remotely. The from_adapter value
        will be true if the connection adapter has been disconnected from
        the broker and the method was invoked directly instead of by receiving
        a Connection.Close frame.

        :param pika.frame.Method: The Connection.Close frame
        :param bool from_adapter: Called by the connection adapter
        :raises: pika.exceptions.ConnectionClosed

        """
        if self._is_connection_close_frame(method_frame):
            self.closing = (method_frame.method.reply_code,
                            method_frame.method.reply_text)
            LOGGER.warning("Disconnected from RabbitMQ at %s:%i (%s): %s",
                           self.params.host, self.params.port,
                           self.closing[0], self.closing[1])
        self._set_connection_state(self.CONNECTION_CLOSED)
        self._remove_connection_callbacks()
        if not from_adapter:
            self._adapter_disconnect()
        for channel in self._channels:
            self._channels[channel]._on_close(method_frame)
        self._remove_connection_callbacks()
        if self.closing[0] not in [0, 200]:
            raise exceptions.ConnectionClosed(*self.closing)

    def _send_frame(self, frame_value):
        """This appends the fully generated frame to send to the broker to the
        output buffer which will be then sent via the connection adapter.

        :param frame_value: The frame to write
        :type frame_value:  pika.frame.Frame|pika.frame.ProtocolHeader

        """
        super(BlockingConnection, self)._send_frame(frame_value)
        self._frames_written_without_read += 1
        if self._frames_written_without_read >= self.WRITE_TO_READ_RATIO:
            if not isinstance(frame_value, frame.Method):
                self._frames_written_without_read = 0
                self.process_data_events()


class BlockingChannel(channel.Channel):
    """The BlockingChannel implements blocking semantics for most things that
    one would use callback-passing-style for with the
    :py:class:`~pika.channel.Channel` class. In addition,
    the `BlockingChannel` class implements a :term:`generator` that allows you
    to :doc:`consume messages </examples/blocking_consumer_generator>` without
    using callbacks.

    Example of creating a BlockingChannel::

        import pika

        # Create our connection object
        connection = pika.BlockingConnection()

        # The returned object will be a blocking channel
        channel = connection.channel()

    :param BlockingConnection connection: The connection
    :param int channel_number: The channel number for this instance

    """
    NO_RESPONSE_FRAMES = ['Basic.Ack', 'Basic.Reject', 'Basic.RecoverAsync']

    def __init__(self, connection, channel_number):
        """Create a new instance of the Channel

        :param BlockingConnection connection: The connection
        :param int channel_number: The channel number for this instance

        """
        super(BlockingChannel, self).__init__(connection, channel_number)
        self.connection = connection
        self._confirmation = False
        self._force_data_events_override = None
        self._generator = None
        self._generator_messages = list()
        self._frames = dict()
        self._replies = list()
        self._wait = False
        self._received_response = False
        self.open()

    def basic_cancel(self, consumer_tag='', nowait=False):
        """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.

        :param str consumer_tag: Identifier for the consumer
        :param bool nowait: Do not expect a Basic.CancelOk response

        """
        if consumer_tag not in self._consumers:
            return
        self._cancelled.append(consumer_tag)
        replies = [(spec.Basic.CancelOk,
                   {'consumer_tag': consumer_tag})] if nowait is False else []
        self._rpc(spec.Basic.Cancel(consumer_tag=consumer_tag,
                                             nowait=nowait),
                  self._on_cancelok, replies)

    def basic_get(self, queue=None, no_ack=False):
        """Get a single message from the AMQP broker. The callback method
        signature should have 3 parameters: The method frame, header frame and
        the body, like the consumer callback for Basic.Consume.

        :param queue: The queue to get a message from
        :type queue: str or unicode
        :param bool no_ack: Tell the broker to not expect a reply
        :rtype: (None, None, None)|(spec.Basic.Get,
                                    spec.Basic.Properties,
                                    str or unicode)

        """
        self._response = None
        self._send_method(spec.Basic.Get(queue=queue,
                                         no_ack=no_ack))
        while not self._response:
            self.connection.process_data_events()
        if isinstance(self._response[0], spec.Basic.GetEmpty):
            return None, None, None
        return self._response[0], self._response[1], self._response[2]

    def basic_publish(self, exchange, routing_key, body,
                      properties=None, mandatory=False, immediate=False):
        """Publish to the channel with the given exchange, routing key and body.
        Returns a boolean value indicating the success of the operation. 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 exchange: The exchange to publish to
        :type exchange: str or unicode
        :param routing_key: The routing key to bind on
        :type routing_key: str or unicode
        :param body: The message body
        :type body: str or unicode
        :param pika.spec.Properties properties: Basic.properties
        :param bool mandatory: The mandatory flag
        :param bool immediate: The immediate flag

        """
        if not self.is_open:
            raise exceptions.ChannelClosed()
        if immediate:
            LOGGER.warning('The immediate flag is deprecated in RabbitMQ')
        properties = properties or spec.BasicProperties()

        if mandatory:
            self._response = None

        if isinstance(body, unicode):
            body = body.encode('utf-8')

        if self._confirmation:
            response = self._rpc(spec.Basic.Publish(exchange=exchange,
                                                    routing_key=routing_key,
                                                    mandatory=mandatory,
                                                    immediate=immediate),
                                 None,
                                 [spec.Basic.Ack,
                                  spec.Basic.Nack],
                                 (properties, body))
            if mandatory and self._response:
                response = self._response[0]
                LOGGER.warning('Message was returned (%s): %s',
                               response.reply_code,
                               response.reply_text)
                return False

            if isinstance(response.method, spec.Basic.Ack):
                return True
            elif isinstance(response.method, spec.Basic.Nack):
                return False
            else:
                raise ValueError('Unexpected frame type: %r', response)
        else:
            self._send_method(spec.Basic.Publish(exchange=exchange,
                                                 routing_key=routing_key,
                                                 mandatory=mandatory,
                                                 immediate=immediate),
                              (properties, body), False)
            if mandatory:
                if self._response:
                    response = self._response[0]
                    LOGGER.warning('Message was returned (%s): %s',
                                   response.reply_code,
                                   response.reply_text)
                    return False
                return True

    def basic_qos(self, prefetch_size=0, prefetch_count=0, all_channels=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 if the
                                   no-ack option is set.
        :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 if the no-ack
                                   option is set.
        :param bool all_channels: Should the QoS apply to all channels

        """
        self._rpc(spec.Basic.Qos(prefetch_size, prefetch_count, all_channels),
                  None, [spec.Basic.QosOk])

    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.

        """
        self._rpc(spec.Basic.Recover(requeue), None, [spec.Basic.RecoverOk])

    def confirm_delivery(self, nowait=False):
        """Turn on Confirm mode in the channel.

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

        :param bool nowait: Do not send a reply frame (Confirm.SelectOk)

        """
        if (not self.connection.publisher_confirms or
            not self.connection.basic_nack):
            raise exceptions.MethodNotImplemented('Not Supported on Server')
        self._confirmation = True
        replies = [spec.Confirm.SelectOk] if nowait is False else []
        self._rpc(spec.Confirm.Select(nowait), None, replies)
        self.connection.process_data_events()

    def cancel(self):
        """Cancel the consumption of a queue, rejecting all pending messages.
        This should only work with the generator based BlockingChannel.consume
        method. If you're looking to cancel a consumer issues with
        BlockingChannel.basic_consume then you should call
        BlockingChannel.basic_cancel.

        :return int: The number of messages requeued by Basic.Nack

        """
        messages = 0
        self.basic_cancel(self._generator)
        if self._generator_messages:
            # Get the last item
            (method, properties, body) = self._generator_messages.pop()
            messages = len(self._generator_messages)
            LOGGER.info('Requeueing %i messages with delivery tag %s',
                        messages, method.delivery_tag)
            self.basic_nack(method.delivery_tag, multiple=True, requeue=True)
            self.connection.process_data_events()
        self._generator = None
        self._generator_messages = list()
        return messages

    def close(self, reply_code=0, reply_text="Normal Shutdown"):
        """Will invoke a clean shutdown of the channel with the AMQP Broker.

        :param int reply_code: The reply code to close the channel with
        :param str reply_text: The reply text to close the channel with

        """

        LOGGER.info('Channel.close(%s, %s)', reply_code, reply_text)
        if not self.is_open:
            raise exceptions.ChannelClosed()

        # Cancel the generator if it's running
        if self._generator:
            self.cancel()

        # If there are any consumers, cancel them as well
        if self._consumers:
            LOGGER.debug('Cancelling %i consumers', len(self._consumers))
            for consumer_tag in self._consumers.keys():
                self.basic_cancel(consumer_tag=consumer_tag)
        self._set_state(self.CLOSING)
        self._rpc(spec.Channel.Close(reply_code, reply_text, 0, 0),
                  None,
                  [spec.Channel.CloseOk])
        self._set_state(self.CLOSED)
        self._cleanup()

    def consume(self, queue, no_ack=False, exclusive=False):
        """Blocking consumption of a queue instead of via a callback. This
        method is a generator that returns messages a tuple of method,
        properties, and body.

        Example:

            for method, properties, body in channel.consume('queue'):
                print body
                channel.basic_ack(method.delivery_tag)

        You should call BlockingChannel.cancel() when you escape out of the
        generator loop. Also note this turns on forced data events to make
        sure that any acked messages actually get acked.

        :param queue: The queue name to consume
        :type queue: str or unicode
        :param no_ack: Tell the broker to not expect a response
        :type no_ack: bool
        :param exclusive: Don't allow other consumers on the queue
        :type exclusive: bool
        :rtype: tuple(spec.Basic.Deliver, spec.BasicProperties, str or unicode)

        """
        LOGGER.debug('Forcing data events on')
        if not self._generator:
            LOGGER.debug('Issuing Basic.Consume')
            self._generator = self.basic_consume(self._generator_callback,
                                                 queue,
                                                 no_ack,
                                                 exclusive)
        while True:
            if self._generator_messages:
                yield self._generator_messages.pop(0)
            self.connection.process_data_events()

    def force_data_events(self, enable):
        """Turn on and off forcing the blocking adapter to stop and look to see
        if there are any frames from RabbitMQ in the read buffer. By default
        the BlockingChannel will check for a read after every RPC command which
        can cause performance to degrade in scenarios where you do not care if
        RabbitMQ is trying to send RPC commands to your client connection.

        Examples of RPC commands of this sort are:

        - Heartbeats
        - Connection.Close
        - Channel.Close
        - Basic.Return
        - Basic.Ack and Basic.Nack when using delivery confirmations

        Turning off forced data events can be a bad thing and prevents your
        client from properly communicating with RabbitMQ. Forced data events
        were added in 0.9.6 to enforce proper channel behavior when
        communicating with RabbitMQ.

        Note that the BlockingConnection also has the constant
        WRITE_TO_READ_RATIO which forces the connection to stop and try and
        read after writing the number of frames specified in the constant.
        This is a way to force the client to received these types of frames
        in a very publish/write IO heavy workload.

        :param bool enable: Set to False to disable

        """
        self._force_data_events_override = enable

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

        :param destination: The destination exchange to bind
        :type destination: str or unicode
        :param source: The source exchange to bind to
        :type source: str or unicode
        :param routing_key: The routing key to bind on
        :type routing_key: str or unicode
        :param bool nowait: Do not wait for an Exchange.BindOk
        :param dict arguments: Custom key/value pair arguments for the binding

        """
        replies = [spec.Exchange.BindOk] if nowait is False else []
        return self._rpc(spec.Exchange.Bind(0, destination, source,
                                            routing_key, nowait,
                                            arguments or dict()), None, replies)

    def exchange_declare(self, exchange=None,
                         exchange_type='direct', passive=False, durable=False,
                         auto_delete=False, internal=False, nowait=False,
                         arguments=None, type=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 exchange: The exchange name consists of a non-empty sequence of
                          these characters: letters, digits, hyphen, underscore,
                          period, or colon.
        :type exchange: str or unicode
        :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 bool nowait: Do not expect an Exchange.DeclareOk response
        :param dict arguments: Custom key/value pair arguments for the exchange
        :param str type: The deprecated exchange type parameter

        """
        if type is not None:
            warnings.warn('type is deprecated, use exchange_type instead',
                          DeprecationWarning)
            if exchange_type == 'direct' and type != exchange_type:
                exchange_type = type
        replies = [spec.Exchange.DeclareOk] if nowait is False else []
        return self._rpc(spec.Exchange.Declare(0, exchange, exchange_type,
                                               passive, durable, auto_delete,
                                               internal, nowait,
                                               arguments or dict()),
                         None, replies)

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

        :param exchange: The exchange name
        :type exchange: str or unicode
        :param bool if_unused: only delete if the exchange is unused
        :param bool nowait: Do not wait for an Exchange.DeleteOk

        """
        replies = [spec.Exchange.DeleteOk] if nowait is False else []
        return self._rpc(spec.Exchange.Delete(0, exchange, if_unused, nowait),
                         None, replies)

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

        :param destination: The destination exchange to unbind
        :type destination: str or unicode
        :param source: The source exchange to unbind from
        :type source: str or unicode
        :param routing_key: The routing key to unbind
        :type routing_key: str or unicode
        :param bool nowait: Do not wait for an Exchange.UnbindOk
        :param dict arguments: Custom key/value pair arguments for the binding

        """
        replies = [spec.Exchange.UnbindOk] if nowait is False else []
        return self._rpc(spec.Exchange.Unbind(0, destination, source,
                                              routing_key, nowait, arguments),
                         None, replies)

    def open(self):
        """Open the channel"""
        self._set_state(self.OPENING)
        self._add_callbacks()
        self._rpc(spec.Channel.Open(), self._on_openok, [spec.Channel.OpenOk])

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

        :param queue: The queue to bind to the exchange
        :type queue: str or unicode
        :param exchange: The source exchange to bind to
        :type exchange: str or unicode
        :param routing_key: The routing key to bind on
        :type routing_key: str or unicode
        :param bool nowait: Do not wait for a Queue.BindOk
        :param dict arguments: Custom key/value pair arguments for the binding

        """
        replies = [spec.Queue.BindOk] if nowait is False else []
        if routing_key is None:
            routing_key = queue
        return self._rpc(spec.Queue.Bind(0, queue, exchange, routing_key,
                                         nowait, arguments or dict()),
                         None, replies)

    def queue_declare(self, queue='', passive=False, durable=False,
                      exclusive=False, auto_delete=False, nowait=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.

        Leave the queue name empty for a auto-named queue in RabbitMQ

        :param queue: The queue name
        :type queue: str or unicode
        :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 bool nowait: Do not wait for a Queue.DeclareOk
        :param dict arguments: Custom key/value arguments for the queue

        """
        condition = (spec.Queue.DeclareOk,
                     {'queue': queue}) if queue else spec.Queue.DeclareOk
        replies = [condition] if nowait is False else []
        return self._rpc(spec.Queue.Declare(0, queue, passive, durable,
                                            exclusive, auto_delete, nowait,
                                            arguments or dict()),
                         None, replies)

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

        :param queue: The queue to delete
        :type queue: str or unicode
        :param bool if_unused: only delete if it's unused
        :param bool if_empty: only delete if the queue is empty
        :param bool nowait: Do not wait for a Queue.DeleteOk

        """
        replies = [spec.Queue.DeleteOk] if nowait is False else []
        return self._rpc(spec.Queue.Delete(0, queue, if_unused, if_empty,
                                           nowait), None, replies)

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

        :param queue: The queue to purge
        :type  queue: str or unicode
        :param bool nowait: Do not expect a Queue.PurgeOk response

        """
        replies = [spec.Queue.PurgeOk] if nowait is False else []
        return self._rpc(spec.Queue.Purge(0, queue, nowait), None, replies)

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

        :param queue: The queue to unbind from the exchange
        :type queue: str or unicode
        :param exchange: The source exchange to bind from
        :type exchange: str or unicode
        :param routing_key: The routing key to unbind
        :type routing_key: str or unicode
        :param dict arguments: Custom key/value pair arguments for the binding

        """
        if routing_key is None:
            routing_key = queue
        return self._rpc(spec.Queue.Unbind(0, queue, exchange, routing_key,
                                           arguments or dict()), None,
                         [spec.Queue.UnbindOk])

    def start_consuming(self):
        """Starts consuming from registered callbacks."""
        while len(self._consumers):
            self.connection.process_data_events()

    def stop_consuming(self, consumer_tag=None):
        """Sends off the Basic.Cancel to let RabbitMQ know to stop consuming and
        sets our internal state to exit out of the basic_consume.

        """
        if consumer_tag:
            self.basic_cancel(consumer_tag)
        else:
            for consumer_tag in self._consumers.keys():
                self.basic_cancel(consumer_tag)
        self.wait = True

    def tx_commit(self):
        """Commit a transaction."""
        return self._rpc(spec.Tx.Commit(), None, [spec.Tx.CommitOk])

    def tx_rollback(self):
        """Rollback a transaction."""
        return self._rpc(spec.Tx.Rollback(), None, [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._rpc(spec.Tx.Select(), None, [spec.Tx.SelectOk])

    # Internal methods

    def _add_reply(self, reply):
        reply = callback._name_or_value(reply)
        self._replies.append(reply)

    def _add_callbacks(self):
        """Add callbacks for when the channel opens and closes."""
        self.connection.callbacks.add(self.channel_number,
                                      spec.Channel.Close,
                                      self._on_close)
        self.callbacks.add(self.channel_number,
                           spec.Basic.GetEmpty,
                           self._on_getempty,
                           False)
        self.callbacks.add(self.channel_number,
                           spec.Basic.Cancel,
                           self._on_cancel,
                           False)
        self.connection.callbacks.add(self.channel_number,
                                      spec.Channel.CloseOk,
                                      self._on_rpc_complete)

    def _generator_callback(self, unused, method, properties, body):
        """Called when a message is received from RabbitMQ and appended to the
        list of messages to be returned when a message is received by RabbitMQ.

        :param pika.spec.Basic.Deliver: The method frame received
        :param pika.spec.BasicProperties: The  message properties
        :param body: The body received
        :type body: str or unicode

        """
        self._generator_messages.append((method, properties, body))
        LOGGER.debug('%i pending messages', len(self._generator_messages))

    def _on_cancel(self, method_frame):
        """Raises a ConsumerCanceled exception after processing the frame


        :param pika.frame.Method method_frame: The method frame received

        """
        super(BlockingChannel, self)._on_cancel(method_frame)
        raise exceptions.ConsumerCancelled(method_frame.method)

    def _on_getok(self, method_frame, header_frame, body):
        """Called in reply to a Basic.Get when there is a message.

        :param pika.frame.Method method_frame: The method frame received
        :param pika.frame.Header header_frame: The header frame received
        :param body: The body received
        :type body: str or unicode

        """
        self._received_response = True
        self._response = method_frame.method, header_frame.properties, body

    def _on_getempty(self, frame):
        self._received_response = True
        self._response = frame.method, None, None

    def _on_close(self, method_frame):
        LOGGER.warning('Received Channel.Close, closing: %r', method_frame)
        if not self.connection.is_closed:
            self._send_method(spec.Channel.CloseOk(), None, False)
        self._set_state(self.CLOSED)
        self._cleanup()
        self._generator_messages = list()
        self._generator = None
        if method_frame is None:
            raise exceptions.ChannelClosed(0, 'Not specified')
        else:
            raise exceptions.ChannelClosed(method_frame.method.reply_code,
                                           method_frame.method.reply_text)

    def _on_openok(self, method_frame):
        """Open the channel by sending the RPC command and remove the reply
        from the stack of replies.

        """
        super(BlockingChannel, self)._on_openok(method_frame)
        self._remove_reply(method_frame)

    def _on_return(self, method_frame, header_frame, body):
        """Called when a Basic.Return is received from publishing

        :param pika.frame.Method method_frame: The method frame received
        :param pika.frame.Header header_frame: The header frame received
        :param body: The body received
        :type body: str or unicode

        """
        self._received_response = True
        self._response = method_frame.method, header_frame.properties, body

    def _on_rpc_complete(self, frame):
        key = callback._name_or_value(frame)
        self._replies.append(key)
        self._frames[key] = frame
        self._received_response = True

    def _process_replies(self, replies, callback):
        """Process replies from RabbitMQ, looking in the stack of callback
        replies for a match. Will optionally call callback prior to
        returning the frame_value.

        :param list replies: The reply handles to iterate
        :param method callback: The method to optionally call
        :rtype: pika.frame.Frame

        """
        for reply in self._replies:
            if reply in replies:
                frame_value = self._frames[reply]
                self._received_response = True
                if callback:
                    callback(frame_value)
                del(self._frames[reply])
                return frame_value

    def _remove_reply(self, frame):
        key = callback._name_or_value(frame)
        if key in self._replies:
            self._replies.remove(key)

    def _rpc(self, method_frame, callback=None, acceptable_replies=None,
             content=None, force_data_events=True):
        """Make an RPC call for the given callback, channel number and method.
        acceptable_replies lists out what responses we'll process from the
        server with the specified callback.

        :param pika.amqp_object.Method method_frame: The method frame to call
        :param method callback: The callback for the RPC response
        :param list acceptable_replies: The replies this RPC call expects
        :param tuple content: Properties and Body for content frames
        :param bool force_data_events: Call process data events before reply
        :rtype: pika.frame.Method

        """
        if self.is_closed:
            raise exceptions.ChannelClosed
        self._validate_acceptable_replies(acceptable_replies)
        self._validate_callback(callback)
        replies = list()
        for reply in acceptable_replies or list():
            if isinstance(reply, tuple):
                reply, arguments = reply
            else:
                arguments = None
            prefix, key = self.callbacks.add(self.channel_number,
                                             reply,
                                             self._on_rpc_complete,
                                             arguments=arguments)
            replies.append(key)
        self._send_method(method_frame, content,
                          self._wait_on_response(method_frame))
        if force_data_events and self._force_data_events_override is not False:
            self.connection.process_data_events()
        return self._process_replies(replies, callback)

    def _send_method(self, method_frame, content=None, wait=False):
        """Shortcut wrapper to send a method through our connection, passing in
        our channel number.

        :param pika.amqp_object.Method method_frame: The method frame to send
        :param content: The content to send
        :type content: tuple
        :param bool wait: Wait for a response

        """
        self.wait = wait
        prev_received_response = self._received_response
        self._received_response = False
        self.connection.send_method(self.channel_number, method_frame, content)
        while wait and not self._received_response:
            try:
                self.connection.process_data_events()
            except exceptions.AMQPConnectionError:
                break
        self._received_response = prev_received_response

    def _validate_acceptable_replies(self, acceptable_replies):
        """Validate the list of acceptable replies

        :param acceptable_replies:
        :raises: TypeError

        """
        if acceptable_replies and not isinstance(acceptable_replies, list):
            raise TypeError("acceptable_replies should be list or None, is %s",
                            type(acceptable_replies))

    def _validate_callback(self, callback):
        """Validate the value passed in is a method or function.

        :param method callback callback: The method to validate
        :raises: TypeError

        """
        if (callback is not None and
            not utils.is_callable(callback)):
            raise TypeError("Callback should be a function or method, is %s",
                            type(callback))

    def _wait_on_response(self, method_frame):
        """Returns True if the rpc call should wait on a response.

        :param pika.frame.Method method_frame: The frame to check

        """
        return method_frame.NAME not in self.NO_RESPONSE_FRAMES