API¶

Basics¶

There are two principal objects when using aioamqp:

The protocol object, used to begin a connection to aioamqp,

The channel object, used when creating a new channel to effectively use an AMQP channel.

Starting a connection¶

Starting a connection to AMQP really mean instanciate a new asyncio Protocol subclass.

aioamqp.connect(host, port, login, password, virtualhost, ssl, login_method, insist, protocol_factory, verify_ssl, loop, kwargs) → Transport, AmqpProtocol¶

Convenient method to connect to an AMQP broker

Parameters:

host (str) – the host to connect to
port (int) – broker port
login (str) – login
password (str) – password
virtualhost (str) – AMQP virtualhost to use for this connection
ssl (bool) – create an SSL connection instead of a plain unencrypted one
verify_ssl (bool) – verify server’s SSL certificate (True by default)
login_method (str) – AMQP auth method
insist (bool) – insist on connecting to a server
protocol_factory (AmqpProtocol) – factory to use, if you need to subclass AmqpProtocol
loop (EventLopp) – set the event loop to use
kwargs (dict) – arguments to be given to the protocol_factory instance

import asyncio
import aioamqp

@asyncio.coroutine
def connect():
    try:
        transport, protocol = yield from aioamqp.connect()  # use default parameters
    except aioamqp.AmqpClosedConnection:
        print("closed connections")
        return

    print("connected !")
    yield from asyncio.sleep(1)

    print("close connection")
    yield from protocol.close()
    transport.close()

asyncio.get_event_loop().run_until_complete(connect())

In this example, we just use the method “start_connection” to begin a communication with the server, which deals with credentials and connection tunning.

If you’re not using the default event loop (e.g. because you’re using aioamqp from a different thread), call aioamqp.connect(loop=your_loop).

The AmqpProtocol uses the kwargs arguments to configure the connection to the AMQP Broker:

AmqpProtocol.__init__(self, *args, **kwargs):

The protocol to communicate with AMQP

Parameters:

channel_max (int) – specifies highest channel number that the server permits. Usable channel numbers are in the range 1..channel-max. Zero indicates no specified limit.
frame_max (int) – the largest frame size that the server proposes for the connection, including frame header and end-byte. The client can negotiate a lower value. Zero means that the server does not impose any specific limit but may reject very large frames if it cannot allocate resources for them.
heartbeat (int) – the delay, in seconds, of the connection heartbeat that the server wants. Zero means the server does not want a heartbeat.
loop (Asyncio.EventLoop) – specify the eventloop to use.
client_properties (dict) – configure the client to connect to the AMQP server.

Handling errors¶

The connect() method has an extra ‘on_error’ kwarg option. This on_error is a callback or a coroutine function which is called with an exception as the argument:

import asyncio
import socket
import aioamqp

@asyncio.coroutine
def error_callback(exception):
    print(exception)

@asyncio.coroutine
def connect():
    try:
        transport, protocol = yield from aioamqp.connect(
            host='nonexistant.com',
            on_error=error_callback,
            client_properties={
                'program_name': "test",
                'hostname' : socket.gethostname(),
            },

        )
    except aioamqp.AmqpClosedConnection:
        print("closed connections")
        return

asyncio.get_event_loop().run_until_complete(connect())

Publishing messages¶

A channel is the main object when you want to send message to an exchange, or to consume message from a queue:

channel = yield from protocol.channel()

When you want to produce some content, you declare a queue then publish message into it:

yield from channel.queue_declare("my_queue")
yield from channel.publish("aioamqp hello", '', "my_queue")

Note: we’re pushing message to “my_queue” queue, through the default amqp exchange.

Consuming messages¶

When consuming message, you connect to the same queue you previously created:

import asyncio
import aioamqp

@asyncio.coroutine
def callback(channel, body, envelope, properties):
    print(body)

channel = yield from protocol.channel()
yield from channel.basic_consume(callback, queue_name="my_queue")

The basic_consume method tells the server to send us the messages, and will call callback with amqp response arguments.

The consumer_tag is the id of your consumer, and the delivery_tag is the tag used if you want to acknowledge the message.

In the callback:

the first body parameter is the message
the envelope is an instance of envelope.Envelope class which encapsulate a group of amqp parameter such as:
```
consumer_tag
delivery_tag
exchange_name
routing_key
is_redeliver
```

the properties are message properties, an instance of properties.Properties with the following members:

content_type
content_encoding
headers
delivery_mode
priority
correlation_id
reply_to
expiration
message_id
timestamp
type
user_id
app_id
cluster_id

Server Cancellation¶

RabbitMQ offers an AMQP extension to notify a consumer when a queue is deleted. See Consumer Cancel Notification for additional details. aioamqp enables the extension for all channels but takes no action when the consumer is cancelled. Your application can be notified of consumer cancellations by adding a callback to the channel:

@asyncio.coroutine
def consumer_cancelled(channel, consumer_tag):
    # implement required cleanup here
    pass


@asyncio.coroutine
def consumer(channel, body, envelope, properties):
    channel.basic_ack(envelope.delivery_tag)


channel = yield from protocol.channel()
channel.add_cancellation_callback(consumer_cancelled)
yield from channel.basic_consume(consumer, queue_name="my_queue")

The callback can be a simple callable or an asynchronous co-routine. It can be used to restart consumption on the channel, close the channel, or anything else that is appropriate for your application.

Queues¶

Queues are managed from the Channel object.

Channel.queue_declare(queue_name, passive, durable, exclusive, auto_delete, no_wait, arguments, timeout) → dict¶

Coroutine, creates or checks a queue on the broker

Parameters:

queue_name (str) – the queue to receive message from
passive (bool) – if set, the server will reply with Declare-Ok if the queue already exists with the same name, and raise an error if not. Checks for the same parameter as well.
durable (bool) – if set when creating a new queue, the queue will be marked as durable. Durable queues remain active when a server restarts.
exclusive (bool) – request exclusive consumer access, meaning only this consumer can access the queue
no_wait (bool) – if set, the server will not respond to the method
arguments (dict) – AMQP arguments to be passed when creating the queue.
timeout (int) – wait for the server to respond after timeout

Here is an example to create a randomly named queue with special arguments x-max-priority:

result = yield from channel.queue_declare(
    queue_name='', durable=True, arguments={'x-max-priority': 4}
)

Channel.queue_delete(queue_name, if_unused, if_empty, no_wait, timeout)¶

Coroutine, delete a queue on the broker

Parameters:

queue_name (str) – the queue to receive message from
if_unused (bool) – the queue is deleted if it has no consumers. Raise if not.
if_empty (bool) – the queue is deleted if it has no messages. Raise if not.
no_wait (bool) – if set, the server will not respond to the method
arguments (dict) – AMQP arguments to be passed when creating the queue.
timeout (int) – wait for the server to respond after timeout

Channel.queue_bind(queue_name, exchange_name, routing_key, no_wait, arguments, timeout)¶

Coroutine, bind a queue to an exchange

Parameters:

queue_name (str) – the queue to receive message from.
exchange_name (str) – the exchange to bind the queue to.
routing_key (str) – the routing_key to route message.
no_wait (bool) – if set, the server will not respond to the method
arguments (dict) – AMQP arguments to be passed when creating the queue.
timeout (int) – wait for the server to respond after timeout

This simple example creates a queue, an exchange and bind them together.

channel = yield from protocol.channel()
yield from channel.queue_declare(queue_name='queue')
yield from channel.exchange_declare(exchange_name='exchange')

yield from channel.queue_bind('queue', 'exchange', routing_key='')

Channel.queue_unbind(queue_name, exchange_name, routing_key, arguments, timeout)¶

Coroutine, unbind a queue and an exchange.

PARAM STR ROUTING_KEY:
Parameters:	queue_name (str) – the queue to receive message from. exchange_name (str) – the exchange to bind the queue to. no_wait (bool) – if set, the server will not respond to the method arguments (dict) – AMQP arguments to be passed when creating the queue. timeout (int) – wait for the server to respond after timeout
	THE ROUTING_KEY TO ROUTE MESSAGE.

Channel.queue_purge(queue_name, no_wait, timeout)¶

Coroutine, purge a queue

Parameters:	queue_name (str) – the queue to receive message from.

Exchanges¶

Exchanges are used to correctly route message to queue: a publisher publishes a message into an exchanges, which routes the message to the corresponding queue.

Channel.exchange_declare(exchange_name, type_name, passive, durable, auto_delete, no_wait, arguments, timeout) → dict¶

Coroutine, creates or checks an exchange on the broker

Parameters:

exchange_name (str) – the exchange to receive message from
type_name (str) – the exchange type (fanout, direct, topics …)
passive (bool) – if set, the server will reply with Declare-Ok if the exchange already exists with the same name, and raise an error if not. Checks for the same parameter as well.
durable (bool) – if set when creating a new exchange, the exchange will be marked as durable. Durable exchanges remain active when a server restarts.
auto_delete (bool) – if set, the exchange is deleted when all queues have finished using it.
no_wait (bool) – if set, the server will not respond to the method
arguments (dict) – AMQP arguments to be passed when creating the exchange.
timeout (int) – wait for the server to respond after timeout

Note: the internal flag is deprecated and not used in this library.

channel = yield from protocol.channel()
yield from channel.exchange_declare(exchange_name='exchange', auto_delete=True)

Channel.exchange_delete(exchange_name, if_unused, no_wait, timeout)¶

Coroutine, delete a exchange on the broker

Parameters:	exchange_name (str) – the exchange to receive message from if_unused (bool) – the exchange is deleted if it has no consumers. Raise if not. no_wait (bool) – if set, the server will not respond to the method arguments (dict) – AMQP arguments to be passed when creating the exchange. timeout (int) – wait for the server to respond after timeout

Channel.exchange_bind(exchange_destination, exchange_source, routing_key, no_wait, arguments, timeout)¶

Coroutine, binds two exchanges together

Parameters:

exchange_destination (str) – specifies the name of the destination exchange to bind
exchange_source (str) – specified the name of the source exchange to bind.
exchange_destination – specifies the name of the destination exchange to bind
no_wait (bool) – if set, the server will not respond to the method
arguments (dict) – AMQP arguments to be passed when creating the exchange.
timeout (int) – wait for the server to respond after timeout

Channel.exchange_unbind(exchange_destination, exchange_source, routing_key, no_wait, arguments, timeout)¶

Coroutine, unbind an exchange from an exchange.

Parameters:

exchange_destination (str) – specifies the name of the destination exchange to bind
exchange_source (str) – specified the name of the source exchange to bind.
exchange_destination – specifies the name of the destination exchange to bind
no_wait (bool) – if set, the server will not respond to the method
arguments (dict) – AMQP arguments to be passed when creating the exchange.
timeout (int) – wait for the server to respond after timeout