# 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

    async def connect():
        try:
            transport, protocol = await aioamqp.connect()  # use default parameters
        except aioamqp.AmqpClosedConnection:
            print("closed connections")
            return

        print("connected !")
        await asyncio.sleep(1)

        print("close connection")
        await 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

    async def error_callback(exception):
        print(exception)

    async def connect():
        try:
            transport, protocol = await 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 = await protocol.channel()

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

    await channel.queue_declare("my_queue")
    await 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

    async def callback(channel, body, envelope, properties):
        print(body)

    channel = await protocol.channel()
    await 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
      message_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:

    async def consumer_cancelled(channel, consumer_tag):
        # implement required cleanup here
        pass

    async def consumer(channel, body, envelope, properties):
        await channel.basic_client_ack(envelope.delivery_tag)

    channel = await protocol.channel()
    channel.add_cancellation_callback(consumer_cancelled)
    await 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 = await 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 = await protocol.channel()
>     await channel.queue_declare(queue_name='queue')
>     await channel.exchange_declare(exchange_name='exchange')

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

  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

  PARAM STR ROUTING_KEY

    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 = await protocol.channel()
>     await 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