"""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