"""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 ` 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() ` 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 `. :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 `, 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 `, 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