Chapter 5. Using the API

AMQ Python is an asynchronous event-driven API. To define how an application handles events, the user implements callback methods on the MessagingHandler class. These methods are then called as network activity or timers trigger new events.

class ExampleHandler(MessagingHandler):
    def on_start(self, event):
        print("The container event loop has started")

    def on_sendable(self, event):
        print("A message can be sent")

    def on_message(self, event):
        print("A message is received")

These are only a few common-case events. The full set is documented in the API reference.

The event argument has attributes for accessing the object the event is regarding. Attributes with no relevance to a particular event are null.

event.container
event.connection
event.session
event.sender
event.receiver
event.delivery
event.message

The container is the top-level API object. It is the entry point for creating connections, and it is responsible for running the main event loop. It is often constructed with a global event handler.

handler = ExampleHandler()
container = Container(handler)
container.run()

Setting the container identity

Each container instance has a unique identity called the container ID. When AMQ Python makes a connection, it sends the container ID to the remote peer. To set the container ID, pass it to the Container constructor.

container = Container(handler)
container.container_id = "job-processor-3"

If the user does not set the ID, the library will generate a UUID when the container is constucted.

Connection URLs encode the information used to establish new connections.

scheme://host[:port]

amqps://example.com
amqps://example.net:56720
amqp://127.0.0.1
amqp://[::1]:2000

To connect to a remote server, call the Container.connect() method with a connection URL. This is typically done inside the MessagingHandler.on_start() method.

class ExampleHandler(MessagingHandler):
    def on_start(self, event):
        event.container.connect("amqp://example.com")

    def on_connection_opened(self, event):
        print("Connection", event.connection, "is open")

See the Section 5.5, “Security” section for information about creating secure connections.

AMQ Python can accept inbound network connections, allowing you to build messaging servers. To start listening for connections, use the Container.listen() method with a URL containing the local host address and port to listen on.

class ExampleHandler(MessagingHandler):
    def on_start(self, event):
        event.container.listen("0.0.0.0")

    def on_connection_opened(self, event):
        print("New incoming connection", event.connection)

The special IP address 0.0.0.0 listens on all available IPv4 interfaces. To listen on all IPv6 interfaces, use [::0].

For more information, see the server receive.py example.

Reconnect allows a client to recover from lost connections. It is used to ensure that the components in a distributed system reestablish communication after temporary network or component failures.

AMQ Python enables reconnect by default. If a connection is lost or a connection attempt fails, the client will try again after a brief delay. The delay increases exponentially for each new attempt, up to a default maximum of 10 seconds.

To disable reconnect, set the reconnect connection option to False.

container.connect("amqp://example.com", reconnect=False)

To control the delays between connection attempts, define a class implementing the reset() and next() methods and set the reconnect connection option to an instance of that class.

class ExampleReconnect(object):
    def __init__(self):
        self.delay = 0

    def reset(self):
        self.delay = 0

    def next(self):
        if self.delay == 0:
            self.delay = 0.1
        else:
            self.delay = min(10, 2 * self.delay)

        return self.delay

container.connect("amqp://example.com", reconnect=ExampleReconnect())

The next method returns the next delay in seconds. The reset method is called once before the reconnect process begins.

AMQ Python allows you to configure multiple connection endpoints. If connecting to one fails, the client attempts to connect to the next in the list. If the list is exhausted, the process starts over.

To specify multiple connection endpoints, set the urls connection option to a list of connection URLs.

urls = ["amqp://alpha.example.com", "amqp://beta.example.com"]
container.connect(urls=urls)

It is an error to use the url and urls options at the same time.

Some message servers support on-demand creation of queues and topics. When a sender or receiver is attached, the server uses the sender target address or the receiver source address to create a queue or topic with a name matching the address.

The message server typically defaults to creating either a queue (for one-to-one message delivery) or a topic (for one-to-many message delivery). The client can indicate which it prefers by setting the queue or topic capability on the source or target.

To select queue or topic semantics, follow these steps:

class CapabilityOptions(SenderOption):
    def apply(self, sender):
        sender.target.capabilities.put_object(symbol("queue"))

class ExampleHandler(MessagingHandler):
    def on_start(self, event):
        conn = event.container.connect("amqp://example.com")
        event.container.create_sender(conn, "jobs", options=CapabilityOptions())

class CapabilityOptions(ReceiverOption):
    def apply(self, receiver):
        receiver.source.capabilities.put_object(symbol("topic"))

class ExampleHandler(MessagingHandler):
    def on_start(self, event):
        conn = event.container.connect("amqp://example.com")
        event.container.create_receiver(conn, "notifications", options=CapabilityOptions())

For more information, see the following examples:

A durable subscription is a piece of state on the remote server representing a message receiver. Ordinarily, message receivers are discarded when a client closes. However, because durable subscriptions are persistent, clients can detach from them and then re-attach later. Any messages received while detached are available when the client re-attaches.

Durable subscriptions are uniquely identified by combining the client container ID and receiver name to form a subscription ID. These must have stable values so that the subscription can be recovered.

To create a durable subscription, follow these steps:

To detach from a subscription, use the Receiver.detach() method. To terminate the subscription, use the Receiver.close() method.

For more information, see the durable-subscribe.py example.

A shared subscription is a piece of state on the remote server representing one or more message receivers. Because it is shared, multiple clients can consume from the same stream of messages.

The client configures a shared subscription by setting the shared capability on the receiver source.

Shared subscriptions are uniquely identified by combining the client container ID and receiver name to form a subscription ID. These must have stable values so that multiple client processes can locate the same subscription. If the global capability is set in addition to shared, the receiver name alone is used to identify the subscription.

To create a durable subscription, follow these steps:

To detach from a subscription, use the Receiver.detach() method. To terminate the subscription, use the Receiver.close() method.

For more information, see the shared-subscribe.py example.

To send a message, override the on_sendable event handler and call the Sender.send() method. The sendable event fires when the Sender has enough credit to send at least one message.

class ExampleHandler(MessagingHandler):
    def on_start(self, event):
        conn = event.container.connect("amqp://example.com")
        sender = event.container.create_sender(conn, "jobs")

    def on_sendable(self, event):
        message = Message("job-content")
        event.sender.send(message)

For more information, see the send.py example.

When a message is sent, the sender can keep a reference to the delivery object representing the transfer. After the message is delivered, the receiver accepts or rejects it. The sender is notified of the outcome for each delivery.

To monitor the outcome of a sent message, override the on_accepted and on_rejected event handlers and map the delivery state update to the delivery returned from send().

def on_sendable(self, event):
    message = Message(self.message_body)
    delivery = event.sender.send(message)

def on_accepted(self, event):
    print("Delivery", event.delivery, "is accepted")

def on_rejected(self, event):
    print("Delivery", event.delivery, "is rejected")

To receive a message, create a receiver and override the on_message event handler.

class ExampleHandler(MessagingHandler):
    def on_start(self, event):
        conn = event.container.connect("amqp://example.com")
        receiver = event.container.create_receiver(conn, "jobs")

    def on_message(self, event):
        print("Received message", event.message, "from", event.receiver)

For more information, see the receive.py example.

To explicitly accept or reject a delivery, use the Delivery.update() method with the ACCEPTED or REJECTED state in the on_message event handler.

def on_message(self, event):
    try:
        process_message(event.message)
        event.delivery.update(ACCEPTED)
    except:
        event.delivery.update(REJECTED)

By default, if you do not explicity acknowledge a delivery, then the library accepts it after on_message returns. To disable this behavior, set the auto_accept receiver option to false.

AMQ Python uses SSL/TLS to encrypt communication between clients and servers.

To connect to a remote server with SSL/TLS, use a connection URL with the amqps scheme.

container.connect("amqps://example.com")

AMQ Python can authenticate connections with a user and password.

To specify the credentials used for authentication, set the user and password options on the connect() method.

container.connect("amqps://example.com", user="alice", password="secret")

AMQ Python uses the SASL protocol to perform authentication. SASL can use a number of different authentication mechanisms. When two network peers connect, they exchange their allowed mechanisms, and the strongest mechanism allowed by both is selected.

By default, AMQ Python allows all of the mechanisms supported by the local SASL library configuration. To restrict the allowed mechanisms and thereby control what mechanisms can be negotiated, use the allowed_mechs connection option. It takes a string containing a space-separated list of mechanism names.

container.connect("amqps://example.com", allowed_mechs="ANONYMOUS")

This example forces the connection to authenticate using the ANONYMOUS mechanism even if the server we connect to offers other options. Valid mechanisms include ANONYMOUS, PLAIN, SCRAM-SHA-256, SCRAM-SHA-1, GSSAPI, and EXTERNAL.

AMQ Python enables SASL by default. To disable it, set the sasl_enabled connection option to false.

event.container.connect("amqps://example.com", sasl_enabled=False)

Kerberos is a network protocol for centrally managed authentication based on the exchange of encrypted tickets. See Using Kerberos for more information.

The client can log AMQP protocol frames to the console. This data is often critical when diagnosing problems.

To enable protocol logging, set the PN_TRACE_FRM environment variable to 1:

$ export PN_TRACE_FRM=1
$ <your-client-program>

To disable protocol logging, unset the PN_TRACE_FRM environment variable.