Introduction to Pika

Pika is a pure-Python implementation of the AMQP 0-9-1 protocol that tries to stay fairly independent of the underlying network support library.

If you have not developed with Pika or RabbitMQ before, the Introduction to Pika documentation is a good place to get started.

Installing Pika

Pika is available for download via PyPI and may be installed using easy_install or pip:

pip install pika

or:

easy_install pika

To install from source, run “python setup.py install” in the root source directory.

Using Pika

Introduction to Pika

IO and Event Looping

As AMQP is a two-way RPC protocol where the client can send requests to the server and the server can send requests to a client, Pika implements or extends IO loops in each of its asynchronous connection adapters. These IO loops are blocking methods which loop and listen for events. Each asynchronous adapters follows the same standard for invoking the IO loop. The IO loop is created when the connection adapter is created. To start an IO loop for any given adapter, call the connection.ioloop.start() method.

If you are using an external IO loop such as Tornado’s IOLoop, you invoke it as you normally would and then add the adapter to it.

Example:

import pika

def on_open(connection):
    # Invoked when the connection is open
    pass

# Create our connection object, passing in the on_open method
connection = pika.SelectConnection(on_open_callback=on_open)

try:
    # Loop so we can communicate with RabbitMQ
    connection.ioloop.start()
except KeyboardInterrupt:
    # Gracefully close the connection
    connection.close()
    # Loop until we're fully closed, will stop on its own
    connection.ioloop.start()

Continuation-Passing Style

Interfacing with Pika asynchronously is done by passing in callback methods you would like to have invoked when a certain event has completed. For example, if you are going to declare a queue, you pass in a method that will be called when the RabbitMQ server returns a Queue.DeclareOk response.

In our example below we use the following four easy steps:

  1. We start by creating our connection object, then starting our event loop.
  2. When we are connected, the on_connected method is called. In that method we create a channel.
  3. When the channel is created, the on_channel_open method is called. In that method we declare a queue.
  4. When the queue is declared successfully, on_queue_declared is called. In that method we call channel.basic_consume telling it to call the handle_delivery for each message RabbitMQ delivers to us.
  5. When RabbitMQ has a message to send us, it call the handle_delivery method passing the AMQP Method frame, Header frame and Body.

Note

Step #1 is on line #28 and Step #2 is on line #6. This is so that Python knows about the functions we’ll call in Steps #2 through #5.

Example:

import pika

# Create a global channel variable to hold our channel object in
channel = None

# Step #2
def on_connected(connection):
    """Called when we are fully connected to RabbitMQ"""
    # Open a channel
    connection.channel(on_channel_open)

# Step #3
def on_channel_open(new_channel):
    """Called when our channel has opened"""
    global channel
    channel = new_channel
    channel.queue_declare(queue="test", durable=True, exclusive=False, auto_delete=False, callback=on_queue_declared)

# Step #4
def on_queue_declared(frame):
    """Called when RabbitMQ has told us our Queue has been declared, frame is the response from RabbitMQ"""
    channel.basic_consume(handle_delivery, queue='test')

# Step #5
def handle_delivery(channel, method, header, body):
    """Called when we receive a message from RabbitMQ"""
    print body

# Step #1: Connect to RabbitMQ using the default parameters
parameters = pika.ConnectionParameters()
connection = pika.SelectConnection(parameters, on_connected)

try:
    # Loop so we can communicate with RabbitMQ
    connection.ioloop.start()
except KeyboardInterrupt:
    # Gracefully close the connection
    connection.close()
    # Loop until we're fully closed, will stop on its own
    connection.ioloop.start()

Credentials

The pika.credentials module provides the mechanism by which you pass the username and password to the ConnectionParameters class when it is created.

Example:

import pika
credentials = pika.PlainCredentials('username', 'password')
parameters = pika.ConnectionParameters(credentials=credentials)

Connection Parameters

There are two types of connection parameter classes in Pika to allow you to pass the connection information into a connection adapter, ConnectionParameters and URLParameters. Both classes share the same default connection values.

TCP Backpressure

As of RabbitMQ 2.0, client side Channel.Flow has been removed [1]. Instead, the RabbitMQ broker uses TCP Backpressure to slow your client if it is delivering messages too fast. If you pass in backpressure_detection into your connection parameters, Pika attempts to help you handle this situation by providing a mechanism by which you may be notified if Pika has noticed too many frames have yet to be delivered. By registering a callback function with the add_backpressure_callback method of any connection adapter, your function will be called when Pika sees that a backlog of 10 times the average frame size you have been sending has been exceeded. You may tweak the notification multiplier value by calling the set_backpressure_multiplier method passing any integer value.

Example:

import pika

parameters = pika.URLParameters('amqp://guest:guest@rabbit-server1:5672/%2F?backpressure_detection=t')

Footnotes

[1]“more effective flow control mechanism that does not require cooperation from clients and reacts quickly to prevent the broker from exhausing memory - see http://www.rabbitmq.com/extensions.html#memsup” from http://lists.rabbitmq.com/pipermail/rabbitmq-announce/attachments/20100825/2c672695/attachment.txt

Core Class and Module Documentation

For the end user, Pika is organized into a small set of objects for all communication with RabbitMQ.

Connection Adapters

Pika uses connection adapters to provide a flexible method for adapting pika’s core communication to different IOLoop implementations. In addition to asynchronous adapters, there is the BlockingConnection adapter that provides a more idomatic procedural approach to using Pika.

Adapters
BlockingConnection

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 BlockingConnection and the BlockingChannel classes.

Be sure to check out examples in Usage Examples.

class pika.adapters.blocking_connection.BlockingConnection(parameters=None, _impl_class=None)[source]

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 basic_consume or if you want to be notified of a delivery failure when using basic_publish .

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

add_on_connection_blocked_callback(callback_method)[source]

Add a callback to be notified when RabbitMQ has sent a Connection.Blocked frame indicating that RabbitMQ is low on resources. Publishers can use this to voluntarily suspend publishing, instead of relying on back pressure throttling. The callback will be passed the Connection.Blocked method frame.

Parameters:callback_method (method) – Callback to call on Connection.Blocked, having the signature callback_method(pika.frame.Method), where the method frame’s method member is of type pika.spec.Connection.Blocked
add_on_connection_unblocked_callback(callback_method)[source]

Add a callback to be notified when RabbitMQ has sent a Connection.Unblocked frame letting publishers know it’s ok to start publishing again. The callback will be passed the Connection.Unblocked method frame.

Parameters:callback_method (method) – Callback to call on Connection.Unblocked, having the signature callback_method(pika.frame.Method), where the method frame’s method member is of type pika.spec.Connection.Unblocked
add_timeout(deadline, callback_method)[source]

Create a single-shot timer to fire after deadline seconds. 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.

NOTE: the timer callbacks are dispatched only in the scope of specially-designated methods: see BlockingConnection.process_data_events and BlockingChannel.start_consuming.

Parameters:
  • deadline (float) – The number of seconds to wait to call callback
  • callback_method (callable) – The callback method with the signature callback_method()
Returns:

opaque timer id

basic_nack

Specifies if the server supports basic.nack on the active connection.

Return type:bool
basic_nack_supported

Specifies if the server supports basic.nack on the active connection.

Return type:bool
channel(channel_number=None)[source]

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.

Return type:pika.synchronous_connection.BlockingChannel
close(reply_code=200, reply_text='Normal shutdown')[source]

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.

Parameters:
  • reply_code (int) – The code number for the close
  • reply_text (str) – The text reason for the close
consumer_cancel_notify

Specifies if the server supports consumer cancel notification on the active connection.

Return type:bool
consumer_cancel_notify_supported

Specifies if the server supports consumer cancel notification on the active connection.

Return type:bool
exchange_exchange_bindings

Specifies if the active connection supports exchange to exchange bindings.

Return type:bool
exchange_exchange_bindings_supported

Specifies if the active connection supports exchange to exchange bindings.

Return type:bool
is_closed

Returns a boolean reporting the current connection state.

is_closing

Returns a boolean reporting the current connection state.

is_open

Returns a boolean reporting the current connection state.

process_data_events(time_limit=0)[source]

Will make sure that data events are processed. Dispatches timer and channel callbacks if not called from the scope of BlockingConnection or BlockingChannel callback. Your app can block on this method.

Parameters:time_limit (float) – suggested upper bound on processing time in seconds. The actual blocking time depends on the granularity of the underlying ioloop. Zero means return as soon as possible. None means there is no limit on processing time and the function will block until I/O produces actionalable events. Defaults to 0 for backward compatibility. This parameter is NEW in pika 0.10.0.
publisher_confirms

Specifies if the active connection can use publisher confirmations.

Return type:bool
publisher_confirms_supported

Specifies if the active connection can use publisher confirmations.

Return type:bool
remove_timeout(timeout_id)[source]

Remove a timer if it’s still in the timeout stack

Parameters:timeout_id – The opaque timer id to remove
sleep(duration)[source]

A safer way to sleep than calling time.sleep() directly that would keep the adapter from ignoring frames sent from the broker. The connection will “sleep” or block the number of seconds specified in duration in small intervals.

Parameters:duration (float) – The time to sleep in seconds
class pika.adapters.blocking_connection.BlockingChannel(channel_impl, connection)[source]

The BlockingChannel implements blocking semantics for most things that one would use callback-passing-style for with the Channel class. In addition, the BlockingChannel class implements a generator that allows you to consume messages without using callbacks.

Example of creating a BlockingChannel:

import pika

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

# The returned object will be a synchronous channel
channel = connection.channel()
add_on_cancel_callback(callback)[source]

Pass a callback function that will be called when Basic.Cancel is sent by the broker. The callback function should receive a method frame parameter.

Parameters:callback (callable) – a callable for handling broker’s Basic.Cancel notification with the call signature: callback(method_frame) where method_frame is of type pika.frame.Method with method of type spec.Basic.Cancel
add_on_return_callback(callback)[source]

Pass a callback function that will be called when a published message is rejected and returned by the server via Basic.Return.

Parameters:callback (callable) – The method to call on callback with the signature callback(channel, method, properties, body), where channel: pika.Channel method: pika.spec.Basic.Return properties: pika.spec.BasicProperties body: str, unicode, or bytes (python 3.x)
basic_ack(delivery_tag=0, multiple=False)[source]

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.

Parameters:
  • delivery-tag (int) – The server-assigned delivery tag
  • multiple (bool) – 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.
basic_cancel(consumer_tag)[source]

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.

NOTE: When cancelling a no_ack=False consumer, this implementation automatically Nacks and suppresses any incoming messages that have not yet been dispatched to the consumer’s callback. However, when cancelling a no_ack=True consumer, this method will return any pending messages that arrived before broker confirmed the cancellation.

Parameters:consumer_tag (str) – Identifier for the consumer; the result of passing a consumer_tag that was created on another channel is undefined (bad things will happen)
Returns:(NEW IN pika 0.10.0) empty sequence for a no_ack=False consumer; for a no_ack=True consumer, returns a (possibly empty) sequence of pending messages that arrived before broker confirmed the cancellation (this is done instead of via consumer’s callback in order to prevent reentrancy/recursion. Each message is four-tuple: (channel, method, properties, body)
channel: BlockingChannel method: spec.Basic.Deliver properties: spec.BasicProperties body: str or unicode
basic_consume(consumer_callback, queue, no_ack=False, exclusive=False, consumer_tag=None, arguments=None)[source]

Sends the AMQP command Basic.Consume to the broker and binds messages for the consumer_tag to the consumer callback. If you do not pass in a consumer_tag, one will be automatically generated for you. Returns the consumer tag.

NOTE: the consumer callbacks are dispatched only in the scope of specially-designated methods: see BlockingConnection.process_data_events and BlockingChannel.start_consuming.

For more information about Basic.Consume, see: http://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.consume

Parameters:
  • consumer_callback (callable) –

    The function for dispatching messages to user, having the signature: consumer_callback(channel, method, properties, body)

    channel: BlockingChannel method: spec.Basic.Deliver properties: spec.BasicProperties body: str or unicode
  • queue (str or unicode) – The queue to consume from
  • no_ack (bool) – Tell the broker to not expect a response (i.e., no ack/nack)
  • exclusive (bool) – Don’t allow other consumers on the queue
  • consumer_tag (str or unicode) – You may specify your own consumer tag; if left empty, a consumer tag will be generated automatically
  • arguments (dict) – Custom key/value pair arguments for the consumer
Returns:

consumer tag

Return type:

str

Raises pika.exceptions.DuplicateConsumerTag:
 

if consumer with given consumer_tag is already present.

basic_get(queue=None, no_ack=False)[source]

Get a single message from the AMQP broker. Returns a sequence with the method frame, message properties, and body.

Parameters:
  • queue (str or unicode) – Name of queue to get a message from
  • no_ack (bool) – Tell the broker to not expect a reply
Returns:

a three-tuple; (None, None, None) if the queue was empty; otherwise (method, properties, body); NOTE: body may be None

Return type:

(None, None, None)|(spec.Basic.GetOk, spec.BasicProperties, str or unicode or None)

basic_nack(delivery_tag=None, multiple=False, requeue=True)[source]

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.

Parameters:
  • delivery-tag (int) – The server-assigned delivery tag
  • multiple (bool) – 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.
  • requeue (bool) – 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.
basic_publish(exchange, routing_key, body, properties=None, mandatory=False, immediate=False)[source]

Publish to the channel with the given exchange, routing key and body. Returns a boolean value indicating the success of the operation.

This is the legacy BlockingChannel method for publishing. See also BasicChannel.publish that provides more information about failures.

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

NOTE: mandatory and immediate may be enabled even without delivery
confirmation, but in the absence of delivery confirmation the synchronous implementation has no way to know how long to wait for the Basic.Return or lack thereof.
Parameters:
  • exchange (str or unicode) – The exchange to publish to
  • routing_key (str or unicode) – The routing key to bind on
  • body (str or unicode) – The message body; empty string if no body
  • properties (pika.spec.BasicProperties) – message properties
  • mandatory (bool) – The mandatory flag
  • immediate (bool) – The immediate flag
Returns:

True if delivery confirmation is not enabled (NEW in pika 0.10.0); otherwise returns False if the message could not be deliveved (Basic.nack and/or Basic.Return) and True if the message was delivered (Basic.ack and no Basic.Return)

basic_qos(prefetch_size=0, prefetch_count=0, all_channels=False)[source]

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.

Parameters:
  • prefetch_size (int) – 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 in the consumer.
  • prefetch_count (int) – 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 in the consumer.
  • all_channels (bool) – Should the QoS apply to all channels
basic_recover(requeue=False)[source]

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.

Parameters:requeue (bool) – 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.
basic_reject(delivery_tag=None, requeue=True)[source]

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.

Parameters:
  • delivery-tag (int) – The server-assigned delivery tag
  • requeue (bool) – 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.
cancel()[source]

Cancel the queue consumer created by BlockingChannel.consume, rejecting all pending ackable messages.

NOTE: If you’re looking to cancel a consumer issued with BlockingChannel.basic_consume then you should call BlockingChannel.basic_cancel.

Return int:The number of messages requeued by Basic.Nack. NEW in 0.10.0: returns 0
channel_number

Channel number

close(reply_code=0, reply_text='Normal Shutdown')[source]

Will invoke a clean shutdown of the channel with the AMQP Broker.

Parameters:
  • reply_code (int) – The reply code to close the channel with
  • reply_text (str) – The reply text to close the channel with
confirm_delivery()[source]

Turn on RabbitMQ-proprietary Confirm mode in the channel.

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

The channel’s BlockingConnection instance

consume(queue, no_ack=False, exclusive=False, arguments=None, inactivity_timeout=None)[source]

Blocking consumption of a queue instead of via a callback. This method is a generator that yields each message as a tuple of method, properties, and body. The active generator iterator terminates when the consumer is cancelled by client or broker.

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.

If you don’t cancel this consumer, then next call on the same channel to consume() with the exact same (queue, no_ack, exclusive) parameters will resume the existing consumer generator; however, calling with different parameters will result in an exception.

Parameters:
  • queue (str or unicode) – The queue name to consume
  • no_ack (bool) – Tell the broker to not expect a ack/nack response
  • exclusive (bool) – Don’t allow other consumers on the queue
  • arguments (dict) – Custom key/value pair arguments for the consumer
  • inactivity_timeout (float) – if a number is given (in seconds), will cause the method to yield None after the given period of inactivity; this permits for pseudo-regular maintenance activities to be carried out by the user while waiting for messages to arrive. If None is given (default), then the method blocks until the next event arrives. NOTE that timing granularity is limited by the timer resolution of the underlying implementation. NEW in pika 0.10.0.
Yields:

tuple(spec.Basic.Deliver, spec.BasicProperties, str or unicode)

Raises ValueError:
 

if consumer-creation parameters don’t match those of the existing queue consumer generator, if any. NEW in pika 0.10.0

exchange_bind(destination=None, source=None, routing_key='', arguments=None)[source]

Bind an exchange to another exchange.

Parameters:
  • destination (str or unicode) – The destination exchange to bind
  • source (str or unicode) – The source exchange to bind to
  • routing_key (str or unicode) – The routing key to bind on
  • arguments (dict) – Custom key/value pair arguments for the binding
Returns:

Method frame from the Exchange.Bind-ok response

Return type:

pika.frame.Method having method attribute of type spec.Exchange.BindOk

exchange_declare(exchange=None, exchange_type='direct', passive=False, durable=False, auto_delete=False, internal=False, arguments=None, **kwargs)[source]

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).

Parameters:
  • exchange (str or unicode) – The exchange name consists of a non-empty sequence of these characters: letters, digits, hyphen, underscore, period, or colon.
  • exchange_type (str) – The exchange type to use
  • passive (bool) – Perform a declare or just check to see if it exists
  • durable (bool) – Survive a reboot of RabbitMQ
  • auto_delete (bool) – Remove when no more queues are bound to it
  • internal (bool) – Can only be published to by other exchanges
  • arguments (dict) – Custom key/value pair arguments for the exchange
  • type (str) – via kwargs: the deprecated exchange type parameter
Returns:

Method frame from the Exchange.Declare-ok response

Return type:

pika.frame.Method having method attribute of type spec.Exchange.DeclareOk

exchange_delete(exchange=None, if_unused=False)[source]

Delete the exchange.

Parameters:
  • exchange (str or unicode) – The exchange name
  • if_unused (bool) – only delete if the exchange is unused
Returns:

Method frame from the Exchange.Delete-ok response

Return type:

pika.frame.Method having method attribute of type spec.Exchange.DeleteOk

exchange_unbind(destination=None, source=None, routing_key='', arguments=None)[source]

Unbind an exchange from another exchange.

Parameters:
  • destination (str or unicode) – The destination exchange to unbind
  • source (str or unicode) – The source exchange to unbind from
  • routing_key (str or unicode) – The routing key to unbind
  • arguments (dict) – Custom key/value pair arguments for the binding
Returns:

Method frame from the Exchange.Unbind-ok response

Return type:

pika.frame.Method having method attribute of type spec.Exchange.UnbindOk

flow(active)[source]

Turn Channel flow control off and on.

NOTE: RabbitMQ doesn’t support active=False; per https://www.rabbitmq.com/specification.html: “active=false is not supported by the server. Limiting prefetch with basic.qos provides much better control”

For more information, please reference:

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

Parameters:active (bool) – Turn flow on (True) or off (False)
Returns:True if broker will start or continue sending; False if not
Return type:bool
get_waiting_message_count()[source]

Returns the number of messages that may be retrieved from the current queue consumer generator via BasicChannel.consume without blocking. NEW in pika 0.10.0

Return type:int
is_closed

Returns True if the channel is closed.

Return type:bool
is_closing

Returns True if the channel is closing.

Return type:bool
is_open

Returns True if the channel is open.

Return type:bool
publish(exchange, routing_key, body, properties=None, mandatory=False, immediate=False)[source]

Publish to the channel with the given exchange, routing key, and body. Unlike the legacy BlockingChannel.basic_publish, this method provides more information about failures via exceptions.

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

NOTE: mandatory and immediate may be enabled even without delivery
confirmation, but in the absence of delivery confirmation the synchronous implementation has no way to know how long to wait for the Basic.Return.
Parameters:
  • exchange (str or unicode) – The exchange to publish to
  • routing_key (str or unicode) – The routing key to bind on
  • body (str or unicode) – The message body; empty string if no body
  • properties (pika.spec.BasicProperties) – message properties
  • mandatory (bool) – The mandatory flag
  • immediate (bool) – The immediate flag
Raises:
  • UnroutableError – raised when a message published in publisher-acknowledgments mode (see BlockingChannel.confirm_delivery) is returned via Basic.Return followed by Basic.Ack.
  • NackError – raised when a message published in publisher-acknowledgements mode is Nack’ed by the broker. See BlockingChannel.confirm_delivery.
queue_bind(queue, exchange, routing_key=None, arguments=None)[source]

Bind the queue to the specified exchange

Parameters:
  • queue (str or unicode) – The queue to bind to the exchange
  • exchange (str or unicode) – The source exchange to bind to
  • routing_key (str or unicode) – The routing key to bind on
  • arguments (dict) – Custom key/value pair arguments for the binding
Returns:

Method frame from the Queue.Bind-ok response

Return type:

pika.frame.Method having method attribute of type spec.Queue.BindOk

queue_declare(queue='', passive=False, durable=False, exclusive=False, auto_delete=False, arguments=None)[source]

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

Parameters:
  • queue (str or unicode; if empty string, the broker will create a unique queue name;) – The queue name
  • passive (bool) – Only check to see if the queue exists
  • durable (bool) – Survive reboots of the broker
  • exclusive (bool) – Only allow access by the current connection
  • auto_delete (bool) – Delete after consumer cancels or disconnects
  • arguments (dict) – Custom key/value arguments for the queue
Returns:

Method frame from the Queue.Declare-ok response

Return type:

pika.frame.Method having method attribute of type spec.Queue.DeclareOk

queue_delete(queue='', if_unused=False, if_empty=False)[source]

Delete a queue from the broker.

Parameters:
  • queue (str or unicode) – The queue to delete
  • if_unused (bool) – only delete if it’s unused
  • if_empty (bool) – only delete if the queue is empty
Returns:

Method frame from the Queue.Delete-ok response

Return type:

pika.frame.Method having method attribute of type spec.Queue.DeleteOk

queue_purge(queue='')[source]

Purge all of the messages from the specified queue

Parameters:queue (str or unicode) – The queue to purge
Returns:Method frame from the Queue.Purge-ok response
Return type:pika.frame.Method having method attribute of type spec.Queue.PurgeOk
queue_unbind(queue='', exchange=None, routing_key=None, arguments=None)[source]

Unbind a queue from an exchange.

Parameters:
  • queue (str or unicode) – The queue to unbind from the exchange
  • exchange (str or unicode) – The source exchange to bind from
  • routing_key (str or unicode) – The routing key to unbind
  • arguments (dict) – Custom key/value pair arguments for the binding
Returns:

Method frame from the Queue.Unbind-ok response

Return type:

pika.frame.Method having method attribute of type spec.Queue.UnbindOk

start_consuming()[source]

Processes I/O events and dispatches timers and basic_consume callbacks until all consumers are cancelled.

NOTE: this blocking function may not be called from the scope of a pika callback, because dispatching basic_consume callbacks from this context would constitute recursion.

Raises pika.exceptions.RecursionError:
 if called from the scope of a BlockingConnection or BlockingChannel callback
stop_consuming(consumer_tag=None)[source]

Cancels all consumers, signalling the start_consuming loop to exit.

NOTE: pending non-ackable messages will be lost; pending ackable messages will be rejected.

tx_commit()[source]

Commit a transaction.

Returns:Method frame from the Tx.Commit-ok response
Return type:pika.frame.Method having method attribute of type spec.Tx.CommitOk
tx_rollback()[source]

Rollback a transaction.

Returns:Method frame from the Tx.Commit-ok response
Return type:pika.frame.Method having method attribute of type spec.Tx.CommitOk
tx_select()[source]

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:Method frame from the Tx.Select-ok response
Return type:pika.frame.Method having method attribute of type spec.Tx.SelectOk
Select Connection Adapter

A connection adapter that tries to use the best polling method for the platform pika is running on.

class pika.adapters.select_connection.SelectConnection(parameters=None, on_open_callback=None, on_open_error_callback=None, on_close_callback=None, stop_ioloop_on_close=True, custom_ioloop=None)[source]

An asynchronous connection adapter that attempts to use the fastest event loop adapter for the given platform.

add_backpressure_callback(callback_method)

Call method “callback” when pika believes backpressure is being applied.

Parameters:callback_method (method) – The method to call
add_on_close_callback(callback_method)

Add a callback notification when the connection has closed. The callback will be passed the connection, the reply_code (int) and the reply_text (str), if sent by the remote server.

Parameters:callback_method (method) – Callback to call on close
add_on_connection_blocked_callback(callback_method)

Add a callback to be notified when RabbitMQ has sent a Connection.Blocked frame indicating that RabbitMQ is low on resources. Publishers can use this to voluntarily suspend publishing, instead of relying on back pressure throttling. The callback will be passed the Connection.Blocked method frame.

Parameters:callback_method (method) – Callback to call on Connection.Blocked
add_on_connection_unblocked_callback(callback_method)

Add a callback to be notified when RabbitMQ has sent a Connection.Unblocked frame letting publishers know it’s ok to start publishing again. The callback will be passed the Connection.Unblocked method frame.

Parameters:callback_method (method) – Callback to call on Connection.Unblocked
add_on_open_callback(callback_method)

Add a callback notification when the connection has opened.

Parameters:callback_method (method) – Callback to call when open
add_on_open_error_callback(callback_method, remove_default=True)

Add a callback notification when the connection can not be opened.

The callback method should accept the connection object that could not connect, and an optional error message.

Parameters:
  • callback_method (method) – Callback to call when can’t connect
  • remove_default (bool) – Remove default exception raising callback
add_timeout(deadline, callback_method)

Add the callback_method to the IOLoop timer to fire after deadline seconds. Returns a handle to the timeout

Parameters:
  • deadline (int) – The number of seconds to wait to call callback
  • callback_method (method) – The callback method
Return type:

str

basic_nack

Specifies if the server supports basic.nack on the active connection.

Return type:bool
channel(on_open_callback, channel_number=None)

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.

Parameters:
  • on_open_callback (method) – The callback when the channel is opened
  • channel_number (int) – The channel number to use, defaults to the next available.
Return type:

pika.channel.Channel

close(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.

Parameters:
  • reply_code (int) – The code number for the close
  • reply_text (str) – The text reason for the close
connect()

Invoke if trying to reconnect to a RabbitMQ server. Constructing the Connection object should connect on its own.

consumer_cancel_notify

Specifies if the server supports consumer cancel notification on the active connection.

Return type:bool
exchange_exchange_bindings

Specifies if the active connection supports exchange to exchange bindings.

Return type:bool
is_closed

Returns a boolean reporting the current connection state.

is_closing

Returns a boolean reporting the current connection state.

is_open

Returns a boolean reporting the current connection state.

publisher_confirms

Specifies if the active connection can use publisher confirmations.

Return type:bool
remove_timeout(timeout_id)

Remove the timeout from the IOLoop by the ID returned from add_timeout.

Return type:str
set_backpressure_multiplier(value=10)

Alter the backpressure multiplier value. We set this to 10 by default. This value is used to raise warnings and trigger the backpressure callback.

Parameters:value (int) – The multiplier value to set
Tornado Connection Adapter

Use pika with the Tornado IOLoop

Be sure to check out the asynchronous examples including the Tornado specific consumer example.

class pika.adapters.tornado_connection.TornadoConnection(parameters=None, on_open_callback=None, on_open_error_callback=None, on_close_callback=None, stop_ioloop_on_close=False, custom_ioloop=None)[source]

The TornadoConnection runs on the Tornado IOLoop. If you’re running the connection in a web app, make sure you set stop_ioloop_on_close to False, which is the default behavior for this adapter, otherwise the web app will stop taking requests.

Parameters:
  • parameters (pika.connection.Parameters) – Connection parameters
  • on_open_callback (method) – The method to call when the connection is open
  • on_open_error_callback (method) – Method to call if the connection cant be opened
  • stop_ioloop_on_close (bool) – Call ioloop.stop() if disconnected
  • custom_ioloop – Override using the global IOLoop in Tornado
add_backpressure_callback(callback_method)

Call method “callback” when pika believes backpressure is being applied.

Parameters:callback_method (method) – The method to call
add_on_close_callback(callback_method)

Add a callback notification when the connection has closed. The callback will be passed the connection, the reply_code (int) and the reply_text (str), if sent by the remote server.

Parameters:callback_method (method) – Callback to call on close
add_on_connection_blocked_callback(callback_method)

Add a callback to be notified when RabbitMQ has sent a Connection.Blocked frame indicating that RabbitMQ is low on resources. Publishers can use this to voluntarily suspend publishing, instead of relying on back pressure throttling. The callback will be passed the Connection.Blocked method frame.

Parameters:callback_method (method) – Callback to call on Connection.Blocked
add_on_connection_unblocked_callback(callback_method)

Add a callback to be notified when RabbitMQ has sent a Connection.Unblocked frame letting publishers know it’s ok to start publishing again. The callback will be passed the Connection.Unblocked method frame.

Parameters:callback_method (method) – Callback to call on Connection.Unblocked
add_on_open_callback(callback_method)

Add a callback notification when the connection has opened.

Parameters:callback_method (method) – Callback to call when open
add_on_open_error_callback(callback_method, remove_default=True)

Add a callback notification when the connection can not be opened.

The callback method should accept the connection object that could not connect, and an optional error message.

Parameters:
  • callback_method (method) – Callback to call when can’t connect
  • remove_default (bool) – Remove default exception raising callback
add_timeout(deadline, callback_method)[source]

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.

Parameters:
  • deadline (int) – The number of seconds to wait to call callback
  • callback_method (method) – The callback method
Return type:

str

basic_nack

Specifies if the server supports basic.nack on the active connection.

Return type:bool
channel(on_open_callback, channel_number=None)

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.

Parameters:
  • on_open_callback (method) – The callback when the channel is opened
  • channel_number (int) – The channel number to use, defaults to the next available.
Return type:

pika.channel.Channel

close(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.

Parameters:
  • reply_code (int) – The code number for the close
  • reply_text (str) – The text reason for the close
connect()

Invoke if trying to reconnect to a RabbitMQ server. Constructing the Connection object should connect on its own.

consumer_cancel_notify

Specifies if the server supports consumer cancel notification on the active connection.

Return type:bool
exchange_exchange_bindings

Specifies if the active connection supports exchange to exchange bindings.

Return type:bool
is_closed

Returns a boolean reporting the current connection state.

is_closing

Returns a boolean reporting the current connection state.

is_open

Returns a boolean reporting the current connection state.

publisher_confirms

Specifies if the active connection can use publisher confirmations.

Return type:bool
remove_timeout(timeout_id)[source]

Remove the timeout from the IOLoop by the ID returned from add_timeout.

Return type:str
set_backpressure_multiplier(value=10)

Alter the backpressure multiplier value. We set this to 10 by default. This value is used to raise warnings and trigger the backpressure callback.

Parameters:value (int) – The multiplier value to set
Twisted Connection Adapter

Using Pika with a Twisted reactor.

Supports two methods of establishing the connection, using TwistedConnection or TwistedProtocolConnection. For details about each method, see the docstrings of the corresponding classes.

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 TwistedConnection.channel() and the TwistedChannel class for details.

class pika.adapters.twisted_connection.TwistedConnection(parameters=None, on_open_callback=None, on_open_error_callback=None, on_close_callback=None, stop_ioloop_on_close=False)[source]

A standard Pika connection adapter. You instantiate the class passing the connection parameters and the connected callback and when it gets called you can start using it.

The problem is that connection establishing is done using the blocking socket module. For instance, if the host you are connecting to is behind a misconfigured firewall that just drops packets, the whole process will freeze until the connection timeout passes. To work around that problem, use TwistedProtocolConnection, but read its docstring first.

Objects of this class get put in the Twisted reactor which will notify them when the socket connection becomes readable or writable, so apart from implementing the BaseConnection interface, they also provide Twisted’s IReadWriteDescriptor interface.

add_backpressure_callback(callback_method)

Call method “callback” when pika believes backpressure is being applied.

Parameters:callback_method (method) – The method to call
add_on_close_callback(callback_method)

Add a callback notification when the connection has closed. The callback will be passed the connection, the reply_code (int) and the reply_text (str), if sent by the remote server.

Parameters:callback_method (method) – Callback to call on close
add_on_connection_blocked_callback(callback_method)

Add a callback to be notified when RabbitMQ has sent a Connection.Blocked frame indicating that RabbitMQ is low on resources. Publishers can use this to voluntarily suspend publishing, instead of relying on back pressure throttling. The callback will be passed the Connection.Blocked method frame.

Parameters:callback_method (method) – Callback to call on Connection.Blocked
add_on_connection_unblocked_callback(callback_method)

Add a callback to be notified when RabbitMQ has sent a Connection.Unblocked frame letting publishers know it’s ok to start publishing again. The callback will be passed the Connection.Unblocked method frame.

Parameters:callback_method (method) – Callback to call on Connection.Unblocked
add_on_open_callback(callback_method)

Add a callback notification when the connection has opened.

Parameters:callback_method (method) – Callback to call when open
add_on_open_error_callback(callback_method, remove_default=True)

Add a callback notification when the connection can not be opened.

The callback method should accept the connection object that could not connect, and an optional error message.

Parameters:
  • callback_method (method) – Callback to call when can’t connect
  • remove_default (bool) – Remove default exception raising callback
add_timeout(deadline, callback_method)

Add the callback_method to the IOLoop timer to fire after deadline seconds. Returns a handle to the timeout

Parameters:
  • deadline (int) – The number of seconds to wait to call callback
  • callback_method (method) – The callback method
Return type:

str

basic_nack

Specifies if the server supports basic.nack on the active connection.

Return type:bool
channel(channel_number=None)[source]

Return a Deferred that fires with an instance of a wrapper around the Pika Channel class.

close(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.

Parameters:
  • reply_code (int) – The code number for the close
  • reply_text (str) – The text reason for the close
connect()

Invoke if trying to reconnect to a RabbitMQ server. Constructing the Connection object should connect on its own.

consumer_cancel_notify

Specifies if the server supports consumer cancel notification on the active connection.

Return type:bool
exchange_exchange_bindings

Specifies if the active connection supports exchange to exchange bindings.

Return type:bool
is_closed

Returns a boolean reporting the current connection state.

is_closing

Returns a boolean reporting the current connection state.

is_open

Returns a boolean reporting the current connection state.

publisher_confirms

Specifies if the active connection can use publisher confirmations.

Return type:bool
remove_timeout(timeout_id)

Remove the timeout from the IOLoop by the ID returned from add_timeout.

Return type:str
set_backpressure_multiplier(value=10)

Alter the backpressure multiplier value. We set this to 10 by default. This value is used to raise warnings and trigger the backpressure callback.

Parameters:value (int) – The multiplier value to set
class pika.adapters.twisted_connection.TwistedProtocolConnection(parameters)[source]

A hybrid between a Pika Connection and a Twisted Protocol. Allows using Twisted’s non-blocking connectTCP/connectSSL methods for connecting to the server.

It has one caveat: 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.

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.

add_backpressure_callback(callback_method)

Call method “callback” when pika believes backpressure is being applied.

Parameters:callback_method (method) – The method to call
add_on_close_callback(callback_method)

Add a callback notification when the connection has closed. The callback will be passed the connection, the reply_code (int) and the reply_text (str), if sent by the remote server.

Parameters:callback_method (method) – Callback to call on close
add_on_connection_blocked_callback(callback_method)

Add a callback to be notified when RabbitMQ has sent a Connection.Blocked frame indicating that RabbitMQ is low on resources. Publishers can use this to voluntarily suspend publishing, instead of relying on back pressure throttling. The callback will be passed the Connection.Blocked method frame.

Parameters:callback_method (method) – Callback to call on Connection.Blocked
add_on_connection_unblocked_callback(callback_method)

Add a callback to be notified when RabbitMQ has sent a Connection.Unblocked frame letting publishers know it’s ok to start publishing again. The callback will be passed the Connection.Unblocked method frame.

Parameters:callback_method (method) – Callback to call on Connection.Unblocked
add_on_open_callback(callback_method)

Add a callback notification when the connection has opened.

Parameters:callback_method (method) – Callback to call when open
add_on_open_error_callback(callback_method, remove_default=True)

Add a callback notification when the connection can not be opened.

The callback method should accept the connection object that could not connect, and an optional error message.

Parameters:
  • callback_method (method) – Callback to call when can’t connect
  • remove_default (bool) – Remove default exception raising callback
add_timeout(deadline, callback_method)

Add the callback_method to the IOLoop timer to fire after deadline seconds. Returns a handle to the timeout

Parameters:
  • deadline (int) – The number of seconds to wait to call callback
  • callback_method (method) – The callback method
Return type:

str

basic_nack

Specifies if the server supports basic.nack on the active connection.

Return type:bool
channel(channel_number=None)[source]

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.

Return a Deferred that fires with an instance of a wrapper around the Pika Channel class.

Parameters:channel_number (int) – The channel number to use, defaults to the next available.
close(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.

Parameters:
  • reply_code (int) – The code number for the close
  • reply_text (str) – The text reason for the close
consumer_cancel_notify

Specifies if the server supports consumer cancel notification on the active connection.

Return type:bool
exchange_exchange_bindings

Specifies if the active connection supports exchange to exchange bindings.

Return type:bool
is_closed

Returns a boolean reporting the current connection state.

is_closing

Returns a boolean reporting the current connection state.

is_open

Returns a boolean reporting the current connection state.

publisher_confirms

Specifies if the active connection can use publisher confirmations.

Return type:bool
remove_timeout(timeout_id)

Remove the timeout from the IOLoop by the ID returned from add_timeout.

Return type:str
set_backpressure_multiplier(value=10)

Alter the backpressure multiplier value. We set this to 10 by default. This value is used to raise warnings and trigger the backpressure callback.

Parameters:value (int) – The multiplier value to set
class pika.adapters.twisted_connection.TwistedChannel(channel)[source]

A wrapper wround 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.

The basic_consume method is wrapped in a special way, see its docstring for details.

basic_consume(*args, **kwargs)[source]

Consume from a server queue. Returns a Deferred that fires with a tuple: (queue_object, consumer_tag). The queue object is an instance of ClosableDeferredQueue, where data received from the queue will be stored. Clients should use its get() method to fetch individual message.

basic_publish(*args, **kwargs)[source]

Make sure the channel is not closed and then publish. Return a Deferred that fires with the result of the channel’s basic_publish.

queue_delete(*args, **kwargs)[source]

Wraps the method the same way all the others are wrapped, but removes the reference to the queue object after it gets deleted on the server.

Channel

The Channel class provides a wrapper for interacting with RabbitMQ implementing the methods and behaviors for an AMQP Channel.

Channel
class pika.channel.Channel(connection, channel_number, on_open_callback=None)[source]

A Channel is the primary communication method for interacting with RabbitMQ. It is recommended that you do not directly invoke the creation of a channel object in your application code but rather construct the a channel by calling the active connection’s channel() method.

add_callback(callback, replies, one_shot=True)[source]

Pass in a callback handler and a list replies from the RabbitMQ broker which you’d like the callback notified of. Callbacks should allow for the frame parameter to be passed in.

Parameters:
  • callback (method) – The method to call
  • replies (list) – The replies to get a callback for
  • one_shot (bool) – Only handle the first type callback
add_on_cancel_callback(callback)[source]

Pass a callback function that will be called when the basic_cancel is sent by the server. The callback function should receive a frame parameter.

Parameters:callback (method) – The method to call on callback
add_on_close_callback(callback)[source]

Pass a callback function that will be called when the channel is closed. The callback function will receive the channel, the reply_code (int) and the reply_text (int) sent by the server describing why the channel was closed.

Parameters:callback (method) – The method to call on callback
add_on_flow_callback(callback)[source]

Pass a callback function that will be called when Channel.Flow is called by the remote server. Note that newer versions of RabbitMQ will not issue this but instead use TCP backpressure

Parameters:callback (method) – The method to call on callback
add_on_return_callback(callback)[source]

Pass a callback function that will be called when basic_publish as sent a message that has been rejected and returned by the server.

Parameters:callback (method) –

The method to call on callback with the signature callback(channel, method, properties,

body), where

channel: pika.Channel method: pika.spec.Basic.Return properties: pika.spec.BasicProperties body: str, unicode, or bytes (python 3.x)

basic_ack(delivery_tag=0, multiple=False)[source]

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.

Parameters:
  • delivery-tag (int) – The server-assigned delivery tag
  • multiple (bool) – 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.
basic_cancel(callback=None, consumer_tag='', nowait=False)[source]

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.

Parameters:
  • callback (method) – Method to call for a Basic.CancelOk response
  • consumer_tag (str) – Identifier for the consumer
  • nowait (bool) – Do not expect a Basic.CancelOk response
Raises:

ValueError

basic_consume(consumer_callback, queue='', no_ack=False, exclusive=False, consumer_tag=None, arguments=None)[source]

Sends the AMQP command Basic.Consume to the broker and binds messages for the consumer_tag to the consumer callback. If you do not pass in a consumer_tag, one will be automatically generated for you. Returns the consumer tag.

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

Parameters:
  • consumer_callback (method) –

    The method to callback when consuming with the signature consumer_callback(channel, method, properties,

    body), where

    channel: pika.Channel method: pika.spec.Basic.Deliver properties: pika.spec.BasicProperties body: str, unicode, or bytes (python 3.x)

  • queue (str or unicode) – The queue to consume from
  • no_ack (bool) – Tell the broker to not expect a response
  • exclusive (bool) – Don’t allow other consumers on the queue
  • consumer_tag (str or unicode) – Specify your own consumer tag
  • arguments (dict) – Custom key/value pair arguments for the consume
Return type:

str

basic_get(callback=None, queue='', no_ack=False)[source]

Get a single message from the AMQP broker. 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. For more information on basic_get and its parameters, see:

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

Parameters:
  • callback (method) – The method to callback with a message that has the signature callback(channel, method, properties, body), where: channel: pika.Channel method: pika.spec.Basic.GetOk properties: pika.spec.BasicProperties body: str, unicode, or bytes (python 3.x)
  • queue (str or unicode) – The queue to get a message from
  • no_ack (bool) – Tell the broker to not expect a reply
basic_nack(delivery_tag=None, multiple=False, requeue=True)[source]

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.

Parameters:
  • delivery-tag (int) – The server-assigned delivery tag
  • multiple (bool) – 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.
  • requeue (bool) – 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.
basic_publish(exchange, routing_key, body, properties=None, mandatory=False, immediate=False)[source]

Publish to the channel with the given exchange, routing key and body. For more information on basic_publish and what the parameters do, see:

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

Parameters:
  • exchange (str or unicode) – The exchange to publish to
  • routing_key (str or unicode) – The routing key to bind on
  • body (str or unicode) – The message body
  • properties (pika.spec.BasicProperties) – Basic.properties
  • mandatory (bool) – The mandatory flag
  • immediate (bool) – The immediate flag
basic_qos(callback=None, prefetch_size=0, prefetch_count=0, all_channels=False)[source]

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.

Parameters:
  • callback (method) – The method to callback for Basic.QosOk response
  • prefetch_size (int) – 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.
  • prefetch_count (int) – 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.
  • all_channels (bool) – Should the QoS apply to all channels
basic_reject(delivery_tag, requeue=True)[source]

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.

Parameters:
  • delivery-tag (int) – The server-assigned delivery tag
  • requeue (bool) – 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

basic_recover(callback=None, requeue=False)[source]

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.

Parameters:
  • callback (method) – Method to call when receiving Basic.RecoverOk
  • requeue (bool) – 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.
close(reply_code=0, reply_text='Normal Shutdown')[source]

Will invoke a clean shutdown of the channel with the AMQP Broker.

Parameters:
  • reply_code (int) – The reply code to close the channel with
  • reply_text (str) – The reply text to close the channel with
confirm_delivery(callback=None, nowait=False)[source]

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/extensions.html#confirms
Parameters:
  • callback (method) – The callback for delivery confirmations
  • nowait (bool) – Do not send a reply frame (Confirm.SelectOk)
consumer_tags

Property method that returns a list of currently active consumers

Return type:list
exchange_bind(callback=None, destination=None, source=None, routing_key='', nowait=False, arguments=None)[source]

Bind an exchange to another exchange.

Parameters:
  • callback (method) – The method to call on Exchange.BindOk
  • destination (str or unicode) – The destination exchange to bind
  • source (str or unicode) – The source exchange to bind to
  • routing_key (str or unicode) – The routing key to bind on
  • nowait (bool) – Do not wait for an Exchange.BindOk
  • arguments (dict) – Custom key/value pair arguments for the binding
exchange_declare(callback=None, exchange=None, exchange_type='direct', passive=False, durable=False, auto_delete=False, internal=False, nowait=False, arguments=None, type=None)[source]

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).

Parameters:
  • callback (method) – Call this method on Exchange.DeclareOk
  • exchange (str or unicode sequence of these characters: letters, digits, hyphen, underscore, period, or colon.) – The exchange name consists of a non-empty
  • exchange_type (str) – The exchange type to use
  • passive (bool) – Perform a declare or just check to see if it exists
  • durable (bool) – Survive a reboot of RabbitMQ
  • auto_delete (bool) – Remove when no more queues are bound to it
  • internal (bool) – Can only be published to by other exchanges
  • nowait (bool) – Do not expect an Exchange.DeclareOk response
  • arguments (dict) – Custom key/value pair arguments for the exchange
  • type (str) – The deprecated exchange type parameter
exchange_delete(callback=None, exchange=None, if_unused=False, nowait=False)[source]

Delete the exchange.

Parameters:
  • callback (method) – The method to call on Exchange.DeleteOk
  • exchange (str or unicode) – The exchange name
  • if_unused (bool) – only delete if the exchange is unused
  • nowait (bool) – Do not wait for an Exchange.DeleteOk
exchange_unbind(callback=None, destination=None, source=None, routing_key='', nowait=False, arguments=None)[source]

Unbind an exchange from another exchange.

Parameters:
  • callback (method) – The method to call on Exchange.UnbindOk
  • destination (str or unicode) – The destination exchange to unbind
  • source (str or unicode) – The source exchange to unbind from
  • routing_key (str or unicode) – The routing key to unbind
  • nowait (bool) – Do not wait for an Exchange.UnbindOk
  • arguments (dict) – Custom key/value pair arguments for the binding
flow(callback, active)[source]

Turn Channel flow control off and on. Pass a callback to be notified of the response from the server. active is a bool. Callback should expect a bool in response indicating channel flow state. For more information, please reference:

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

Parameters:
  • callback (method) – The callback method
  • active (bool) – Turn flow on or off
is_closed

Returns True if the channel is closed.

Return type:bool
is_closing

Returns True if the channel is closing.

Return type:bool
is_open

Returns True if the channel is open.

Return type:bool
open()[source]

Open the channel

queue_bind(callback, queue, exchange, routing_key=None, nowait=False, arguments=None)[source]

Bind the queue to the specified exchange

Parameters:
  • callback (method) – The method to call on Queue.BindOk
  • queue (str or unicode) – The queue to bind to the exchange
  • exchange (str or unicode) – The source exchange to bind to
  • routing_key (str or unicode) – The routing key to bind on
  • nowait (bool) – Do not wait for a Queue.BindOk
  • arguments (dict) – Custom key/value pair arguments for the binding
queue_declare(callback, queue='', passive=False, durable=False, exclusive=False, auto_delete=False, nowait=False, arguments=None)[source]

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

Parameters:
  • callback (method) – The method to call on Queue.DeclareOk
  • queue (str or unicode) – The queue name
  • passive (bool) – Only check to see if the queue exists
  • durable (bool) – Survive reboots of the broker
  • exclusive (bool) – Only allow access by the current connection
  • auto_delete (bool) – Delete after consumer cancels or disconnects
  • nowait (bool) – Do not wait for a Queue.DeclareOk
  • arguments (dict) – Custom key/value arguments for the queue
queue_delete(callback=None, queue='', if_unused=False, if_empty=False, nowait=False)[source]

Delete a queue from the broker.

Parameters:
  • callback (method) – The method to call on Queue.DeleteOk
  • queue (str or unicode) – The queue to delete
  • if_unused (bool) – only delete if it’s unused
  • if_empty (bool) – only delete if the queue is empty
  • nowait (bool) – Do not wait for a Queue.DeleteOk
queue_purge(callback=None, queue='', nowait=False)[source]

Purge all of the messages from the specified queue

Parameters:
  • callback (method) – The method to call on Queue.PurgeOk
  • queue (str or unicode) – The queue to purge
  • nowait (bool) – Do not expect a Queue.PurgeOk response
queue_unbind(callback=None, queue='', exchange=None, routing_key=None, arguments=None)[source]

Unbind a queue from an exchange.

Parameters:
  • callback (method) – The method to call on Queue.UnbindOk
  • queue (str or unicode) – The queue to unbind from the exchange
  • exchange (str or unicode) – The source exchange to bind from
  • routing_key (str or unicode) – The routing key to unbind
  • arguments (dict) – Custom key/value pair arguments for the binding
tx_commit(callback=None)[source]

Commit a transaction

Parameters:callback (method) – The callback for delivery confirmations
tx_rollback(callback=None)[source]

Rollback a transaction.

Parameters:callback (method) – The callback for delivery confirmations
tx_select(callback=None)[source]

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.

Parameters:callback (method) – The callback for delivery confirmations

Connection

The Connection class implements the base behavior that all connection adapters extend.

class pika.connection.Connection(parameters=None, on_open_callback=None, on_open_error_callback=None, on_close_callback=None)[source]

This is the core class that implements communication with RabbitMQ. This class should not be invoked directly but rather through the use of an adapter such as SelectConnection or BlockingConnection.

Parameters:
  • parameters (pika.connection.Parameters) – Connection parameters
  • on_open_callback (method) – Called when the connection is opened
  • on_open_error_callback (method) – Called if the connection cant be opened
  • on_close_callback (method) – Called when the connection is closed
add_backpressure_callback(callback_method)[source]

Call method “callback” when pika believes backpressure is being applied.

Parameters:callback_method (method) – The method to call
add_on_close_callback(callback_method)[source]

Add a callback notification when the connection has closed. The callback will be passed the connection, the reply_code (int) and the reply_text (str), if sent by the remote server.

Parameters:callback_method (method) – Callback to call on close
add_on_connection_blocked_callback(callback_method)[source]

Add a callback to be notified when RabbitMQ has sent a Connection.Blocked frame indicating that RabbitMQ is low on resources. Publishers can use this to voluntarily suspend publishing, instead of relying on back pressure throttling. The callback will be passed the Connection.Blocked method frame.

Parameters:callback_method (method) – Callback to call on Connection.Blocked
add_on_connection_unblocked_callback(callback_method)[source]

Add a callback to be notified when RabbitMQ has sent a Connection.Unblocked frame letting publishers know it’s ok to start publishing again. The callback will be passed the Connection.Unblocked method frame.

Parameters:callback_method (method) – Callback to call on Connection.Unblocked
add_on_open_callback(callback_method)[source]

Add a callback notification when the connection has opened.

Parameters:callback_method (method) – Callback to call when open
add_on_open_error_callback(callback_method, remove_default=True)[source]

Add a callback notification when the connection can not be opened.

The callback method should accept the connection object that could not connect, and an optional error message.

Parameters:
  • callback_method (method) – Callback to call when can’t connect
  • remove_default (bool) – Remove default exception raising callback
add_timeout(deadline, callback_method)[source]

Adapters should override to call the callback after the specified number of seconds have elapsed, using a timer, or a thread, or similar.

Parameters:
  • deadline (int) – The number of seconds to wait to call callback
  • callback_method (method) – The callback method
basic_nack

Specifies if the server supports basic.nack on the active connection.

Return type:bool
channel(on_open_callback, channel_number=None)[source]

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.

Parameters:
  • on_open_callback (method) – The callback when the channel is opened
  • channel_number (int) – The channel number to use, defaults to the next available.
Return type:

pika.channel.Channel

close(reply_code=200, reply_text='Normal shutdown')[source]

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.

Parameters:
  • reply_code (int) – The code number for the close
  • reply_text (str) – The text reason for the close
connect()[source]

Invoke if trying to reconnect to a RabbitMQ server. Constructing the Connection object should connect on its own.

consumer_cancel_notify

Specifies if the server supports consumer cancel notification on the active connection.

Return type:bool
exchange_exchange_bindings

Specifies if the active connection supports exchange to exchange bindings.

Return type:bool
is_closed

Returns a boolean reporting the current connection state.

is_closing

Returns a boolean reporting the current connection state.

is_open

Returns a boolean reporting the current connection state.

publisher_confirms

Specifies if the active connection can use publisher confirmations.

Return type:bool
remove_timeout(callback_method)[source]

Adapters should override to call the callback after the specified number of seconds have elapsed, using a timer, or a thread, or similar.

Parameters:callback_method (method) – The callback to remove a timeout for
set_backpressure_multiplier(value=10)[source]

Alter the backpressure multiplier value. We set this to 10 by default. This value is used to raise warnings and trigger the backpressure callback.

Parameters:value (int) – The multiplier value to set

Authentication Credentials

The credentials classes are used to encapsulate all authentication information for the ConnectionParameters class.

The PlainCredentials class returns the properly formatted username and password to the Connection.

To authenticate with Pika, create a PlainCredentials object passing in the username and password and pass it as the credentials argument value to the ConnectionParameters object.

If you are using URLParameters you do not need a credentials object, one will automatically be created for you.

If you are looking to implement SSL certificate style authentication, you would extend the ExternalCredentials class implementing the required behavior.

PlainCredentials
class pika.credentials.PlainCredentials(username, password, erase_on_connect=False)[source]

A credentials object for the default authentication methodology with RabbitMQ.

If you do not pass in credentials to the ConnectionParameters object, it will create credentials for ‘guest’ with the password of ‘guest’.

If you pass True to erase_on_connect the credentials will not be stored in memory after the Connection attempt has been made.

Parameters:
  • username (str) – The username to authenticate with
  • password (str) – The password to authenticate with
  • erase_on_connect (bool) – erase credentials on connect.
erase_credentials()[source]

Called by Connection when it no longer needs the credentials

response_for(start)[source]

Validate that this type of authentication is supported

Parameters:start (spec.Connection.Start) – Connection.Start method
Return type:tuple(str|None, str|None)
ExternalCredentials
class pika.credentials.ExternalCredentials[source]

The ExternalCredentials class allows the connection to use EXTERNAL authentication, generally with a client SSL certificate.

erase_credentials()[source]

Called by Connection when it no longer needs the credentials

response_for(start)[source]

Validate that this type of authentication is supported

Parameters:start (spec.Connection.Start) – Connection.Start method
Return type:tuple(str or None, str or None)

Exceptions

Pika specific exceptions

exception pika.exceptions.AMQPChannelError[source]
exception pika.exceptions.AMQPConnectionError[source]
exception pika.exceptions.AMQPError[source]
exception pika.exceptions.AuthenticationError[source]
exception pika.exceptions.BodyTooLongError[source]
exception pika.exceptions.ChannelClosed[source]
exception pika.exceptions.ChannelError[source]
exception pika.exceptions.ConnectionClosed[source]
exception pika.exceptions.ConsumerCancelled[source]
exception pika.exceptions.DuplicateConsumerTag[source]
exception pika.exceptions.IncompatibleProtocolError[source]
exception pika.exceptions.InvalidChannelNumber[source]
exception pika.exceptions.InvalidFieldTypeException[source]
exception pika.exceptions.InvalidFrameError[source]
exception pika.exceptions.InvalidMaximumFrameSize[source]
exception pika.exceptions.InvalidMinimumFrameSize[source]
exception pika.exceptions.MethodNotImplemented[source]
exception pika.exceptions.NackError(messages)[source]

This exception is raised when a message published in publisher-acknowledgements mode is Nack’ed by the broker.

Used by BlockingChannel.

exception pika.exceptions.NoFreeChannels[source]
exception pika.exceptions.ProbableAccessDeniedError[source]
exception pika.exceptions.ProbableAuthenticationError[source]
exception pika.exceptions.ProtocolSyntaxError[source]
exception pika.exceptions.ProtocolVersionMismatch[source]
exception pika.exceptions.RecursionError[source]

The requested operation would result in unsupported recursion or reentrancy.

Used by BlockingConnection/BlockingChannel

exception pika.exceptions.ShortStringTooLong[source]
exception pika.exceptions.UnexpectedFrameError[source]
exception pika.exceptions.UnroutableError(messages)[source]

Exception containing one or more unroutable messages returned by broker via Basic.Return.

Used by BlockingChannel.

In publisher-acknowledgements mode, this is raised upon receipt of Basic.Ack from broker; in the event of Basic.Nack from broker, NackError is raised instead

exception pika.exceptions.UnspportedAMQPFieldException[source]

Deprecated version of UnsupportedAMQPFieldException

exception pika.exceptions.UnsupportedAMQPFieldException[source]

Connection Parameters

To maintain flexibility in how you specify the connection information required for your applications to properly connect to RabbitMQ, pika implements two classes for encapsulating the information, ConnectionParameters and URLParameters.

ConnectionParameters

The classic object for specifying all of the connection parameters required to connect to RabbitMQ, ConnectionParameters provides attributes for tweaking every possible connection option.

Example:

import pika

# Set the connection parameters to connect to rabbit-server1 on port 5672
# on the / virtual host using the username "guest" and password "guest"
credentials = pika.PlainCredentials('guest', 'guest')
parameters = pika.ConnectionParameters('rabbit-server1',
                                       5672,
                                       '/',
                                       credentials)
class pika.connection.ConnectionParameters(host=None, port=None, virtual_host=None, credentials=None, channel_max=None, frame_max=None, heartbeat_interval=None, ssl=None, ssl_options=None, connection_attempts=None, retry_delay=None, socket_timeout=None, locale=None, backpressure_detection=None)[source]

Connection parameters object that is passed into the connection adapter upon construction.

Parameters:
  • host (str) – Hostname or IP Address to connect to
  • port (int) – TCP port to connect to
  • virtual_host (str) – RabbitMQ virtual host to use
  • credentials (pika.credentials.Credentials) – auth credentials
  • channel_max (int) – Maximum number of channels to allow
  • frame_max (int) – The maximum byte size for an AMQP frame
  • heartbeat_interval (int) – How often to send heartbeats
  • ssl (bool) – Enable SSL
  • ssl_options (dict) – Arguments passed to ssl.wrap_socket as
  • connection_attempts (int) – Maximum number of retry attempts
  • retry_delay (int|float) – Time to wait in seconds, before the next
  • socket_timeout (int|float) – Use for high latency networks
  • locale (str) – Set the locale value
  • backpressure_detection (bool) – Toggle backpressure detection
URLParameters

The URLParameters class allows you to pass in an AMQP URL when creating the object and supports the host, port, virtual host, ssl, username and password in the base URL and other options are passed in via query parameters.

Example:

import pika

# Set the connection parameters to connect to rabbit-server1 on port 5672
# on the / virtual host using the username "guest" and password "guest"
parameters = pika.URLParameters('amqp://guest:guest@rabbit-server1:5672/%2F')
class pika.connection.URLParameters(url)[source]

Connect to RabbitMQ via an AMQP URL in the format:

amqp://username:password@host:port/<virtual_host>[?query-string]

Ensure that the virtual host is URI encoded when specified. For example if you are using the default “/” virtual host, the value should be %2f.

Valid query string values are:

  • backpressure_detection:

    Toggle backpressure detection, possible values are t or f

  • channel_max:

    Override the default maximum channel count value

  • connection_attempts:

    Specify how many times pika should try and reconnect before it gives up

  • frame_max:

    Override the default maximum frame size for communication

  • heartbeat_interval:

    Specify the number of seconds between heartbeat frames to ensure that the link between RabbitMQ and your application is up

  • locale:

    Override the default en_US locale value

  • ssl:

    Toggle SSL, possible values are t, f

  • ssl_options:

    Arguments passed to ssl.wrap_socket()

  • retry_delay:

    The number of seconds to sleep before attempting to connect on connection failure.

  • socket_timeout:

    Override low level socket timeout value

Parameters:url (str) – The AMQP URL to connect to

pika.spec

class pika.spec.Connection[source]
INDEX = 10
NAME = 'Connection'
class Start(version_major=0, version_minor=9, server_properties=None, mechanisms='PLAIN', locales='en_US')[source]
INDEX = 655370
NAME = 'Connection.Start'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Connection.StartOk(client_properties=None, mechanism='PLAIN', response=None, locale='en_US')[source]
INDEX = 655371
NAME = 'Connection.StartOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Connection.Secure(challenge=None)[source]
INDEX = 655380
NAME = 'Connection.Secure'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Connection.SecureOk(response=None)[source]
INDEX = 655381
NAME = 'Connection.SecureOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Connection.Tune(channel_max=0, frame_max=0, heartbeat=0)[source]
INDEX = 655390
NAME = 'Connection.Tune'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Connection.TuneOk(channel_max=0, frame_max=0, heartbeat=0)[source]
INDEX = 655391
NAME = 'Connection.TuneOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Connection.Open(virtual_host='/', capabilities='', insist=False)[source]
INDEX = 655400
NAME = 'Connection.Open'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Connection.OpenOk(known_hosts='')[source]
INDEX = 655401
NAME = 'Connection.OpenOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Connection.Close(reply_code=None, reply_text='', class_id=None, method_id=None)[source]
INDEX = 655410
NAME = 'Connection.Close'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Connection.CloseOk[source]
INDEX = 655411
NAME = 'Connection.CloseOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Connection.Blocked(reason='')[source]
INDEX = 655420
NAME = 'Connection.Blocked'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Connection.Unblocked[source]
INDEX = 655421
NAME = 'Connection.Unblocked'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class pika.spec.Channel[source]
INDEX = 20
NAME = 'Channel'
class Open(out_of_band='')[source]
INDEX = 1310730
NAME = 'Channel.Open'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Channel.OpenOk(channel_id='')[source]
INDEX = 1310731
NAME = 'Channel.OpenOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Channel.Flow(active=None)[source]
INDEX = 1310740
NAME = 'Channel.Flow'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Channel.FlowOk(active=None)[source]
INDEX = 1310741
NAME = 'Channel.FlowOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Channel.Close(reply_code=None, reply_text='', class_id=None, method_id=None)[source]
INDEX = 1310760
NAME = 'Channel.Close'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Channel.CloseOk[source]
INDEX = 1310761
NAME = 'Channel.CloseOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class pika.spec.Access[source]
INDEX = 30
NAME = 'Access'
class Request(realm='/data', exclusive=False, passive=True, active=True, write=True, read=True)[source]
INDEX = 1966090
NAME = 'Access.Request'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Access.RequestOk(ticket=1)[source]
INDEX = 1966091
NAME = 'Access.RequestOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class pika.spec.Exchange[source]
INDEX = 40
NAME = 'Exchange'
class Declare(ticket=0, exchange=None, type='direct', passive=False, durable=False, auto_delete=False, internal=False, nowait=False, arguments={})[source]
INDEX = 2621450
NAME = 'Exchange.Declare'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Exchange.DeclareOk[source]
INDEX = 2621451
NAME = 'Exchange.DeclareOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Exchange.Delete(ticket=0, exchange=None, if_unused=False, nowait=False)[source]
INDEX = 2621460
NAME = 'Exchange.Delete'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Exchange.DeleteOk[source]
INDEX = 2621461
NAME = 'Exchange.DeleteOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Exchange.Bind(ticket=0, destination=None, source=None, routing_key='', nowait=False, arguments={})[source]
INDEX = 2621470
NAME = 'Exchange.Bind'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Exchange.BindOk[source]
INDEX = 2621471
NAME = 'Exchange.BindOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Exchange.Unbind(ticket=0, destination=None, source=None, routing_key='', nowait=False, arguments={})[source]
INDEX = 2621480
NAME = 'Exchange.Unbind'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Exchange.UnbindOk[source]
INDEX = 2621491
NAME = 'Exchange.UnbindOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class pika.spec.Queue[source]
INDEX = 50
NAME = 'Queue'
class Declare(ticket=0, queue='', passive=False, durable=False, exclusive=False, auto_delete=False, nowait=False, arguments={})[source]
INDEX = 3276810
NAME = 'Queue.Declare'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Queue.DeclareOk(queue=None, message_count=None, consumer_count=None)[source]
INDEX = 3276811
NAME = 'Queue.DeclareOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Queue.Bind(ticket=0, queue='', exchange=None, routing_key='', nowait=False, arguments={})[source]
INDEX = 3276820
NAME = 'Queue.Bind'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Queue.BindOk[source]
INDEX = 3276821
NAME = 'Queue.BindOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Queue.Purge(ticket=0, queue='', nowait=False)[source]
INDEX = 3276830
NAME = 'Queue.Purge'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Queue.PurgeOk(message_count=None)[source]
INDEX = 3276831
NAME = 'Queue.PurgeOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Queue.Delete(ticket=0, queue='', if_unused=False, if_empty=False, nowait=False)[source]
INDEX = 3276840
NAME = 'Queue.Delete'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Queue.DeleteOk(message_count=None)[source]
INDEX = 3276841
NAME = 'Queue.DeleteOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Queue.Unbind(ticket=0, queue='', exchange=None, routing_key='', arguments={})[source]
INDEX = 3276850
NAME = 'Queue.Unbind'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Queue.UnbindOk[source]
INDEX = 3276851
NAME = 'Queue.UnbindOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class pika.spec.Basic[source]
INDEX = 60
NAME = 'Basic'
class Qos(prefetch_size=0, prefetch_count=0, global_=False)[source]
INDEX = 3932170
NAME = 'Basic.Qos'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Basic.QosOk[source]
INDEX = 3932171
NAME = 'Basic.QosOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Basic.Consume(ticket=0, queue='', consumer_tag='', no_local=False, no_ack=False, exclusive=False, nowait=False, arguments={})[source]
INDEX = 3932180
NAME = 'Basic.Consume'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Basic.ConsumeOk(consumer_tag=None)[source]
INDEX = 3932181
NAME = 'Basic.ConsumeOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Basic.Cancel(consumer_tag=None, nowait=False)[source]
INDEX = 3932190
NAME = 'Basic.Cancel'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Basic.CancelOk(consumer_tag=None)[source]
INDEX = 3932191
NAME = 'Basic.CancelOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Basic.Publish(ticket=0, exchange='', routing_key='', mandatory=False, immediate=False)[source]
INDEX = 3932200
NAME = 'Basic.Publish'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Basic.Return(reply_code=None, reply_text='', exchange=None, routing_key=None)[source]
INDEX = 3932210
NAME = 'Basic.Return'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Basic.Deliver(consumer_tag=None, delivery_tag=None, redelivered=False, exchange=None, routing_key=None)[source]
INDEX = 3932220
NAME = 'Basic.Deliver'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Basic.Get(ticket=0, queue='', no_ack=False)[source]
INDEX = 3932230
NAME = 'Basic.Get'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Basic.GetOk(delivery_tag=None, redelivered=False, exchange=None, routing_key=None, message_count=None)[source]
INDEX = 3932231
NAME = 'Basic.GetOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Basic.GetEmpty(cluster_id='')[source]
INDEX = 3932232
NAME = 'Basic.GetEmpty'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Basic.Ack(delivery_tag=0, multiple=False)[source]
INDEX = 3932240
NAME = 'Basic.Ack'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Basic.Reject(delivery_tag=None, requeue=True)[source]
INDEX = 3932250
NAME = 'Basic.Reject'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Basic.RecoverAsync(requeue=False)[source]
INDEX = 3932260
NAME = 'Basic.RecoverAsync'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Basic.Recover(requeue=False)[source]
INDEX = 3932270
NAME = 'Basic.Recover'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Basic.RecoverOk[source]
INDEX = 3932271
NAME = 'Basic.RecoverOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Basic.Nack(delivery_tag=0, multiple=False, requeue=True)[source]
INDEX = 3932280
NAME = 'Basic.Nack'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class pika.spec.Tx[source]
INDEX = 90
NAME = 'Tx'
class Select[source]
INDEX = 5898250
NAME = 'Tx.Select'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Tx.SelectOk[source]
INDEX = 5898251
NAME = 'Tx.SelectOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Tx.Commit[source]
INDEX = 5898260
NAME = 'Tx.Commit'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Tx.CommitOk[source]
INDEX = 5898261
NAME = 'Tx.CommitOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Tx.Rollback[source]
INDEX = 5898270
NAME = 'Tx.Rollback'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Tx.RollbackOk[source]
INDEX = 5898271
NAME = 'Tx.RollbackOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class pika.spec.Confirm[source]
INDEX = 85
NAME = 'Confirm'
class Select(nowait=False)[source]
INDEX = 5570570
NAME = 'Confirm.Select'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class Confirm.SelectOk[source]
INDEX = 5570571
NAME = 'Confirm.SelectOk'
synchronous
decode(encoded, offset=0)[source]
encode()[source]
get_body()

Return the message body if it is set.

Return type:str|unicode
get_properties()

Return the properties if they are set.

Return type:pika.frame.Properties
class pika.spec.BasicProperties(content_type=None, content_encoding=None, headers=None, delivery_mode=None, priority=None, correlation_id=None, reply_to=None, expiration=None, message_id=None, timestamp=None, type=None, user_id=None, app_id=None, cluster_id=None)[source]
CLASS

alias of Basic

INDEX = 60
NAME = 'BasicProperties'
FLAG_CONTENT_TYPE = 32768
FLAG_CONTENT_ENCODING = 16384
FLAG_HEADERS = 8192
FLAG_DELIVERY_MODE = 4096
FLAG_PRIORITY = 2048
FLAG_CORRELATION_ID = 1024
FLAG_REPLY_TO = 512
FLAG_EXPIRATION = 256
FLAG_MESSAGE_ID = 128
FLAG_TIMESTAMP = 64
FLAG_TYPE = 32
FLAG_USER_ID = 16
FLAG_APP_ID = 8
FLAG_CLUSTER_ID = 4
decode(encoded, offset=0)[source]
encode()[source]
pika.spec.has_content(methodNumber)[source]

Usage Examples

Pika has various methods of use, between the synchronous BlockingConnection adapter and the various asynchronous connection adapter. The following examples illustrate the various ways that you can use Pika in your projects.

Using URLParameters

Pika has two methods of encapsulating the data that lets it know how to connect to RabbitMQ, pika.connection.ConnectionParameters and pika.connection.URLParameters.

Note

If you’re connecting to RabbitMQ on localhost on port 5672, with the default virtual host of / and the default username and password of guest and guest, you do not need to specify connection parameters when connecting.

Using pika.connection.URLParameters is an easy way to minimize the variables required to connect to RabbitMQ and supports all of the directives that pika.connection.ConnectionParameters supports.

The following is the format for the URLParameters connection value:

scheme://username:password@host:port/virtual_host?key=value&key=value

As you can see, by default, the scheme (amqp, amqps), username, password, host, port and virtual host make up the core of the URL and any other parameter is passed in as query string values.

Example Connection URLS

The default connection URL connects to the / virtual host as guest using the guest password on localhost port 5672. Note the forwardslash in the URL is encoded to %2F:

amqp://guest:guest@localhost:5672/%2F

Connect to a host rabbit1 as the user www-data using the password rabbit_pwd on the virtual host web_messages:

amqp://www-data:rabbit_pwd@rabbit1/web_messages

Connecting via SSL is pretty easy too. To connect via SSL for the previous example, simply change the scheme to amqps. If you do not specify a port, Pika will use the default SSL port of 5671:

amqps://www-data:rabbit_pwd@rabbit1/web_messages

If you’re looking to tweak other parameters, such as enabling heartbeats, simply add the key/value pair as a query string value. The following builds upon the SSL connection, enabling heartbeats every 30 seconds:

amqps://www-data:rabbit_pwd@rabbit1/web_messages?heartbeat_interval=30

Options that are available as query string values:

  • backpressure_detection: Pass in a value of t to enable backpressure detection, it is disabled by default.

  • channel_max: Alter the default channel maximum by passing in a 32-bit integer value here

  • connection_attempts: Alter the default of 1 connection attempt by passing in an integer value here [1].

  • frame_max: Alter the default frame maximum size value by passing in a long integer value [2].

  • heartbeat_interval: Pass a value greater than zero to enable heartbeats between the server and your application. The integer value you pass here will be the number of seconds between heartbeats.

  • locale: Set the locale of the client using underscore delimited posix Locale code in ll_CC format (en_US, pt_BR, de_DE).

  • retry_delay: The number of seconds to wait before attempting to reconnect on a failed connection, if connection_attempts is > 0.

  • socket_timeout: Change the default socket timeout duration from 0.25 seconds to another integer or float value. Adjust with caution.

  • ssl_options: A url encoded dict of values for the SSL connection. The available keys are:
    • ca_certs
    • cert_reqs
    • certfile
    • keyfile
    • ssl_version

For an information on what the ssl_options can be set to reference the official Python documentation. Here is an example of setting the client certificate and key:

amqp://www-data:rabbit_pwd@rabbit1/web_messages?heartbeat_interval=30&ssl_options=%7B%27keyfile%27%3A+%27%2Fetc%2Fssl%2Fmykey.pem%27%2C+%27certfile%27%3A+%27%2Fetc%2Fssl%2Fmycert.pem%27%7D

The following example demonstrates how to generate the ssl_options string with Python’s urllib:

import urllib
urllib.urlencode({'ssl_options': {'certfile': '/etc/ssl/mycert.pem', 'keyfile': '/etc/ssl/mykey.pem'}})

Footnotes

[1]The pika.adapters.blocking_connection.BlockingConnection adapter does not respect the connection_attempts parameter.
[2]The AMQP specification states that a server can reject a request for a frame size larger than the value it passes during content negotiation.

Connecting to RabbitMQ with Callback-Passing Style

When you connect to RabbitMQ with an asynchronous adapter, you are writing event oriented code. The connection adapter will block on the IOLoop that is watching to see when pika should read data from and write data to RabbitMQ. Because you’re now blocking on the IOLoop, you will receive callback notifications when specific events happen.

Example Code

In the example, there are three steps that take place:

  1. Setup the connection to RabbitMQ

  2. Start the IOLoop

  3. Once connected, the on_open method will be called by Pika with a handle to the connection. In this method, a new channel will be opened on the connection.

  4. Once the channel is opened, you can do your other actions, whether they be publishing messages, consuming messages or other RabbitMQ related activities.:

    import pika
    
    # Step #3
    def on_open(connection):
        connection.channel(on_channel_open)
    
    # Step #4
    def on_channel_open(channel):
        channel.basic_publish('exchange_name',
                              'routing_key',
                              'Test Message',
                              pika.BasicProperties(content_type='text/plain',
                                                   type='example'))
    
    # Step #1: Connect to RabbitMQ
    connection = pika.SelectConnection(on_open_callback=on_open)
    
    try:
        # Step #2 - Block on the IOLoop
        connection.ioloop.start()
    
    # Catch a Keyboard Interrupt to make sure that the connection is closed cleanly
    except KeyboardInterrupt:
    
        # Gracefully close the connection
        connection.close()
    
        # Start the IOLoop again so Pika can communicate, it will stop on its own when the connection is closed
        connection.ioloop.start()
    

Using the Blocking Connection to get a message from RabbitMQ

The BlockingChannel.basic_get method will return a tuple with the members.

If the server returns a message, the first item in the tuple will be a pika.spec.Basic.GetOk object with the current message count, the redelivered flag, the routing key that was used to put the message in the queue, and the exchange the message was published to. The second item will be a BasicProperties object and the third will be the message body.

If the server did not return a message a tuple of None, None, None will be returned.

Example of getting a message and acknowledging it:

import pika

connection = pika.BlockingConnection()
channel = connection.channel()
method_frame, header_frame, body = channel.basic_get('test')
if method_frame:
    print method_frame, header_frame, body
    channel.basic_ack(method_frame.delivery_tag)
else:
    print 'No message returned'

Using the Blocking Connection to consume messages from RabbitMQ

The BlockingChannel.basic_consume method assign a callback method to be called every time that RabbitMQ delivers messages to your consuming application.

When pika calls your method, it will pass in the channel, a pika.spec.Basic.Deliver object with the delivery tag, the redelivered flag, the routing key that was used to put the message in the queue, and the exchange the message was published to. The third argument will be a pika.spec.BasicProperties object and the last will be the message body.

Example of consuming messages and acknowledging them:

import pika


def on_message(channel, method_frame, header_frame, body):
    print method_frame.delivery_tag
    print body
    print
    channel.basic_ack(delivery_tag=method_frame.delivery_tag)


connection = pika.BlockingConnection()
channel = connection.channel()
channel.basic_consume(on_message, 'test')
try:
    channel.start_consuming()
except KeyboardInterrupt:
    channel.stop_consuming()
connection.close()

Using the BlockingChannel.consume generator to consume messages

The BlockingChannel.consume method is a generator that will return a tuple of method, properties and body.

When you escape out of the loop, be sure to call consumer.cancel() to return any unprocessed messages.

Example of consuming messages and acknowledging them:

import pika

connection = pika.BlockingConnection()
channel = connection.channel()

# Get ten messages and break out
for method_frame, properties, body in channel.consume('test'):

    # Display the message parts
    print method_frame
    print properties
    print body

    # Acknowledge the message
    channel.basic_ack(method_frame.delivery_tag)

    # Escape out of the loop after 10 messages
    if method_frame.delivery_tag == 10:
        break

# Cancel the consumer and return any pending messages
requeued_messages = channel.cancel()
print 'Requeued %i messages' % requeued_messages

# Close the channel and the connection
channel.close()
connection.close()

If you have pending messages in the test queue, your output should look something like:

(pika)gmr-0x02:pika gmr$ python blocking_nack.py
<Basic.Deliver(['consumer_tag=ctag1.0', 'redelivered=True', 'routing_key=test', 'delivery_tag=1', 'exchange=test'])>
<BasicProperties(['delivery_mode=1', 'content_type=text/plain'])>
Hello World!
<Basic.Deliver(['consumer_tag=ctag1.0', 'redelivered=True', 'routing_key=test', 'delivery_tag=2', 'exchange=test'])>
<BasicProperties(['delivery_mode=1', 'content_type=text/plain'])>
Hello World!
<Basic.Deliver(['consumer_tag=ctag1.0', 'redelivered=True', 'routing_key=test', 'delivery_tag=3', 'exchange=test'])>
<BasicProperties(['delivery_mode=1', 'content_type=text/plain'])>
Hello World!
<Basic.Deliver(['consumer_tag=ctag1.0', 'redelivered=True', 'routing_key=test', 'delivery_tag=4', 'exchange=test'])>
<BasicProperties(['delivery_mode=1', 'content_type=text/plain'])>
Hello World!
<Basic.Deliver(['consumer_tag=ctag1.0', 'redelivered=True', 'routing_key=test', 'delivery_tag=5', 'exchange=test'])>
<BasicProperties(['delivery_mode=1', 'content_type=text/plain'])>
Hello World!
<Basic.Deliver(['consumer_tag=ctag1.0', 'redelivered=True', 'routing_key=test', 'delivery_tag=6', 'exchange=test'])>
<BasicProperties(['delivery_mode=1', 'content_type=text/plain'])>
Hello World!
<Basic.Deliver(['consumer_tag=ctag1.0', 'redelivered=True', 'routing_key=test', 'delivery_tag=7', 'exchange=test'])>
<BasicProperties(['delivery_mode=1', 'content_type=text/plain'])>
Hello World!
<Basic.Deliver(['consumer_tag=ctag1.0', 'redelivered=True', 'routing_key=test', 'delivery_tag=8', 'exchange=test'])>
<BasicProperties(['delivery_mode=1', 'content_type=text/plain'])>
Hello World!
<Basic.Deliver(['consumer_tag=ctag1.0', 'redelivered=True', 'routing_key=test', 'delivery_tag=9', 'exchange=test'])>
<BasicProperties(['delivery_mode=1', 'content_type=text/plain'])>
Hello World!
<Basic.Deliver(['consumer_tag=ctag1.0', 'redelivered=True', 'routing_key=test', 'delivery_tag=10', 'exchange=test'])>
<BasicProperties(['delivery_mode=1', 'content_type=text/plain'])>
Hello World!
Requeued 1894 messages

Comparing Message Publishing with BlockingConnection and SelectConnection

For those doing simple, non-asynchronous programing, pika.adapters.blocking_connection.BlockingConnection() proves to be the easiest way to get up and running with Pika to publish messages.

In the following example, a connection is made to RabbitMQ listening to port 5672 on localhost using the username guest and password guest and virtual host /. Once connected, a channel is opened and a message is published to the test_exchange exchange using the test_routing_key routing key. The BasicProperties value passed in sets the message to delivery mode 1 (non-persisted) with a content-type of text/plain. Once the message is published, the connection is closed:

import pika

parameters = pika.URLParameters('amqp://guest:guest@localhost:5672/%2F')

connection = pika.BlockingConnection(parameters)

channel = connection.channel()

channel.basic_publish('test_exchange',
                      'test_routing_key',
                      'message body value',
                      pika.BasicProperties(content_type='text/plain',
                                           delivery_mode=1))

connection.close()

In contrast, using pika.adapters.select_connection.SelectConnection() and the other asynchronous adapters is more complicated and less pythonic, but when used with other asynchronous services can have tremendous performance improvements. In the following code example, all of the same parameters and values are used as were used in the previous example:

import pika

# Step #3
def on_open(connection):

    connection.channel(on_channel_open)

# Step #4
def on_channel_open(channel):

    channel.basic_publish('test_exchange',
                            'test_routing_key',
                            'message body value',
                            pika.BasicProperties(content_type='text/plain',
                                                 delivery_mode=1))

    connection.close()

# Step #1: Connect to RabbitMQ
parameters = pika.URLParameters('amqp://guest:guest@localhost:5672/%2F')

connection = pika.SelectConnection(parameters=parameters,
                                   on_open_callback=on_open)

try:

    # Step #2 - Block on the IOLoop
    connection.ioloop.start()

# Catch a Keyboard Interrupt to make sure that the connection is closed cleanly
except KeyboardInterrupt:

    # Gracefully close the connection
    connection.close()

    # Start the IOLoop again so Pika can communicate, it will stop on its own when the connection is closed
    connection.ioloop.start()

Using Delivery Confirmations with the BlockingConnection

The following code demonstrates how to turn on delivery confirmations with the BlockingConnection and how to check for confirmation from RabbitMQ:

import pika

# Open a connection to RabbitMQ on localhost using all default parameters
connection = pika.BlockingConnection()

# Open the channel
channel = connection.channel()

# Declare the queue
channel.queue_declare(queue="test", durable=True, exclusive=False, auto_delete=False)

# Turn on delivery confirmations
channel.confirm_delivery()

# Send a message
if channel.basic_publish(exchange='test',
                         routing_key='test',
                         body='Hello World!',
                         properties=pika.BasicProperties(content_type='text/plain',
                                                         delivery_mode=1)):
    print 'Message publish was confirmed'
else:
    print 'Message could not be confirmed'

Ensuring message delivery with the mandatory flag

The following example demonstrates how to check if a message is delivered by setting the mandatory flag and checking the return result when using the BlockingConnection:

import pika

# Open a connection to RabbitMQ on localhost using all default parameters
connection = pika.BlockingConnection()

# Open the channel
channel = connection.channel()

# Declare the queue
channel.queue_declare(queue="test", durable=True, exclusive=False, auto_delete=False)

# Enabled delivery confirmations
channel.confirm_delivery()

# Send a message
if channel.basic_publish(exchange='test',
                         routing_key='test',
                         body='Hello World!',
                         properties=pika.BasicProperties(content_type='text/plain',
                                                         delivery_mode=1),
                         mandatory=True):
    print 'Message was published'
else:
    print 'Message was returned'

Asynchronous consumer example

The following example implements a consumer that will respond to RPC commands sent from RabbitMQ. For example, it will reconnect if RabbitMQ closes the connection and will shutdown if RabbitMQ cancels the consumer or closes the channel. While it may look intimidating, each method is very short and represents a individual actions that a consumer can do.

consumer.py:

# -*- coding: utf-8 -*-

import logging
import pika

LOG_FORMAT = ('%(levelname) -10s %(asctime)s %(name) -30s %(funcName) '
              '-35s %(lineno) -5d: %(message)s')
LOGGER = logging.getLogger(__name__)


class ExampleConsumer(object):
    """This is an example consumer that will handle unexpected interactions
    with RabbitMQ such as channel and connection closures.

    If RabbitMQ closes the connection, it will reopen it. You should
    look at the output, as there are limited reasons why the connection may
    be closed, which usually are tied to permission related issues or
    socket timeouts.

    If the channel is closed, it will indicate a problem with one of the
    commands that were issued and that should surface in the output as well.

    """
    EXCHANGE = 'message'
    EXCHANGE_TYPE = 'topic'
    QUEUE = 'text'
    ROUTING_KEY = 'example.text'

    def __init__(self, amqp_url):
        """Create a new instance of the consumer class, passing in the AMQP
        URL used to connect to RabbitMQ.

        :param str amqp_url: The AMQP url to connect with

        """
        self._connection = None
        self._channel = None
        self._closing = False
        self._consumer_tag = None
        self._url = amqp_url

    def connect(self):
        """This method connects to RabbitMQ, returning the connection handle.
        When the connection is established, the on_connection_open method
        will be invoked by pika.

        :rtype: pika.SelectConnection

        """
        LOGGER.info('Connecting to %s', self._url)
        return pika.SelectConnection(pika.URLParameters(self._url),
                                     self.on_connection_open,
                                     stop_ioloop_on_close=False)

    def on_connection_open(self, unused_connection):
        """This method is called by pika once the connection to RabbitMQ has
        been established. It passes the handle to the connection object in
        case we need it, but in this case, we'll just mark it unused.

        :type unused_connection: pika.SelectConnection

        """
        LOGGER.info('Connection opened')
        self.add_on_connection_close_callback()
        self.open_channel()

    def add_on_connection_close_callback(self):
        """This method adds an on close callback that will be invoked by pika
        when RabbitMQ closes the connection to the publisher unexpectedly.

        """
        LOGGER.info('Adding connection close callback')
        self._connection.add_on_close_callback(self.on_connection_closed)

    def on_connection_closed(self, connection, reply_code, reply_text):
        """This method is invoked by pika when the connection to RabbitMQ is
        closed unexpectedly. Since it is unexpected, we will reconnect to
        RabbitMQ if it disconnects.

        :param pika.connection.Connection connection: The closed connection obj
        :param int reply_code: The server provided reply_code if given
        :param str reply_text: The server provided reply_text if given

        """
        self._channel = None
        if self._closing:
            self._connection.ioloop.stop()
        else:
            LOGGER.warning('Connection closed, reopening in 5 seconds: (%s) %s',
                           reply_code, reply_text)
            self._connection.add_timeout(5, self.reconnect)

    def reconnect(self):
        """Will be invoked by the IOLoop timer if the connection is
        closed. See the on_connection_closed method.

        """
        # This is the old connection IOLoop instance, stop its ioloop
        self._connection.ioloop.stop()

        if not self._closing:

            # Create a new connection
            self._connection = self.connect()

            # There is now a new connection, needs a new ioloop to run
            self._connection.ioloop.start()

    def open_channel(self):
        """Open a new channel with RabbitMQ by issuing the Channel.Open RPC
        command. When RabbitMQ responds that the channel is open, the
        on_channel_open callback will be invoked by pika.

        """
        LOGGER.info('Creating a new channel')
        self._connection.channel(on_open_callback=self.on_channel_open)

    def on_channel_open(self, channel):
        """This method is invoked by pika when the channel has been opened.
        The channel object is passed in so we can make use of it.

        Since the channel is now open, we'll declare the exchange to use.

        :param pika.channel.Channel channel: The channel object

        """
        LOGGER.info('Channel opened')
        self._channel = channel
        self.add_on_channel_close_callback()
        self.setup_exchange(self.EXCHANGE)

    def add_on_channel_close_callback(self):
        """This method tells pika to call the on_channel_closed method if
        RabbitMQ unexpectedly closes the channel.

        """
        LOGGER.info('Adding channel close callback')
        self._channel.add_on_close_callback(self.on_channel_closed)

    def on_channel_closed(self, channel, reply_code, reply_text):
        """Invoked by pika when RabbitMQ unexpectedly closes the channel.
        Channels are usually closed if you attempt to do something that
        violates the protocol, such as re-declare an exchange or queue with
        different parameters. In this case, we'll close the connection
        to shutdown the object.

        :param pika.channel.Channel: The closed channel
        :param int reply_code: The numeric reason the channel was closed
        :param str reply_text: The text reason the channel was closed

        """
        LOGGER.warning('Channel %i was closed: (%s) %s',
                       channel, reply_code, reply_text)
        self._connection.close()

    def setup_exchange(self, exchange_name):
        """Setup the exchange on RabbitMQ by invoking the Exchange.Declare RPC
        command. When it is complete, the on_exchange_declareok method will
        be invoked by pika.

        :param str|unicode exchange_name: The name of the exchange to declare

        """
        LOGGER.info('Declaring exchange %s', exchange_name)
        self._channel.exchange_declare(self.on_exchange_declareok,
                                       exchange_name,
                                       self.EXCHANGE_TYPE)

    def on_exchange_declareok(self, unused_frame):
        """Invoked by pika when RabbitMQ has finished the Exchange.Declare RPC
        command.

        :param pika.Frame.Method unused_frame: Exchange.DeclareOk response frame

        """
        LOGGER.info('Exchange declared')
        self.setup_queue(self.QUEUE)

    def setup_queue(self, queue_name):
        """Setup the queue on RabbitMQ by invoking the Queue.Declare RPC
        command. When it is complete, the on_queue_declareok method will
        be invoked by pika.

        :param str|unicode queue_name: The name of the queue to declare.

        """
        LOGGER.info('Declaring queue %s', queue_name)
        self._channel.queue_declare(self.on_queue_declareok, queue_name)

    def on_queue_declareok(self, method_frame):
        """Method invoked by pika when the Queue.Declare RPC call made in
        setup_queue has completed. In this method we will bind the queue
        and exchange together with the routing key by issuing the Queue.Bind
        RPC command. When this command is complete, the on_bindok method will
        be invoked by pika.

        :param pika.frame.Method method_frame: The Queue.DeclareOk frame

        """
        LOGGER.info('Binding %s to %s with %s',
                    self.EXCHANGE, self.QUEUE, self.ROUTING_KEY)
        self._channel.queue_bind(self.on_bindok, self.QUEUE,
                                 self.EXCHANGE, self.ROUTING_KEY)

    def on_bindok(self, unused_frame):
        """Invoked by pika when the Queue.Bind method has completed. At this
        point we will start consuming messages by calling start_consuming
        which will invoke the needed RPC commands to start the process.

        :param pika.frame.Method unused_frame: The Queue.BindOk response frame

        """
        LOGGER.info('Queue bound')
        self.start_consuming()

    def start_consuming(self):
        """This method sets up the consumer by first calling
        add_on_cancel_callback so that the object is notified if RabbitMQ
        cancels the consumer. It then issues the Basic.Consume RPC command
        which returns the consumer tag that is used to uniquely identify the
        consumer with RabbitMQ. We keep the value to use it when we want to
        cancel consuming. The on_message method is passed in as a callback pika
        will invoke when a message is fully received.

        """
        LOGGER.info('Issuing consumer related RPC commands')
        self.add_on_cancel_callback()
        self._consumer_tag = self._channel.basic_consume(self.on_message,
                                                         self.QUEUE)

    def add_on_cancel_callback(self):
        """Add a callback that will be invoked if RabbitMQ cancels the consumer
        for some reason. If RabbitMQ does cancel the consumer,
        on_consumer_cancelled will be invoked by pika.

        """
        LOGGER.info('Adding consumer cancellation callback')
        self._channel.add_on_cancel_callback(self.on_consumer_cancelled)

    def on_consumer_cancelled(self, method_frame):
        """Invoked by pika when RabbitMQ sends a Basic.Cancel for a consumer
        receiving messages.

        :param pika.frame.Method method_frame: The Basic.Cancel frame

        """
        LOGGER.info('Consumer was cancelled remotely, shutting down: %r',
                    method_frame)
        if self._channel:
            self._channel.close()

    def on_message(self, unused_channel, basic_deliver, properties, body):
        """Invoked by pika when a message is delivered from RabbitMQ. The
        channel is passed for your convenience. The basic_deliver object that
        is passed in carries the exchange, routing key, delivery tag and
        a redelivered flag for the message. The properties passed in is an
        instance of BasicProperties with the message properties and the body
        is the message that was sent.

        :param pika.channel.Channel unused_channel: The channel object
        :param pika.Spec.Basic.Deliver: basic_deliver method
        :param pika.Spec.BasicProperties: properties
        :param str|unicode body: The message body

        """
        LOGGER.info('Received message # %s from %s: %s',
                    basic_deliver.delivery_tag, properties.app_id, body)
        self.acknowledge_message(basic_deliver.delivery_tag)

    def acknowledge_message(self, delivery_tag):
        """Acknowledge the message delivery from RabbitMQ by sending a
        Basic.Ack RPC method for the delivery tag.

        :param int delivery_tag: The delivery tag from the Basic.Deliver frame

        """
        LOGGER.info('Acknowledging message %s', delivery_tag)
        self._channel.basic_ack(delivery_tag)

    def stop_consuming(self):
        """Tell RabbitMQ that you would like to stop consuming by sending the
        Basic.Cancel RPC command.

        """
        if self._channel:
            LOGGER.info('Sending a Basic.Cancel RPC command to RabbitMQ')
            self._channel.basic_cancel(self.on_cancelok, self._consumer_tag)

    def on_cancelok(self, unused_frame):
        """This method is invoked by pika when RabbitMQ acknowledges the
        cancellation of a consumer. At this point we will close the channel.
        This will invoke the on_channel_closed method once the channel has been
        closed, which will in-turn close the connection.

        :param pika.frame.Method unused_frame: The Basic.CancelOk frame

        """
        LOGGER.info('RabbitMQ acknowledged the cancellation of the consumer')
        self.close_channel()

    def close_channel(self):
        """Call to close the channel with RabbitMQ cleanly by issuing the
        Channel.Close RPC command.

        """
        LOGGER.info('Closing the channel')
        self._channel.close()

    def run(self):
        """Run the example consumer by connecting to RabbitMQ and then
        starting the IOLoop to block and allow the SelectConnection to operate.

        """
        self._connection = self.connect()
        self._connection.ioloop.start()

    def stop(self):
        """Cleanly shutdown the connection to RabbitMQ by stopping the consumer
        with RabbitMQ. When RabbitMQ confirms the cancellation, on_cancelok
        will be invoked by pika, which will then closing the channel and
        connection. The IOLoop is started again because this method is invoked
        when CTRL-C is pressed raising a KeyboardInterrupt exception. This
        exception stops the IOLoop which needs to be running for pika to
        communicate with RabbitMQ. All of the commands issued prior to starting
        the IOLoop will be buffered but not processed.

        """
        LOGGER.info('Stopping')
        self._closing = True
        self.stop_consuming()
        self._connection.ioloop.start()
        LOGGER.info('Stopped')

    def close_connection(self):
        """This method closes the connection to RabbitMQ."""
        LOGGER.info('Closing connection')
        self._connection.close()


def main():
    logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
    example = ExampleConsumer('amqp://guest:guest@localhost:5672/%2F')
    try:
        example.run()
    except KeyboardInterrupt:
        example.stop()


if __name__ == '__main__':
    main()

Asynchronous publisher example

The following example implements a publisher that will respond to RPC commands sent from RabbitMQ and uses delivery confirmations. It will reconnect if RabbitMQ closes the connection and will shutdown if RabbitMQ closes the channel. While it may look intimidating, each method is very short and represents a individual actions that a publisher can do.

publisher.py:

# -*- coding: utf-8 -*-

import logging
import pika
import json

LOG_FORMAT = ('%(levelname) -10s %(asctime)s %(name) -30s %(funcName) '
              '-35s %(lineno) -5d: %(message)s')
LOGGER = logging.getLogger(__name__)


class ExamplePublisher(object):
    """This is an example publisher that will handle unexpected interactions
    with RabbitMQ such as channel and connection closures.

    If RabbitMQ closes the connection, it will reopen it. You should
    look at the output, as there are limited reasons why the connection may
    be closed, which usually are tied to permission related issues or
    socket timeouts.

    It uses delivery confirmations and illustrates one way to keep track of
    messages that have been sent and if they've been confirmed by RabbitMQ.

    """
    EXCHANGE = 'message'
    EXCHANGE_TYPE = 'topic'
    PUBLISH_INTERVAL = 1
    QUEUE = 'text'
    ROUTING_KEY = 'example.text'

    def __init__(self, amqp_url):
        """Setup the example publisher object, passing in the URL we will use
        to connect to RabbitMQ.

        :param str amqp_url: The URL for connecting to RabbitMQ

        """
        self._connection = None
        self._channel = None
        self._deliveries = []
        self._acked = 0
        self._nacked = 0
        self._message_number = 0
        self._stopping = False
        self._url = amqp_url
        self._closing = False

    def connect(self):
        """This method connects to RabbitMQ, returning the connection handle.
        When the connection is established, the on_connection_open method
        will be invoked by pika. If you want the reconnection to work, make
        sure you set stop_ioloop_on_close to False, which is not the default
        behavior of this adapter.

        :rtype: pika.SelectConnection

        """
        LOGGER.info('Connecting to %s', self._url)
        return pika.SelectConnection(pika.URLParameters(self._url),
                                     self.on_connection_open,
                                     stop_ioloop_on_close=False)

    def on_connection_open(self, unused_connection):
        """This method is called by pika once the connection to RabbitMQ has
        been established. It passes the handle to the connection object in
        case we need it, but in this case, we'll just mark it unused.

        :type unused_connection: pika.SelectConnection

        """
        LOGGER.info('Connection opened')
        self.add_on_connection_close_callback()
        self.open_channel()

    def add_on_connection_close_callback(self):
        """This method adds an on close callback that will be invoked by pika
        when RabbitMQ closes the connection to the publisher unexpectedly.

        """
        LOGGER.info('Adding connection close callback')
        self._connection.add_on_close_callback(self.on_connection_closed)

    def on_connection_closed(self, connection, reply_code, reply_text):
        """This method is invoked by pika when the connection to RabbitMQ is
        closed unexpectedly. Since it is unexpected, we will reconnect to
        RabbitMQ if it disconnects.

        :param pika.connection.Connection connection: The closed connection obj
        :param int reply_code: The server provided reply_code if given
        :param str reply_text: The server provided reply_text if given

        """
        self._channel = None
        if self._closing:
            self._connection.ioloop.stop()
        else:
            LOGGER.warning('Connection closed, reopening in 5 seconds: (%s) %s',
                           reply_code, reply_text)
            self._connection.add_timeout(5, self.reconnect)

    def reconnect(self):
        """Will be invoked by the IOLoop timer if the connection is
        closed. See the on_connection_closed method.

        """
        self._deliveries = []
        self._acked = 0
        self._nacked = 0
        self._message_number = 0

        # This is the old connection IOLoop instance, stop its ioloop
        self._connection.ioloop.stop()

        # Create a new connection
        self._connection = self.connect()

        # There is now a new connection, needs a new ioloop to run
        self._connection.ioloop.start()

    def open_channel(self):
        """This method will open a new channel with RabbitMQ by issuing the
        Channel.Open RPC command. When RabbitMQ confirms the channel is open
        by sending the Channel.OpenOK RPC reply, the on_channel_open method
        will be invoked.

        """
        LOGGER.info('Creating a new channel')
        self._connection.channel(on_open_callback=self.on_channel_open)

    def on_channel_open(self, channel):
        """This method is invoked by pika when the channel has been opened.
        The channel object is passed in so we can make use of it.

        Since the channel is now open, we'll declare the exchange to use.

        :param pika.channel.Channel channel: The channel object

        """
        LOGGER.info('Channel opened')
        self._channel = channel
        self.add_on_channel_close_callback()
        self.setup_exchange(self.EXCHANGE)

    def add_on_channel_close_callback(self):
        """This method tells pika to call the on_channel_closed method if
        RabbitMQ unexpectedly closes the channel.

        """
        LOGGER.info('Adding channel close callback')
        self._channel.add_on_close_callback(self.on_channel_closed)

    def on_channel_closed(self, channel, reply_code, reply_text):
        """Invoked by pika when RabbitMQ unexpectedly closes the channel.
        Channels are usually closed if you attempt to do something that
        violates the protocol, such as re-declare an exchange or queue with
        different parameters. In this case, we'll close the connection
        to shutdown the object.

        :param pika.channel.Channel: The closed channel
        :param int reply_code: The numeric reason the channel was closed
        :param str reply_text: The text reason the channel was closed

        """
        LOGGER.warning('Channel was closed: (%s) %s', reply_code, reply_text)
        if not self._closing:
            self._connection.close()

    def setup_exchange(self, exchange_name):
        """Setup the exchange on RabbitMQ by invoking the Exchange.Declare RPC
        command. When it is complete, the on_exchange_declareok method will
        be invoked by pika.

        :param str|unicode exchange_name: The name of the exchange to declare

        """
        LOGGER.info('Declaring exchange %s', exchange_name)
        self._channel.exchange_declare(self.on_exchange_declareok,
                                       exchange_name,
                                       self.EXCHANGE_TYPE)

    def on_exchange_declareok(self, unused_frame):
        """Invoked by pika when RabbitMQ has finished the Exchange.Declare RPC
        command.

        :param pika.Frame.Method unused_frame: Exchange.DeclareOk response frame

        """
        LOGGER.info('Exchange declared')
        self.setup_queue(self.QUEUE)

    def setup_queue(self, queue_name):
        """Setup the queue on RabbitMQ by invoking the Queue.Declare RPC
        command. When it is complete, the on_queue_declareok method will
        be invoked by pika.

        :param str|unicode queue_name: The name of the queue to declare.

        """
        LOGGER.info('Declaring queue %s', queue_name)
        self._channel.queue_declare(self.on_queue_declareok, queue_name)

    def on_queue_declareok(self, method_frame):
        """Method invoked by pika when the Queue.Declare RPC call made in
        setup_queue has completed. In this method we will bind the queue
        and exchange together with the routing key by issuing the Queue.Bind
        RPC command. When this command is complete, the on_bindok method will
        be invoked by pika.

        :param pika.frame.Method method_frame: The Queue.DeclareOk frame

        """
        LOGGER.info('Binding %s to %s with %s',
                    self.EXCHANGE, self.QUEUE, self.ROUTING_KEY)
        self._channel.queue_bind(self.on_bindok, self.QUEUE,
                                 self.EXCHANGE, self.ROUTING_KEY)

    def on_bindok(self, unused_frame):
        """This method is invoked by pika when it receives the Queue.BindOk
        response from RabbitMQ. Since we know we're now setup and bound, it's
        time to start publishing."""
        LOGGER.info('Queue bound')
        self.start_publishing()

    def start_publishing(self):
        """This method will enable delivery confirmations and schedule the
        first message to be sent to RabbitMQ

        """
        LOGGER.info('Issuing consumer related RPC commands')
        self.enable_delivery_confirmations()
        self.schedule_next_message()

    def enable_delivery_confirmations(self):
        """Send the Confirm.Select RPC method to RabbitMQ to enable delivery
        confirmations on the channel. The only way to turn this off is to close
        the channel and create a new one.

        When the message is confirmed from RabbitMQ, the
        on_delivery_confirmation method will be invoked passing in a Basic.Ack
        or Basic.Nack method from RabbitMQ that will indicate which messages it
        is confirming or rejecting.

        """
        LOGGER.info('Issuing Confirm.Select RPC command')
        self._channel.confirm_delivery(self.on_delivery_confirmation)

    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

        """
        confirmation_type = method_frame.method.NAME.split('.')[1].lower()
        LOGGER.info('Received %s for delivery tag: %i',
                    confirmation_type,
                    method_frame.method.delivery_tag)
        if confirmation_type == 'ack':
            self._acked += 1
        elif confirmation_type == 'nack':
            self._nacked += 1
        self._deliveries.remove(method_frame.method.delivery_tag)
        LOGGER.info('Published %i messages, %i have yet to be confirmed, '
                    '%i were acked and %i were nacked',
                    self._message_number, len(self._deliveries),
                    self._acked, self._nacked)

    def schedule_next_message(self):
        """If we are not closing our connection to RabbitMQ, schedule another
        message to be delivered in PUBLISH_INTERVAL seconds.

        """
        if self._stopping:
            return
        LOGGER.info('Scheduling next message for %0.1f seconds',
                    self.PUBLISH_INTERVAL)
        self._connection.add_timeout(self.PUBLISH_INTERVAL,
                                     self.publish_message)

    def publish_message(self):
        """If the class is not stopping, publish a message to RabbitMQ,
        appending a list of deliveries with the message number that was sent.
        This list will be used to check for delivery confirmations in the
        on_delivery_confirmations method.

        Once the message has been sent, schedule another message to be sent.
        The main reason I put scheduling in was just so you can get a good idea
        of how the process is flowing by slowing down and speeding up the
        delivery intervals by changing the PUBLISH_INTERVAL constant in the
        class.

        """
        if self._stopping:
            return

        message = {u'مفتاح': u' قيمة',
                   u'键': u'值',
                   u'キー': u'値'}
        properties = pika.BasicProperties(app_id='example-publisher',
                                          content_type='application/json',
                                          headers=message)

        self._channel.basic_publish(self.EXCHANGE, self.ROUTING_KEY,
                                    json.dumps(message, ensure_ascii=False),
                                    properties)
        self._message_number += 1
        self._deliveries.append(self._message_number)
        LOGGER.info('Published message # %i', self._message_number)
        self.schedule_next_message()

    def close_channel(self):
        """Invoke this command to close the channel with RabbitMQ by sending
        the Channel.Close RPC command.

        """
        LOGGER.info('Closing the channel')
        if self._channel:
            self._channel.close()

    def run(self):
        """Run the example code by connecting and then starting the IOLoop.

        """
        self._connection = self.connect()
        self._connection.ioloop.start()

    def stop(self):
        """Stop the example by closing the channel and connection. We
        set a flag here so that we stop scheduling new messages to be
        published. The IOLoop is started because this method is
        invoked by the Try/Catch below when KeyboardInterrupt is caught.
        Starting the IOLoop again will allow the publisher to cleanly
        disconnect from RabbitMQ.

        """
        LOGGER.info('Stopping')
        self._stopping = True
        self.close_channel()
        self.close_connection()
        self._connection.ioloop.start()
        LOGGER.info('Stopped')

    def close_connection(self):
        """This method closes the connection to RabbitMQ."""
        LOGGER.info('Closing connection')
        self._closing = True
        self._connection.close()


def main():
    logging.basicConfig(level=logging.DEBUG, format=LOG_FORMAT)

    # Connect to localhost:5672 as guest with the password guest and virtual host "/" (%2F)
    example = ExamplePublisher('amqp://guest:guest@localhost:5672/%2F?connection_attempts=3&heartbeat_interval=3600')
    try:
        example.run()
    except KeyboardInterrupt:
        example.stop()

if __name__ == '__main__':
    main()

Twisted Consumer Example

Example of writing a consumer using the Twisted connection adapter:

# -*- coding:utf-8 -*-

import pika
from pika import exceptions
from pika.adapters import twisted_connection
from twisted.internet import defer, reactor, protocol,task


@defer.inlineCallbacks
def run(connection):

    channel = yield connection.channel()

    exchange = yield channel.exchange_declare(exchange='topic_link',type='topic')

    queue = yield channel.queue_declare(queue='hello', auto_delete=False, exclusive=False)

    yield channel.queue_bind(exchange='topic_link',queue='hello',routing_key='hello.world')

    yield channel.basic_qos(prefetch_count=1)

    queue_object, consumer_tag = yield channel.basic_consume(queue='hello',no_ack=False)

    l = task.LoopingCall(read, queue_object)

    l.start(0.01)


@defer.inlineCallbacks
def read(queue_object):

    ch,method,properties,body = yield queue_object.get()

    if body:
        print body

    yield ch.basic_ack(delivery_tag=method.delivery_tag)


parameters = pika.ConnectionParameters()
cc = protocol.ClientCreator(reactor, twisted_connection.TwistedProtocolConnection, parameters)
d = cc.connectTCP('hostname', 5672)
d.addCallback(lambda protocol: protocol.ready)
d.addCallback(run)
reactor.run()

Tornado Consumer

The following example implements a consumer using the Tornado adapter for the Tornado framework that will respond to RPC commands sent from RabbitMQ. For example, it will reconnect if RabbitMQ closes the connection and will shutdown if RabbitMQ cancels the consumer or closes the channel. While it may look intimidating, each method is very short and represents a individual actions that a consumer can do.

consumer.py:

from pika import adapters
import pika
import logging

LOG_FORMAT = ('%(levelname) -10s %(asctime)s %(name) -30s %(funcName) '
              '-35s %(lineno) -5d: %(message)s')
LOGGER = logging.getLogger(__name__)


class ExampleConsumer(object):
    """This is an example consumer that will handle unexpected interactions
    with RabbitMQ such as channel and connection closures.

    If RabbitMQ closes the connection, it will reopen it. You should
    look at the output, as there are limited reasons why the connection may
    be closed, which usually are tied to permission related issues or
    socket timeouts.

    If the channel is closed, it will indicate a problem with one of the
    commands that were issued and that should surface in the output as well.

    """
    EXCHANGE = 'message'
    EXCHANGE_TYPE = 'topic'
    QUEUE = 'text'
    ROUTING_KEY = 'example.text'

    def __init__(self, amqp_url):
        """Create a new instance of the consumer class, passing in the AMQP
        URL used to connect to RabbitMQ.

        :param str amqp_url: The AMQP url to connect with

        """
        self._connection = None
        self._channel = None
        self._closing = False
        self._consumer_tag = None
        self._url = amqp_url

    def connect(self):
        """This method connects to RabbitMQ, returning the connection handle.
        When the connection is established, the on_connection_open method
        will be invoked by pika.

        :rtype: pika.SelectConnection

        """
        LOGGER.info('Connecting to %s', self._url)
        return adapters.TornadoConnection(pika.URLParameters(self._url),
                                          self.on_connection_open)

    def close_connection(self):
        """This method closes the connection to RabbitMQ."""
        LOGGER.info('Closing connection')
        self._connection.close()

    def add_on_connection_close_callback(self):
        """This method adds an on close callback that will be invoked by pika
        when RabbitMQ closes the connection to the publisher unexpectedly.

        """
        LOGGER.info('Adding connection close callback')
        self._connection.add_on_close_callback(self.on_connection_closed)

    def on_connection_closed(self, connection, reply_code, reply_text):
        """This method is invoked by pika when the connection to RabbitMQ is
        closed unexpectedly. Since it is unexpected, we will reconnect to
        RabbitMQ if it disconnects.

        :param pika.connection.Connection connection: The closed connection obj
        :param int reply_code: The server provided reply_code if given
        :param str reply_text: The server provided reply_text if given

        """
        self._channel = None
        if self._closing:
            self._connection.ioloop.stop()
        else:
            LOGGER.warning('Connection closed, reopening in 5 seconds: (%s) %s',
                           reply_code, reply_text)
            self._connection.add_timeout(5, self.reconnect)

    def on_connection_open(self, unused_connection):
        """This method is called by pika once the connection to RabbitMQ has
        been established. It passes the handle to the connection object in
        case we need it, but in this case, we'll just mark it unused.

        :type unused_connection: pika.SelectConnection

        """
        LOGGER.info('Connection opened')
        self.add_on_connection_close_callback()
        self.open_channel()

    def reconnect(self):
        """Will be invoked by the IOLoop timer if the connection is
        closed. See the on_connection_closed method.

        """
        if not self._closing:

            # Create a new connection
            self._connection = self.connect()

    def add_on_channel_close_callback(self):
        """This method tells pika to call the on_channel_closed method if
        RabbitMQ unexpectedly closes the channel.

        """
        LOGGER.info('Adding channel close callback')
        self._channel.add_on_close_callback(self.on_channel_closed)

    def on_channel_closed(self, channel, reply_code, reply_text):
        """Invoked by pika when RabbitMQ unexpectedly closes the channel.
        Channels are usually closed if you attempt to do something that
        violates the protocol, such as re-declare an exchange or queue with
        different parameters. In this case, we'll close the connection
        to shutdown the object.

        :param pika.channel.Channel: The closed channel
        :param int reply_code: The numeric reason the channel was closed
        :param str reply_text: The text reason the channel was closed

        """
        LOGGER.warning('Channel %i was closed: (%s) %s',
                       channel, reply_code, reply_text)
        self._connection.close()

    def on_channel_open(self, channel):
        """This method is invoked by pika when the channel has been opened.
        The channel object is passed in so we can make use of it.

        Since the channel is now open, we'll declare the exchange to use.

        :param pika.channel.Channel channel: The channel object

        """
        LOGGER.info('Channel opened')
        self._channel = channel
        self.add_on_channel_close_callback()
        self.setup_exchange(self.EXCHANGE)

    def setup_exchange(self, exchange_name):
        """Setup the exchange on RabbitMQ by invoking the Exchange.Declare RPC
        command. When it is complete, the on_exchange_declareok method will
        be invoked by pika.

        :param str|unicode exchange_name: The name of the exchange to declare

        """
        LOGGER.info('Declaring exchange %s', exchange_name)
        self._channel.exchange_declare(self.on_exchange_declareok,
                                       exchange_name,
                                       self.EXCHANGE_TYPE)

    def on_exchange_declareok(self, unused_frame):
        """Invoked by pika when RabbitMQ has finished the Exchange.Declare RPC
        command.

        :param pika.Frame.Method unused_frame: Exchange.DeclareOk response frame

        """
        LOGGER.info('Exchange declared')
        self.setup_queue(self.QUEUE)

    def setup_queue(self, queue_name):
        """Setup the queue on RabbitMQ by invoking the Queue.Declare RPC
        command. When it is complete, the on_queue_declareok method will
        be invoked by pika.

        :param str|unicode queue_name: The name of the queue to declare.

        """
        LOGGER.info('Declaring queue %s', queue_name)
        self._channel.queue_declare(self.on_queue_declareok, queue_name)

    def on_queue_declareok(self, method_frame):
        """Method invoked by pika when the Queue.Declare RPC call made in
        setup_queue has completed. In this method we will bind the queue
        and exchange together with the routing key by issuing the Queue.Bind
        RPC command. When this command is complete, the on_bindok method will
        be invoked by pika.

        :param pika.frame.Method method_frame: The Queue.DeclareOk frame

        """
        LOGGER.info('Binding %s to %s with %s',
                    self.EXCHANGE, self.QUEUE, self.ROUTING_KEY)
        self._channel.queue_bind(self.on_bindok, self.QUEUE,
                                 self.EXCHANGE, self.ROUTING_KEY)

    def add_on_cancel_callback(self):
        """Add a callback that will be invoked if RabbitMQ cancels the consumer
        for some reason. If RabbitMQ does cancel the consumer,
        on_consumer_cancelled will be invoked by pika.

        """
        LOGGER.info('Adding consumer cancellation callback')
        self._channel.add_on_cancel_callback(self.on_consumer_cancelled)

    def on_consumer_cancelled(self, method_frame):
        """Invoked by pika when RabbitMQ sends a Basic.Cancel for a consumer
        receiving messages.

        :param pika.frame.Method method_frame: The Basic.Cancel frame

        """
        LOGGER.info('Consumer was cancelled remotely, shutting down: %r',
                    method_frame)
        if self._channel:
            self._channel.close()

    def acknowledge_message(self, delivery_tag):
        """Acknowledge the message delivery from RabbitMQ by sending a
        Basic.Ack RPC method for the delivery tag.

        :param int delivery_tag: The delivery tag from the Basic.Deliver frame

        """
        LOGGER.info('Acknowledging message %s', delivery_tag)
        self._channel.basic_ack(delivery_tag)

    def on_message(self, unused_channel, basic_deliver, properties, body):
        """Invoked by pika when a message is delivered from RabbitMQ. The
        channel is passed for your convenience. The basic_deliver object that
        is passed in carries the exchange, routing key, delivery tag and
        a redelivered flag for the message. The properties passed in is an
        instance of BasicProperties with the message properties and the body
        is the message that was sent.

        :param pika.channel.Channel unused_channel: The channel object
        :param pika.Spec.Basic.Deliver: basic_deliver method
        :param pika.Spec.BasicProperties: properties
        :param str|unicode body: The message body

        """
        LOGGER.info('Received message # %s from %s: %s',
                    basic_deliver.delivery_tag, properties.app_id, body)
        self.acknowledge_message(basic_deliver.delivery_tag)

    def on_cancelok(self, unused_frame):
        """This method is invoked by pika when RabbitMQ acknowledges the
        cancellation of a consumer. At this point we will close the channel.
        This will invoke the on_channel_closed method once the channel has been
        closed, which will in-turn close the connection.

        :param pika.frame.Method unused_frame: The Basic.CancelOk frame

        """
        LOGGER.info('RabbitMQ acknowledged the cancellation of the consumer')
        self.close_channel()

    def stop_consuming(self):
        """Tell RabbitMQ that you would like to stop consuming by sending the
        Basic.Cancel RPC command.

        """
        if self._channel:
            LOGGER.info('Sending a Basic.Cancel RPC command to RabbitMQ')
            self._channel.basic_cancel(self.on_cancelok, self._consumer_tag)

    def start_consuming(self):
        """This method sets up the consumer by first calling
        add_on_cancel_callback so that the object is notified if RabbitMQ
        cancels the consumer. It then issues the Basic.Consume RPC command
        which returns the consumer tag that is used to uniquely identify the
        consumer with RabbitMQ. We keep the value to use it when we want to
        cancel consuming. The on_message method is passed in as a callback pika
        will invoke when a message is fully received.

        """
        LOGGER.info('Issuing consumer related RPC commands')
        self.add_on_cancel_callback()
        self._consumer_tag = self._channel.basic_consume(self.on_message,
                                                         self.QUEUE)

    def on_bindok(self, unused_frame):
        """Invoked by pika when the Queue.Bind method has completed. At this
        point we will start consuming messages by calling start_consuming
        which will invoke the needed RPC commands to start the process.

        :param pika.frame.Method unused_frame: The Queue.BindOk response frame

        """
        LOGGER.info('Queue bound')
        self.start_consuming()

    def close_channel(self):
        """Call to close the channel with RabbitMQ cleanly by issuing the
        Channel.Close RPC command.

        """
        LOGGER.info('Closing the channel')
        self._channel.close()

    def open_channel(self):
        """Open a new channel with RabbitMQ by issuing the Channel.Open RPC
        command. When RabbitMQ responds that the channel is open, the
        on_channel_open callback will be invoked by pika.

        """
        LOGGER.info('Creating a new channel')
        self._connection.channel(on_open_callback=self.on_channel_open)

    def run(self):
        """Run the example consumer by connecting to RabbitMQ and then
        starting the IOLoop to block and allow the SelectConnection to operate.

        """
        self._connection = self.connect()
        self._connection.ioloop.start()

    def stop(self):
        """Cleanly shutdown the connection to RabbitMQ by stopping the consumer
        with RabbitMQ. When RabbitMQ confirms the cancellation, on_cancelok
        will be invoked by pika, which will then closing the channel and
        connection. The IOLoop is started again because this method is invoked
        when CTRL-C is pressed raising a KeyboardInterrupt exception. This
        exception stops the IOLoop which needs to be running for pika to
        communicate with RabbitMQ. All of the commands issued prior to starting
        the IOLoop will be buffered but not processed.

        """
        LOGGER.info('Stopping')
        self._closing = True
        self.stop_consuming()
        self._connection.ioloop.start()
        LOGGER.info('Stopped')


def main():
    logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
    example = ExampleConsumer('amqp://guest:guest@localhost:5672/%2F')
    try:
        example.run()
    except KeyboardInterrupt:
        example.stop()


if __name__ == '__main__':
    main()

Frequently Asked Questions

Contributors

The following people have directly contributes code by way of new features and/or bug fixes to Pika:

  • Gavin M. Roy
  • Tony Garnock-Jones
  • Vitaly Kruglikov
  • Michael Laing
  • Marek Majkowski
  • Jan Urbański
  • Brian K. Jones
  • Ask Solem
  • ml
  • Will
  • atatsu
  • Fredrik Svensson
  • Pedro Abranches
  • Kyösti Herrala
  • Erik Andersson
  • Charles Law
  • Alex Chandel
  • Tristan Penman
  • Raphaël De Giusti
  • Jozef Van Eenbergen
  • Josh Braegger
  • Jason J. W. Williams
  • James Mutton
  • Cenk Alti
  • Asko Soukka
  • Antti Haapala
  • Anton Ryzhov
  • cellscape
  • cacovsky
  • bra-fsn
  • ateska
  • Roey Berman
  • Robert Weidlich
  • Riccardo Cirimelli
  • Perttu Ranta-aho
  • Pau Gargallo
  • Kane
  • Kamil Kisiel
  • Jonty Wareing
  • Jonathan Kirsch
  • Jacek ‘Forger’ Całusiński
  • Garth Williamson
  • Erik Olof Gunnar Andersson
  • David Strauss
  • Anton V. Yanchenko
  • Alexey Myasnikov
  • Alessandro Tagliapietra
  • Adam Flynn
  • skftn
  • saarni
  • pavlobaron
  • nonleaf
  • markcf
  • george y
  • eivanov
  • bstemshorn
  • a-tal
  • Yang Yang
  • Stuart Longland
  • Sigurd Høgsbro
  • Sean Dwyer
  • Samuel Stauffer
  • Roberto Decurnex
  • Rikard Hultén
  • Richard Boulton
  • Ralf Nyren
  • Qi Fan
  • Peter Magnusson
  • Pankrat
  • Olivier Le Thanh Duong
  • Njal Karevoll
  • Milan Skuhra
  • Mik Kocikowski
  • Michael Kenney
  • Mark Unsworth
  • Luca Wehrstedt
  • Laurent Eschenauer
  • Lars van de Kerkhof
  • Kyösti Herrala
  • Juhyeong Park
  • JuhaS
  • Josh Hansen
  • Jorge Puente Sarrín
  • Jeff Tang
  • Jeff Fein-Worton
  • Jeff
  • Hunter Morris
  • Guruprasad
  • Garrett Cooper
  • Frank Slaughter
  • Dustin Koupal
  • Bjorn Sandberg
  • Axel Eirola
  • Andrew Smith
  • Andrew Grigorev
  • Andrew
  • Allard Hoeve
  • A.Shaposhnikov

Contributors listed by commit count.

Version History

0.10.0 2015-09-02

  • LibevConnection: Fixed dict chgd size during iteration (Michael Laing)
  • SelectConnection: Fixed KeyError exceptions in IOLoop timeout executions (Shinji Suzuki)
  • BlockingConnection: Add support to make BlockingConnection a Context Manager (@reddec)

0.10.0b2 2015-07-15

  • f72b58f - Fixed failure to purge _ConsumerCancellationEvt from BlockingChannel._pending_events during basic_cancel. (Vitaly Kruglikov)

0.10.0b1 2015-07-10

High-level summary of notable changes:

  • Change to 3-Clause BSD License
  • Python 3.x support
  • Over 150 commits from 19 contributors
  • Refactoring of SelectConnection ioloop
  • This major release contains certain non-backward-compatible API changes as well as significant performance improvements in the BlockingConnection adapter.
  • Non-backward-compatible changes in Channel.add_on_return_callback callback’s signature.
  • The AsynchoreConnection adapter was retired

Details

Python 3.x: this release introduces python 3.x support. Tested on Python 3.3 and 3.4.

AsynchoreConnection: Retired this legacy adapter to reduce maintenance burden; the recommended replacement is the SelectConnection adapter.

SelectConnection: ioloop was refactored for compatibility with other ioloops.

Channel.add_on_return_callback: The callback is now passed the individual parameters channel, method, properties, and body instead of a tuple of those values for congruence with other similar callbacks.

BlockingConnection: This adapter underwent a makeover under the hood and gained significant performance improvements as well as ehnanced timer resolution. It is now implemented as a client of the SelectConnection adapter.

Below is an overview of the BlockingConnection and BlockingChannel API changes:

  • Recursion: the new implementation eliminates callback recursion that sometimes blew out the stack in the legacy implementation (e.g., publish -> consumer_callback -> publish -> consumer_callback, etc.). While BlockingConnection.process_data_events and BlockingConnection.sleep may still be called from the scope of the blocking adapter’s callbacks in order to process pending I/O, additional callbacks will be suppressed whenever BlockingConnection.process_data_events and BlockingConnection.sleep are nested in any combination; in that case, the callback information will be bufferred and dispatched once nesting unwinds and control returns to the level-zero dispatcher.

  • BlockingConnection.connect: this method was removed in favor of the constructor as the only way to establish connections; this reduces maintenance burden, while improving reliability of the adapter.

  • BlockingConnection.process_data_events: added the optional parameter time_limit.

  • BlockingConnection.add_on_close_callback: removed; legacy raised NotImplementedError.

  • BlockingConnection.add_on_open_callback: removed; legacy raised NotImplementedError.

  • BlockingConnection.add_on_open_error_callback: removed; legacy raised NotImplementedError.

  • BlockingConnection.add_backpressure_callback: not supported

  • BlockingConnection.set_backpressure_multiplier: not supported

  • BlockingChannel.add_on_flow_callback: not supported; per docstring in channel.py: “Note that newer versions of RabbitMQ will not issue this but instead use TCP backpressure”.

  • BlockingChannel.flow: not supported

  • BlockingChannel.force_data_events: removed as it is no longer necessary following redesign of the adapter.

  • Removed the nowait parameter from BlockingChannel methods, forcing nowait=False (former API default) in the implementation; this is more suitable for the blocking nature of the adapter and its error-reporting strategy; this concerns the following methods: basic_cancel, confirm_delivery, exchange_bind, exchange_declare, exchange_delete, exchange_unbind, queue_bind, queue_declare, queue_delete, and queue_purge.

  • BlockingChannel.basic_cancel: returns a sequence instead of None; for a no_ack=True consumer, basic_cancel returns a sequence of pending messages that arrived before broker confirmed the cancellation.

  • BlockingChannel.consume: added new optional kwargs arguments and inactivity_timeout. Also, raises ValueError if the consumer creation parameters don’t match those used to create the existing queue consumer generator, if any; this happens when you break out of the consume loop, then call BlockingChannel.consume again with different consumer-creation args without first cancelling the previous queue consumer generator via BlockingChannel.cancel. The legacy implementation would silently resume consuming from the existing queue consumer generator even if the subsequent BlockingChannel.consume was invoked with a different queue name, etc.

  • BlockingChannel.cancel: returns 0; the legacy implementation tried to return the number of requeued messages, but this number was not accurate as it didn’t include the messages returned by the Channel class; this count is not generally useful, so returning 0 is a reasonable replacement.

  • BlockingChannel.open: removed in favor of having a single mechanism for creating a channel (BlockingConnection.channel); this reduces maintenance burden, while improving reliability of the adapter.

  • BlockingChannel.confirm_delivery: raises UnroutableError when unroutable messages that were sent prior to this call are returned before we receive Confirm.Select-ok.

  • BlockingChannel.basic_publish: always returns True when delivery confirmation is not enabled (publisher-acks = off); the legacy implementation returned a bool in this case if `mandatory=True to indicate whether the message was delivered; however, this was non-deterministic, because Basic.Return is asynchronous and there is no way to know how long to wait for it or its absence. The legacy implementation returned None when publishing with publisher-acks = off and mandatory=False. The new implementation always returns True when publishing while publisher-acks = off.

  • BlockingChannel.publish: a new alternate method (vs. basic_publish) for

    publishing a message with more detailed error reporting via UnroutableError and NackError exceptions.

  • BlockingChannel.start_consuming: raises pika.exceptions.RecursionError if called from the scope of a BlockingConnection or BlockingChannel callback.

  • BlockingChannel.get_waiting_message_count: new method; returns the number of messages that may be retrieved from the current queue consumer generator via BasicChannel.consume without blocking.

Commits

  • 5aaa753 - Fixed SSL import and removed no_ack=True in favor of explicit AMQP message handling based on deferreds (skftn)
  • 7f222c2 - Add checkignore for codeclimate (Gavin M. Roy)
  • 4dec370 - Implemented BlockingChannel.flow; Implemented BlockingConnection.add_on_connection_blocked_callback; Implemented BlockingConnection.add_on_connection_unblocked_callback. (Vitaly Kruglikov)
  • 4804200 - Implemented blocking adapter acceptance test for exchange-to-exchange binding. Added rudimentary validation of BasicProperties passthru in blocking adapter publish tests. Updated CHANGELOG. (Vitaly Kruglikov)
  • 4ec07fd - Fixed sending of data in TwistedProtocolConnection (Vitaly Kruglikov)
  • a747fb3 - Remove my copyright from forward_server.py test utility. (Vitaly Kruglikov)
  • 94246d2 - Return True from basic_publish when pubacks is off. Implemented more blocking adapter accceptance tests. (Vitaly Kruglikov)
  • 3ce013d - PIKA-609 Wait for broker to dispatch all messages to client before cancelling consumer in TestBasicCancelWithNonAckableConsumer and TestBasicCancelWithAckableConsumer (Vitaly Kruglikov)
  • 293f778 - Created CHANGELOG entry for release 0.10.0. Fixed up callback documentation for basic_get, basic_consume, and add_on_return_callback. (Vitaly Kruglikov)
  • 16d360a - Removed the legacy AsyncoreConnection adapter in favor of the recommended SelectConnection adapter. (Vitaly Kruglikov)
  • 240a82c - Defer creation of poller’s event loop interrupt socket pair until start is called, because some SelectConnection users (e.g., BlockingConnection adapter) don’t use the event loop, and these sockets would just get reported as resource leaks. (Vitaly Kruglikov)
  • aed5cae - Added EINTR loops in select_connection pollers. Addressed some pylint findings, including an error or two. Wrap socket.send and socket.recv calls in EINTR loops Use the correct exception for socket.error and select.error and get errno depending on python version. (Vitaly Kruglikov)
  • 498f1be - Allow passing exchange, queue and routing_key as text, handle short strings as text in python3 (saarni)
  • 9f7f243 - Restored basic_consume, basic_cancel, and add_on_cancel_callback (Vitaly Kruglikov)
  • 18c9909 - Reintroduced BlockingConnection.process_data_events. (Vitaly Kruglikov)
  • 4b25cb6 - Fixed BlockingConnection/BlockingChannel acceptance and unit tests (Vitaly Kruglikov)
  • bfa932f - Facilitate proper connection state after BasicConnection._adapter_disconnect (Vitaly Kruglikov)
  • 9a09268 - Fixed BlockingConnection test that was failing with ConnectionClosed error. (Vitaly Kruglikov)
  • 5a36934 - Copied synchronous_connection.py from pika-synchronous branch Fixed pylint findings Integrated SynchronousConnection with the new ioloop in SelectConnection Defined dedicated message classes PolledMessage and ConsumerMessage and moved from BlockingChannel to module-global scope. Got rid of nowait args from BlockingChannel public API methods Signal unroutable messages via UnroutableError exception. Signal Nack’ed messages via NackError exception. These expose more information about the failure than legacy basic_publich API. Removed set_timeout and backpressure callback methods Restored legacy is_open, etc. property names (Vitaly Kruglikov)
  • 6226dc0 - Remove deprecated –use-mirrors (Gavin M. Roy)
  • 1a7112f - Raise ConnectionClosed when sending a frame with no connection (#439) (Gavin M. Roy)
  • 9040a14 - Make delivery_tag non-optional (#498) (Gavin M. Roy)
  • 86aabc2 - Bump version (Gavin M. Roy)
  • 562075a - Update a few testing things (Gavin M. Roy)
  • 4954d38 - use unicode_type in blocking_connection.py (Antti Haapala)
  • 133d6bc - Let Travis install ordereddict for Python 2.6, and ttest 3.3, 3.4 too. (Antti Haapala)
  • 0d2287d - Pika Python 3 support (Antti Haapala)
  • 3125c79 - SSLWantRead is not supported before python 2.7.9 and 3.3 (Will)
  • 9a9c46c - Fixed TestDisconnectDuringConnectionStart: it turns out that depending on callback order, it might get either ProbableAuthenticationError or ProbableAccessDeniedError. (Vitaly Kruglikov)
  • cd8c9b0 - A fix the write starvation problem that we see with tornado and pika (Will)
  • 8654fbc - SelectConnection - make interrupt socketpair non-blocking (Will)
  • 4f3666d - Added copyright in forward_server.py and fixed NameError bug (Vitaly Kruglikov)
  • f8ebbbc - ignore docs (Gavin M. Roy)
  • a344f78 - Updated codeclimate config (Gavin M. Roy)
  • 373c970 - Try and fix pathing issues in codeclimate (Gavin M. Roy)
  • 228340d - Ignore codegen (Gavin M. Roy)
  • 4db0740 - Add a codeclimate config (Gavin M. Roy)
  • 7e989f9 - Slight code re-org, usage comment and better naming of test file. (Will)
  • 287be36 - Set up _kqueue member of KQueuePoller before calling super constructor to avoid exception due to missing _kqueue member. Call self._map_event(event) instead of self._map_event(event.filter), because KQueuePoller._map_event() assumes it’s getting an event, not an event filter. (Vitaly Kruglikov)
  • 62810fb - Fix issue #412: reset BlockingConnection._read_poller in BlockingConnection._adapter_disconnect() to guard against accidental access to old file descriptor. (Vitaly Kruglikov)
  • 03400ce - Rationalise adapter acceptance tests (Will)
  • 9414153 - Fix bug selecting non epoll poller (Will)
  • 4f063df - Use user heartbeat setting if server proposes none (Pau Gargallo)
  • 9d04d6e - Deactivate heartbeats when heartbeat_interval is 0 (Pau Gargallo)
  • a52a608 - Bug fix and review comments. (Will)
  • e3ebb6f - Fix incorrect x-expires argument in acceptance tests (Will)
  • 294904e - Get BlockingConnection into consistent state upon loss of TCP/IP connection with broker and implement acceptance tests for those cases. (Vitaly Kruglikov)
  • 7f91a68 - Make SelectConnection behave like an ioloop (Will)
  • dc9db2b - Perhaps 5 seconds is too agressive for travis (Gavin M. Roy)
  • c23e532 - Lower the stuck test timeout (Gavin M. Roy)
  • 1053ebc - Late night bug (Gavin M. Roy)
  • cd6c1bf - More BaseConnection._handle_error cleanup (Gavin M. Roy)
  • a0ff21c - Fix the test to work with Python 2.6 (Gavin M. Roy)
  • 748e8aa - Remove pypy for now (Gavin M. Roy)
  • 1c921c1 - Socket close/shutdown cleanup (Gavin M. Roy)
  • 5289125 - Formatting update from PR (Gavin M. Roy)
  • d235989 - Be more specific when calling getaddrinfo (Gavin M. Roy)
  • b5d1b31 - Reflect the method name change in pika.callback (Gavin M. Roy)
  • df7d3b7 - Cleanup BlockingConnection in a few places (Gavin M. Roy)
  • cd99e1c - Rename method due to use in BlockingConnection (Gavin M. Roy)
  • 7e0d1b3 - Use google style with yapf instead of pep8 (Gavin M. Roy)
  • 7dc9bab - Refactor socket writing to not use sendall #481 (Gavin M. Roy)
  • 4838789 - Dont log the fd #521 (Gavin M. Roy)
  • 765107d - Add Connection.Blocked callback registration methods #476 (Gavin M. Roy)
  • c15b5c1 - Fix _blocking typo pointed out in #513 (Gavin M. Roy)
  • 759ac2c - yapf of codegen (Gavin M. Roy)
  • 9dadd77 - yapf cleanup of codegen and spec (Gavin M. Roy)
  • ddba7ce - Do not reject consumers with no_ack=True #486 #530 (Gavin M. Roy)
  • 4528a1a - yapf reformatting of tests (Gavin M. Roy)
  • e7b6d73 - Remove catching AttributError (#531) (Gavin M. Roy)
  • 41ea5ea - Update README badges [skip ci] (Gavin M. Roy)
  • 6af987b - Add note on contributing (Gavin M. Roy)
  • 161fc0d - yapf formatting cleanup (Gavin M. Roy)
  • edcb619 - Add PYPY to travis testing (Gavin M. Roy)
  • 2225771 - Change the coverage badge (Gavin M. Roy)
  • 8f7d451 - Move to codecov from coveralls (Gavin M. Roy)
  • b80407e - Add confirm_delivery to example (Andrew Smith)
  • 6637212 - Update base_connection.py (bstemshorn)
  • 1583537 - #544 get_waiting_message_count() (markcf)
  • 0c9be99 - Fix #535: pass expected reply_code and reply_text from method frame to Connection._on_disconnect from Connection._on_connection_closed (Vitaly Kruglikov)
  • d11e73f - Propagate ConnectionClosed exception out of BlockingChannel._send_method() and log ConnectionClosed in BlockingConnection._on_connection_closed() (Vitaly Kruglikov)
  • 63d2951 - Fix #541 - make sure connection state is properly reset when BlockingConnection._check_state_on_disconnect raises ConnectionClosed. This supplements the previously-merged PR #450 by getting the connection into consistent state. (Vitaly Kruglikov)
  • 71bc0eb - Remove unused self.fd attribute from BaseConnection (Vitaly Kruglikov)
  • 8c08f93 - PIKA-532 Removed unnecessary params (Vitaly Kruglikov)
  • 6052ecf - PIKA-532 Fix bug in BlockingConnection._handle_timeout that was preventing _on_connection_closed from being called when not closing. (Vitaly Kruglikov)
  • 562aa15 - pika: callback: Display exception message when callback fails. (Stuart Longland)
  • 452995c - Typo fix in connection.py (Andrew)
  • 361c0ad - Added some missing yields (Robert Weidlich)
  • 0ab5a60 - Added complete example for python twisted service (Robert Weidlich)
  • 4429110 - Add deployment and webhooks (Gavin M. Roy)
  • 7e50302 - Fix has_content style in codegen (Andrew Grigorev)
  • 28c2214 - Fix the trove categorization (Gavin M. Roy)
  • de8b545 - Ensure frames can not be interspersed on send (Gavin M. Roy)
  • 8fe6bdd - Fix heartbeat behaviour after connection failure. (Kyösti Herrala)
  • c123472 - Updating BlockingChannel.basic_get doc (it does not receive a callback like the rest of the adapters) (Roberto Decurnex)
  • b5f52fb - Fix number of arguments passed to _on_return callback (Axel Eirola)
  • 765139e - Lower default TIMEOUT to 0.01 (bra-fsn)
  • 6cc22a5 - Fix confirmation on reconnects (bra-fsn)
  • f4faf0a - asynchronous publisher and subscriber examples refactored to follow the StepDown rule (Riccardo Cirimelli)

0.9.14 - 2014-07-11

  • 57fe43e - fix test to generate a correct range of random ints (ml)
  • 0d68dee - fix async watcher for libev_connection (ml)
  • 01710ad - Use default username and password if not specified in URLParameters (Sean Dwyer)
  • fae328e - documentation typo (Jeff Fein-Worton)
  • afbc9e0 - libev_connection: reset_io_watcher (ml)
  • 24332a2 - Fix the manifest (Gavin M. Roy)
  • acdfdef - Remove useless test (Gavin M. Roy)
  • 7918e1a - Skip libev tests if pyev is not installed or if they are being run in pypy (Gavin M. Roy)
  • bb583bf - Remove the deprecated test (Gavin M. Roy)
  • aecf3f2 - Don’t reject a message if the channel is not open (Gavin M. Roy)
  • e37f336 - Remove UTF-8 decoding in spec (Gavin M. Roy)
  • ddc35a9 - Update the unittest to reflect removal of force binary (Gavin M. Roy)
  • fea2476 - PEP8 cleanup (Gavin M. Roy)
  • 9b97956 - Remove force_binary (Gavin M. Roy)
  • a42dd90 - Whitespace required (Gavin M. Roy)
  • 85867ea - Update the content_frame_dispatcher tests to reflect removal of auto-cast utf-8 (Gavin M. Roy)
  • 5a4bd5d - Remove unicode casting (Gavin M. Roy)
  • efea53d - Remove force binary and unicode casting (Gavin M. Roy)
  • e918d15 - Add methods to remove deprecation warnings from asyncore (Gavin M. Roy)
  • 117f62d - Add a coveragerc to ignore the auto generated pika.spec (Gavin M. Roy)
  • 52f4485 - Remove pypy tests from travis for now (Gavin M. Roy)
  • c3aa958 - Update README.rst (Gavin M. Roy)
  • 3e2319f - Delete README.md (Gavin M. Roy)
  • c12b0f1 - Move to RST (Gavin M. Roy)
  • 704f5be - Badging updates (Gavin M. Roy)
  • 7ae33ca - Update for coverage info (Gavin M. Roy)
  • ae7ca86 - add libev_adapter_tests.py; modify .travis.yml to install libev and pyev (ml)
  • f86aba5 - libev_connection: add **kwargs to _handle_event; suppress default_ioloop reuse warning (ml)
  • 603f1cf - async_test_base: add necessary args to _on_cconn_closed (ml)
  • 3422007 - add libev_adapter_tests.py (ml)
  • 6cbab0c - removed relative imports and importing urlparse from urllib.parse for py3+ (a-tal)
  • f808464 - libev_connection: add async watcher; add optional parameters to add_timeout (ml)
  • c041c80 - Remove ev all together for now (Gavin M. Roy)
  • 9408388 - Update the test descriptions and timeout (Gavin M. Roy)
  • 1b552e0 - Increase timeout (Gavin M. Roy)
  • 69a1f46 - Remove the pyev requirement for 2.6 testing (Gavin M. Roy)
  • fe062d2 - Update package name (Gavin M. Roy)
  • 611ad0e - Distribute the LICENSE and README.md (#350) (Gavin M. Roy)
  • df5e1d8 - Ensure that the entire frame is written using socket.sendall (#349) (Gavin M. Roy)
  • 69ec8cf - Move the libev install to before_install (Gavin M. Roy)
  • a75f693 - Update test structure (Gavin M. Roy)
  • 636b424 - Update things to ignore (Gavin M. Roy)
  • b538c68 - Add tox, nose.cfg, update testing config (Gavin M. Roy)
  • a0e7063 - add some tests to increase coverage of pika.connection (Charles Law)
  • c76d9eb - Address issue #459 (Gavin M. Roy)
  • 86ad2db - Raise exception if positional arg for parameters isn’t an instance of Parameters (Gavin M. Roy)
  • 14d08e1 - Fix for python 2.6 (Gavin M. Roy)
  • bd388a3 - Use the first unused channel number addressing #404, #460 (Gavin M. Roy)
  • e7676e6 - removing a debug that was left in last commit (James Mutton)
  • 6c93b38 - Fixing connection-closed behavior to detect on attempt to publish (James Mutton)
  • c3f0356 - Initialize bytes_written in _handle_write() (Jonathan Kirsch)
  • 4510e95 - Fix _handle_write() may not send full frame (Jonathan Kirsch)
  • 12b793f - fixed Tornado Consumer example to successfully reconnect (Yang Yang)
  • f074444 - remove forgotten import of ordereddict (Pedro Abranches)
  • 1ba0aea - fix last merge (Pedro Abranches)
  • 10490a6 - change timeouts structure to list to maintain scheduling order (Pedro Abranches)
  • 7958394 - save timeouts in ordered dict instead of dict (Pedro Abranches)
  • d2746bf - URLParameters and ConnectionParameters accept unicode strings (Allard Hoeve)
  • 596d145 - previous fix for AttributeError made parent and child class methods identical, remove duplication (James Mutton)
  • 42940dd - UrlParameters Docs: fixed amqps scheme examples (Riccardo Cirimelli)
  • 43904ff - Dont test this in PyPy due to sort order issue (Gavin M. Roy)
  • d7d293e - Don’t leave __repr__ sorting up to chance (Gavin M. Roy)
  • 848c594 - Add integration test to travis and fix invocation (Gavin M. Roy)
  • 2678275 - Add pypy to travis tests (Gavin M. Roy)
  • 1877f3d - Also addresses issue #419 (Gavin M. Roy)
  • 470c245 - Address issue #419 (Gavin M. Roy)
  • ca3cb59 - Address issue #432 (Gavin M. Roy)
  • a3ff6f2 - Default frame max should be AMQP FRAME_MAX (Gavin M. Roy)
  • ff3d5cb - Remove max consumer tag test due to change in code. (Gavin M. Roy)
  • 6045dda - Catch KeyError (#437) to ensure that an exception is not raised in a race condition (Gavin M. Roy)
  • 0b4d53a - Address issue #441 (Gavin M. Roy)
  • 180e7c4 - Update license and related files (Gavin M. Roy)
  • 256ed3d - Added Jython support. (Erik Olof Gunnar Andersson)
  • f73c141 - experimental work around for recursion issue. (Erik Olof Gunnar Andersson)
  • a623f69 - Prevent #436 by iterating the keys and not the dict (Gavin M. Roy)
  • 755fcae - Add support for authentication_failure_close, connection.blocked (Gavin M. Roy)
  • c121243 - merge upstream master (Michael Laing)
  • a08dc0d - add arg to channel.basic_consume (Pedro Abranches)
  • 10b136d - Documentation fix (Anton Ryzhov)
  • 9313307 - Fixed minor markup errors. (Jorge Puente Sarrín)
  • fb3e3cf - Fix the spelling of UnsupportedAMQPFieldException (Garrett Cooper)
  • 03d5da3 - connection.py: Propagate the force_channel keyword parameter to methods involved in channel creation (Michael Laing)
  • 7bbcff5 - Documentation fix for basic_publish (JuhaS)
  • 01dcea7 - Expose no_ack and exclusive to BlockingChannel.consume (Jeff Tang)
  • d39b6aa - Fix BlockingChannel.basic_consume does not block on non-empty queues (Juhyeong Park)
  • 6e1d295 - fix for issue 391 and issue 307 (Qi Fan)
  • d9ffce9 - Update parameters.rst (cacovsky)
  • 6afa41e - Add additional badges (Gavin M. Roy)
  • a255925 - Fix return value on dns resolution issue (Laurent Eschenauer)
  • 3f7466c - libev_connection: tweak docs (Michael Laing)
  • 0aaed93 - libev_connection: Fix varable naming (Michael Laing)
  • 0562d08 - libev_connection: Fix globals warning (Michael Laing)
  • 22ada59 - libev_connection: use globals to track sigint and sigterm watchers as they are created globally within libev (Michael Laing)
  • 2649b31 - Move badge [skip ci] (Gavin M. Roy)
  • f70eea1 - Remove pypy and installation attempt of pyev (Gavin M. Roy)
  • f32e522 - Conditionally skip external connection adapters if lib is not installed (Gavin M. Roy)
  • cce97c5 - Only install pyev on python 2.7 (Gavin M. Roy)
  • ff84462 - Add travis ci support (Gavin M. Roy)
  • cf971da - lib_evconnection: improve signal handling; add callback (Michael Laing)
  • 9adb269 - bugfix in returning a list in Py3k (Alex Chandel)
  • c41d5b9 - update exception syntax for Py3k (Alex Chandel)
  • c8506f1 - fix _adapter_connect (Michael Laing)
  • 67cb660 - Add LibevConnection to README (Michael Laing)
  • 1f9e72b - Propagate low-level connection errors to the AMQPConnectionError. (Bjorn Sandberg)
  • e1da447 - Avoid race condition in _on_getok on successive basic_get() when clearing out callbacks (Jeff)
  • 7a09979 - Add support for upcoming Connection.Blocked/Unblocked (Gavin M. Roy)
  • 53cce88 - TwistedChannel correctly handles multi-argument deferreds. (eivanov)
  • 66f8ace - Use uuid when creating unique consumer tag (Perttu Ranta-aho)
  • 4ee2738 - Limit the growth of Channel._cancelled, use deque instead of list. (Perttu Ranta-aho)
  • 0369aed - fix adapter references and tweak docs (Michael Laing)
  • 1738c23 - retry select.select() on EINTR (Cenk Alti)
  • 1e55357 - libev_connection: reset internal state on reconnect (Michael Laing)
  • 708559e - libev adapter (Michael Laing)
  • a6b7c8b - Prioritize EPollPoller and KQueuePoller over PollPoller and SelectPoller (Anton Ryzhov)
  • 53400d3 - Handle socket errors in PollPoller and EPollPoller Correctly check ‘select.poll’ availability (Anton Ryzhov)
  • a6dc969 - Use dict.keys & items instead of iterkeys & iteritems (Alex Chandel)
  • 5c1b0d0 - Use print function syntax, in examples (Alex Chandel)
  • ac9f87a - Fixed a typo in the name of the Asyncore Connection adapter (Guruprasad)
  • dfbba50 - Fixed bug mentioned in Issue #357 (Erik Andersson)
  • c906a2d - Drop additional flags when getting info for the hostnames, log errors (#352) (Gavin M. Roy)
  • baf23dd - retry poll() on EINTR (Cenk Alti)
  • 7cd8762 - Address ticket #352 catching an error when socket.getprotobyname fails (Gavin M. Roy)
  • 6c3ec75 - Prep for 0.9.14 (Gavin M. Roy)
  • dae7a99 - Bump to 0.9.14p0 (Gavin M. Roy)
  • 620edc7 - Use default port and virtual host if omitted in URLParameters (Issue #342) (Gavin M. Roy)
  • 42a8787 - Move the exception handling inside the while loop (Gavin M. Roy)
  • 10e0264 - Fix connection back pressure detection issue #347 (Gavin M. Roy)
  • 0bfd670 - Fixed mistake in commit 3a19d65. (Erik Andersson)
  • da04bc0 - Fixed Unknown state on disconnect error message generated when closing connections. (Erik Andersson)
  • 3a19d65 - Alternative solution to fix #345. (Erik Andersson)
  • abf9fa8 - switch to sendall to send entire frame (Dustin Koupal)
  • 9ce8ce4 - Fixed the async publisher example to work with reconnections (Raphaël De Giusti)
  • 511028a - Fix typo in TwistedChannel docstring (cacovsky)
  • 8b69e5a - calls self._adapter_disconnect() instead of self.disconnect() which doesn’t actually exist #294 (Mark Unsworth)
  • 06a5cf8 - add NullHandler to prevent logging warnings (Cenk Alti)
  • f404a9a - Fix #337 cannot start ioloop after stop (Ralf Nyren)

0.9.13 - 2013-05-15

Major Changes

  • IPv6 Support with thanks to Alessandro Tagliapietra for initial prototype
  • Officially remove support for <= Python 2.5 even though it was broken already
  • Drop pika.simplebuffer.SimpleBuffer in favor of the Python stdlib collections.deque object
  • New default object for receiving content is a “bytes” object which is a str wrapper in Python 2, but paves way for Python 3 support
  • New “Raw” mode for frame decoding content frames (#334) addresses issues #331, #229 added by Garth Williamson
  • Connection and Disconnection logic refactored, allowing for cleaner separation of protocol logic and socket handling logic as well as connection state management
  • New “on_open_error_callback” argument in creating connection objects and new Connection.add_on_open_error_callback method
  • New Connection.connect method to cleanly allow for reconnection code
  • Support for all AMQP field types, using protocol specified signed/unsigned unpacking

Backwards Incompatible Changes

  • Method signature for creating connection objects has new argument “on_open_error_callback” which is positionally before “on_close_callback”
  • Internal callback variable names in connection.Connection have been renamed and constants used. If you relied on any of these callbacks outside of their internal use, make sure to check out the new constants.
  • Connection._connect method, which was an internal only method is now deprecated and will raise a DeprecationWarning. If you relied on this method, your code needs to change.
  • pika.simplebuffer has been removed

Bugfixes

  • BlockingConnection consumer generator does not free buffer when exited (#328)
  • Unicode body payloads in the blocking adapter raises exception (#333)
  • Support “b” short-short-int AMQP data type (#318)
  • Docstring type fix in adapters/select_connection (#316) fix by Rikard Hultén
  • IPv6 not supported (#309)
  • Stop the HeartbeatChecker when connection is closed (#307)
  • Unittest fix for SelectConnection (#336) fix by Erik Andersson
  • Handle condition where no connection or socket exists but SelectConnection needs a timeout for retrying a connection (#322)
  • TwistedAdapter lagging behind BaseConnection changes (#321) fix by Jan Urbański

Other

  • Refactored documentation
  • Added Twisted Adapter example (#314) by nolinksoft

0.9.12 - 2013-03-18

Bugfixes

  • New timeout id hashing was not unique

0.9.11 - 2013-03-17

Bugfixes

  • Address inconsistent channel close callback documentation and add the signature change to the TwistedChannel class (#305)
  • Address a missed timeout related internal data structure name change introduced in the SelectConnection 0.9.10 release. Update all connection adapters to use same signature and docstring (#306).

0.9.10 - 2013-03-16

Bugfixes

  • Fix timeout in twisted adapter (Submitted by cellscape)
  • Fix blocking_connection poll timer resolution to milliseconds (Submitted by cellscape)
  • Fix channel._on_close() without a method frame (Submitted by Richard Boulton)
  • Addressed exception on close (Issue #279 - fix by patcpsc)
  • ‘messages’ not initialized in BlockingConnection.cancel() (Issue #289 - fix by Mik Kocikowski)
  • Make queue_unbind behave like queue_bind (Issue #277)
  • Address closing behavioral issues for connections and channels (Issue #275)
  • Pass a Method frame to Channel._on_close in Connection._on_disconnect (Submitted by Jan Urbański)
  • Fix channel closed callback signature in the Twisted adapter (Submitted by Jan Urbański)
  • Don’t stop the IOLoop on connection close for in the Twisted adapter (Submitted by Jan Urbański)
  • Update the asynchronous examples to fix reconnecting and have it work
  • Warn if the socket was closed such as if RabbitMQ dies without a Close frame
  • Fix URLParameters ssl_options (Issue #296)
  • Add state to BlockingConnection addressing (Issue #301)
  • Encode unicode body content prior to publishing (Issue #282)
  • Fix an issue with unicode keys in BasicProperties headers key (Issue #280)
  • Change how timeout ids are generated (Issue #254)
  • Address post close state issues in Channel (Issue #302)

** Behavior changes **

  • Change core connection communication behavior to prefer outbound writes over reads, addressing a recursion issue
  • Update connection on close callbacks, changing callback method signature
  • Update channel on close callbacks, changing callback method signature
  • Give more info in the ChannelClosed exception
  • Change the constructor signature for BlockingConnection, block open/close callbacks
  • Disable the use of add_on_open_callback/add_on_close_callback methods in BlockingConnection

0.9.9 - 2013-01-29

Bugfixes

  • Only remove the tornado_connection.TornadoConnection file descriptor from the IOLoop if it’s still open (Issue #221)
  • Allow messages with no body (Issue #227)
  • Allow for empty routing keys (Issue #224)
  • Don’t raise an exception when trying to send a frame to a closed connection (Issue #229)
  • Only send a Connection.CloseOk if the connection is still open. (Issue #236 - Fix by noleaf)
  • Fix timeout threshold in blocking connection - (Issue #232 - Fix by Adam Flynn)
  • Fix closing connection while a channel is still open (Issue #230 - Fix by Adam Flynn)
  • Fixed misleading warning and exception messages in BaseConnection (Issue #237 - Fix by Tristan Penman)
  • Pluralised and altered the wording of the AMQPConnectionError exception (Issue #237 - Fix by Tristan Penman)
  • Fixed _adapter_disconnect in TornadoConnection class (Issue #237 - Fix by Tristan Penman)
  • Fixing hang when closing connection without any channel in BlockingConnection (Issue #244 - Fix by Ales Teska)
  • Remove the process_timeouts() call in SelectConnection (Issue #239)
  • Change the string validation to basestring for host connection parameters (Issue #231)
  • Add a poller to the BlockingConnection to address latency issues introduced in Pika 0.9.8 (Issue #242)
  • reply_code and reply_text is not set in ChannelException (Issue #250)
  • Add the missing constraint parameter for Channel._on_return callback processing (Issue #257 - Fix by patcpsc)
  • Channel callbacks not being removed from callback manager when channel is closed or deleted (Issue #261)

0.9.8 - 2012-11-18

Bugfixes

  • Channel.queue_declare/BlockingChannel.queue_declare not setting up callbacks property for empty queue name (Issue #218)
  • Channel.queue_bind/BlockingChannel.queue_bind not allowing empty routing key
  • Connection._on_connection_closed calling wrong method in Channel (Issue #219)
  • Fix tx_commit and tx_rollback bugs in BlockingChannel (Issue #217)

0.9.7 - 2012-11-11

New features

Changes

  • BlockingChannel._send_method will only wait if explicitly told to

Bugfixes

  • Added the exchange “type” parameter back but issue a DeprecationWarning
  • Dont require a queue name in Channel.queue_declare()
  • Fixed KeyError when processing timeouts (Issue # 215 - Fix by Raphael De Giusti)
  • Don’t try and close channels when the connection is closed (Issue #216 - Fix by Charles Law)
  • Dont raise UnexpectedFrame exceptions, log them instead
  • Handle multiple synchronous RPC calls made without waiting for the call result (Issues #192, #204, #211)
  • Typo in docs (Issue #207 Fix by Luca Wehrstedt)
  • Only sleep on connection failure when retry attempts are > 0 (Issue #200)
  • Bypass _rpc method and just send frames for Basic.Ack, Basic.Nack, Basic.Reject (Issue #205)

0.9.6 - 2012-10-29

New features

  • URLParameters
  • BlockingChannel.start_consuming() and BlockingChannel.stop_consuming()
  • Delivery Confirmations
  • Improved unittests

Major bugfix areas

  • Connection handling
  • Blocking functionality in the BlockingConnection
  • SSL
  • UTF-8 Handling

Removals

  • pika.reconnection_strategies
  • pika.channel.ChannelTransport
  • pika.log
  • pika.template
  • examples directory

0.9.5 - 2011-03-29

Changelog

  • Scope changes with adapter IOLoops and CallbackManager allowing for cleaner, multi-threaded operation
  • Add support for Confirm.Select with channel.Channel.confirm_delivery()
  • Add examples of delivery confirmation to examples (demo_send_confirmed.py)
  • Update uses of log.warn with warning.warn for TCP Back-pressure alerting
  • License boilerplate updated to simplify license text in source files
  • Increment the timeout in select_connection.SelectPoller reducing CPU utilization
  • Bug fix in Heartbeat frame delivery addressing issue #35
  • Remove abuse of pika.log.method_call through a majority of the code
  • Rename of key modules: table to data, frames to frame
  • Cleanup of frame module and related classes
  • Restructure of tests and test runner
  • Update functional tests to respect RABBITMQ_HOST, RABBITMQ_PORT environment variables
  • Bug fixes to reconnection_strategies module
  • Fix the scale of timeout for PollPoller to be specified in milliseconds
  • Remove mutable default arguments in RPC calls
  • Add data type validation to RPC calls
  • Move optional credentials erasing out of connection.Connection into credentials module
  • Add support to allow for additional external credential types
  • Add a NullHandler to prevent the ‘No handlers could be found for logger “pika”’ error message when not using pika.log in a client app at all.
  • Clean up all examples to make them easier to read and use
  • Move documentation into its own repository https://github.com/pika/documentation
  • channel.py
    • Move channel.MAX_CHANNELS constant from connection.CHANNEL_MAX
    • Add default value of None to ChannelTransport.rpc
    • Validate callback and acceptable replies parameters in ChannelTransport.RPC
    • Remove unused connection attribute from Channel
  • connection.py
    • Remove unused import of struct
    • Remove direct import of pika.credentials.PlainCredentials - Change to import pika.credentials
    • Move CHANNEL_MAX to channel.MAX_CHANNELS
    • Change ConnectionParameters initialization parameter heartbeat to boolean
    • Validate all inbound parameter types in ConnectionParameters
    • Remove the Connection._erase_credentials stub method in favor of letting the Credentials object deal with that itself.
    • Warn if the credentials object intends on erasing the credentials and a reconnection strategy other than NullReconnectionStrategy is specified.
    • Change the default types for callback and acceptable_replies in Connection._rpc
    • Validate the callback and acceptable_replies data types in Connection._rpc
  • adapters.blocking_connection.BlockingConnection
    • Addition of _adapter_disconnect to blocking_connection.BlockingConnection
    • Add timeout methods to BlockingConnection addressing issue #41
    • BlockingConnection didn’t allow you register more than one consumer callback because basic_consume was overridden to block immediately. New behavior allows you to do so.
    • Removed overriding of base basic_consume and basic_cancel methods. Now uses underlying Channel versions of those methods.
    • Added start_consuming() method to BlockingChannel to start the consumption loop.
    • Updated stop_consuming() to iterate through all the registered consumers in self._consumers and issue a basic_cancel.

Indices and tables