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 five 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. Here that means declaring the exchange to publish to.
5. Once the exchange is declared, publish, then close the connection. Closing it does not stop the IOLoop, so on_close does that and the process exits.

import pika
from pika.exchange_type import ExchangeType


# Step #3
def on_open(connection):
    connection.channel(on_open_callback=on_channel_open)


# Step #4
def on_channel_open(channel):
    # Declare the exchange before publishing to it. Without this the broker
    # answers the publish with a 404 and closes the channel.
    channel.exchange_declare(
        exchange='exchange_name',
        exchange_type=ExchangeType.direct,
        callback=lambda _frame: on_exchange_declared(channel))


# Step #5
def on_exchange_declared(channel):
    channel.basic_publish(
        'exchange_name', 'routing_key', 'Test Message',
        pika.BasicProperties(content_type='text/plain', type='example'))

    # Nothing left to do, so close and let on_close stop the IOLoop.
    connection.close()


def on_close(connection, exception):
    # Invoked when the connection is closed
    connection.ioloop.stop()


# Step #1: Connect to RabbitMQ
connection = pika.SelectConnection(on_open_callback=on_open,
                                   on_close_callback=on_close)

try:
    # Step #2 - Start and block on the IOLoop
    connection.ioloop.start()

# Catch a Keyboard Interrupt to make sure that the connection is closed cleanly
except KeyboardInterrupt:
    # close() raises ConnectionWrongStateError unless the connection is open,
    # and by this point it may have closed itself, so ask first. Note this
    # also skips an interrupt arriving mid-connect, which drops the socket.
    if connection.is_open:
        connection.close()
        # Start the IOLoop again so Pika can communicate,
        # it will stop when on_close is called
        connection.ioloop.start()