Guide for Transport Driver Implementors

Introduction

This document is a best practices guide for the developer interested in creating a new transport driver for Oslo.Messaging. It should also be used by maintainers as a reference for proper driver behavior. This document will describe the driver interface and prescribe the expected behavior of any driver implemented to this interface.

Note well: The API described in this document is internal to the oslo.messaging library and therefore private. Under no circumstances should this API be referenced by code external to the oslo.messaging library.

Driver Interface

The driver interface is defined by a set of abstract base classes. The developer creates a driver by defining concrete classes from these bases. The derived classes embody the logic that is specific for the messaging back-end that is to be supported.

These base classes are defined in the base.py file in the _drivers subdirectory.

IncomingMessage

class oslo_messaging._drivers.base.IncomingMessage(ctxt, message, msg_id=None)

The IncomingMessage class represents a single message received from the messaging backend. Instances of this class are passed to up a server’s messaging processing logic. The backend driver must provide a concrete derivation of this class which provides the backend specific logic for its public methods.

Parameters:

  • ctxt (dict) – Context metadata provided by sending application.

  • message (dict) – The message as provided by the sending application.

acknowledge()

Called by the server to acknowledge receipt of the message. When this is called the driver must notify the backend of the acknowledgment. This call should block at least until the driver has processed the acknowledgment request locally. It may unblock before the acknowledgment state has been acted upon by the backend.

If the acknowledge operation fails this method must issue a log message describing the reason for the failure.

Raises:

Does not raise an exception

abstract requeue()

Called by the server to return the message to the backend so it may be made available for consumption by another server. This call should block at least until the driver has processed the requeue request locally. It may unblock before the backend makes the requeued message available for consumption.

If the requeue operation fails this method must issue a log message describing the reason for the failure.

Support for this method is _optional_. The BaseDriver.require_features() method should indicate whether or not support for requeue is available.

Raises:

Does not raise an exception

RpcIncomingMessage

class oslo_messaging._drivers.base.RpcIncomingMessage(ctxt, message, msg_id=None)

The RpcIncomingMessage represents an RPC request message received from the backend. This class must be used for RPC calls that return a value to the caller.

abstract heartbeat()

Called by the server to send an RPC heartbeat message back to the calling client.

If the client (is new enough to have) passed its timeout value during the RPC call, this method will be called periodically by the server to update the client’s timeout timer while a long-running call is executing.

Raises:

Does not raise an exception

abstract reply(reply=None, failure=None)

Called by the server to send an RPC reply message or an exception back to the calling client.

If an exception is passed via failure the driver must convert it to a form that can be sent as a message and properly converted back to the exception at the remote.

The driver must provide a way to determine the destination address for the reply. For example the driver may use the reply-to field from the corresponding incoming message. Often a driver will also need to set a correlation identifier in the reply to help the remote route the reply to the correct RPCClient.

The driver should provide an at-most-once delivery guarantee for reply messages. This call should block at least until the reply message has been handed off to the backend - there is no need to confirm that the reply has been delivered.

If the reply operation fails this method must issue a log message describing the reason for the failure.

See BaseDriver.send() for details regarding how the received reply is processed.

Parameters:

  • reply (dict) – reply message body

  • failure (Exception) – an exception thrown by the RPC call

Raises:

Does not raise an exception

Listener

class oslo_messaging._drivers.base.Listener(batch_size, batch_timeout, prefetch_size=-1)

A Listener is used to transfer incoming messages from the driver to a server for processing. A callback is used by the driver to transfer the messages.

Parameters:

  • batch_size (int) – desired number of messages passed to single on_incoming_callback notification

  • batch_timeout (float) – defines how long should we wait in seconds for batch_size messages if we already have some messages waiting for processing

  • prefetch_size (int) – defines how many messages we want to prefetch from the messaging backend in a single request. May not be honored by all backend implementations.

abstract cleanup()

Cleanup all resources held by the listener. This method should block until the cleanup is completed.

start(on_incoming_callback)

Start receiving messages. This should cause the driver to start receiving messages from the backend. When message(s) arrive the driver must invoke ‘on_incoming_callback’ passing it the received messages as a list of IncomingMessages.

Parameters:

on_incoming_callback (func) – callback function to be executed when listener receives messages.

stop()

Stop receiving messages. The driver must no longer invoke the callback.

PollStyleListener

class oslo_messaging._drivers.base.PollStyleListener(prefetch_size=-1)

A PollStyleListener is used to transfer received messages to a server for processing. A polling pattern is used to retrieve messages. A PollStyleListener uses a separate thread to run the polling loop. A PollStyleListenerAdapter can be used to create a Listener from a PollStyleListener.

Parameters:

prefetch_size (int) – The number of messages that should be pulled from the backend per receive transaction. May not be honored by all backend implementations.

cleanup()

Cleanup all resources held by the listener. This method should block until the cleanup is completed.

abstract poll(timeout=None, batch_size=1, batch_timeout=None)

poll is called by the server to retrieve incoming messages. It blocks until ‘batch_size’ incoming messages are available, a timeout occurs, or the poll is interrupted by a call to the stop() method.

If ‘batch_size’ is > 1 poll must block until ‘batch_size’ messages are available or at least one message is available and batch_timeout expires

Parameters:

  • timeout (float) – Block up to ‘timeout’ seconds waiting for a message

  • batch_size (int) – Block until this number of messages are received.

  • batch_timeout (float) – Time to wait in seconds for a full batch to arrive. A timer is started when the first message in a batch is received. If a full batch’s worth of messages is not received when the timer expires then poll() returns all messages received thus far.

Raises:

Does not raise an exception.

Returns:

A list of up to batch_size IncomingMessage objects.

stop()

Stop the listener from polling for messages. This method must cause the poll() call to unblock and return whatever messages are currently available. This method is called from a different thread than the poller so it must be thread-safe.

BaseDriver

class oslo_messaging._drivers.base.BaseDriver(conf, url, default_exchange=None, allowed_remote_exmods=None)

Defines the backend driver interface. Each backend driver implementation must provide a concrete derivation of this class implementing the backend specific logic for its public methods.

Parameters:

  • conf (ConfigOpts) – The configuration settings provided by the user.

  • url (TransportURL) – The network address of the messaging backend(s).

  • default_exchange (str) – The exchange to use if no exchange is specified in a Target.

  • allowed_remote_exmods (list) – whitelist of those exception modules which are permitted to be re-raised if an exception is returned in response to an RPC call.

abstract cleanup()

Release all resources used by the driver. This method must block until the cleanup is complete.

abstract listen(target, batch_size, batch_timeout)

Construct a listener for the given target. The listener may be either a Listener or PollStyleListener depending on the driver’s preference. This method is used by the RPC server.

The driver must create subscriptions to the address provided in target. These subscriptions must then be associated with a Listener or PollStyleListener which is returned by this method. See BaseDriver.send() for more detail regarding message addressing.

The driver must support receiving messages sent to the following addresses derived from the values in target:

  • all messages sent to the exchange and topic given in the target. This includes messages sent using a fanout pattern.

  • if the server attribute of the target is set then the driver must also subscribe to messages sent to the exchange, topic, and server

For example, given a target with exchange ‘my-exchange’, topic ‘my-topic’, and server ‘my-server’, the driver would create subscriptions for:

  • all messages sent to my-exchange and my-topic (including fanout)

  • all messages sent to my-exchange, my-topic, and my-server

The driver must pass messages arriving from these subscriptions to the listener. For PollStyleListener the driver should trigger the PollStyleListener.poll() method to unblock and return the incoming messages. For Listener the driver should invoke the callback with the incoming messages.

This method only blocks long enough to establish the subscription(s) and construct the listener. In the case of failover, the driver must restore the subscription(s). Subscriptions should remain active until the listener is stopped.

Parameters:

  • target (Target) – The address(es) to subscribe to.

  • batch_size (int) – passed to the listener

  • batch_timeout (float) – passed to the listener

Returns:

None

Raises:

MessagingException

abstract listen_for_notifications(targets_and_priorities, pool, batch_size, batch_timeout)

Construct a notification listener for the given list of tuples of (target, priority) addresses.

The driver must create a subscription for each (target, priority) pair. The topic for the subscription is created for each pair using the format “%s.%s” % (target.topic, priority). This format is used by the caller of the BaseDriver.send_notification() when setting the topic member of the target parameter.

Only the exchange and topic must be considered when creating subscriptions. server and fanout must be ignored.

The pool parameter, if specified, should cause the driver to create a subscription that is shared with other subscribers using the same pool identifier. Each pool gets a single copy of the message. For example if there is a subscriber pool with identifier foo and another pool bar, then one foo subscriber and one bar subscriber will each receive a copy of the message. The driver should implement a delivery pattern that distributes message in a balanced fashion across the subscribers in a pool.

The driver must raise a NotImplementedError if pooling is not supported and a pool identifier is passed in.

Refer to the description of BaseDriver.send_notification() for further details regarding implementation.

Parameters:

  • targets_and_priorities (list) – List of (target, priority) pairs

  • pool (str) – pool identifier

  • batch_size (int) – passed to the listener

  • batch_timeout (float) – passed to the listener

Returns:

None

Raises:

MessagingException, NotImplementedError

require_features(requeue=False)

The driver must raise a ‘NotImplementedError’ if any of the feature flags passed as True are not supported.

abstract send(target, ctxt, message, wait_for_reply=None, timeout=None, call_monitor_timeout=None, retry=None, transport_options=None)

Send a message to the given target and optionally wait for a reply. This method is used by the RPC client when sending RPC requests to a server.

The driver must use the topic, exchange, and server (if present) attributes of the target to construct the backend-native message address. The message address must match the format used by subscription(s) created by the BaseDriver.listen() method.

If the target’s fanout attribute is set, a copy of the message must be sent to all subscriptions using the exchange and topic values. If fanout is not set, then only one subscriber should receive the message. In the case of multiple subscribers to the same address, only one copy of the message is delivered. In this case the driver should implement a delivery pattern that distributes messages in a balanced fashion across the multiple subscribers.

This method must block the caller until one of the following events occur:

  • the send operation completes successfully

  • timeout seconds elapse (if specified)

  • retry count is reached (if specified)

The wait_for_reply parameter determines whether or not the caller expects a response to the RPC request. If True, this method must block until a response message is received. This method then returns the response message to the caller. The driver must implement a mechanism for routing incoming responses back to their corresponding send request. How this is done may vary based on the type of messaging backend, but typically it involves having the driver create an internal subscription for reply messages and setting the request message’s reply-to header to the subscription address. The driver may also need to supply a correlation identifier for mapping the response back to the sender. See RpcIncomingMessage.reply()

If wait_for_reply is False this method will block until the message has been handed off to the backend - there is no need to confirm that the message has been delivered. Once the handoff completes this method returns.

The driver may attempt to retry sending the message should a recoverable error occur that prevents the message from being passed to the backend. The retry parameter specifies how many attempts to re-send the message the driver may make before raising a MessageDeliveryFailure exception. A value of None or -1 means unlimited retries. 0 means no retry is attempted. N means attempt at most N retries before failing. Note well: the driver MUST guarantee that the message is not duplicated by the retry process.

Parameters:

  • target (Target) – The message’s destination address

  • ctxt (dict) – Context metadata provided by sending application which is transferred along with the message.

  • message (dict) – message provided by the caller

  • wait_for_reply (bool) – If True block until a reply message is received.

  • timeout (float) – Maximum time in seconds to block waiting for the send operation to complete. Should this expire the send() must raise a MessagingTimeout exception

  • call_monitor_timeout (float) – Maximum time the client will wait for the call to complete or receive a message heartbeat indicating the remote side is still executing.

  • retry (int) – maximum message send attempts permitted

  • transport_options (dictionary) – additional parameters to configure the driver for example to send parameters as “mandatory” flag in RabbitMQ

Returns:

A reply message or None if no reply expected

Raises:

MessagingException, any exception thrown by the remote server when executing the RPC call.

abstract send_notification(target, ctxt, message, version, retry)

Send a notification message to the given target. This method is used by the Notifier to send notification messages to a Listener.

Notifications use a store and forward delivery pattern. The driver must allow for delivery in the case where the intended recipient is not present at the time the notification is published. Typically this requires a messaging backend that has the ability to store messages until a consumer is present.

Therefore this method must block at least until the backend accepts ownership of the message. This method does not guarantee that the message has or will be processed by the intended recipient.

The driver must use the topic and exchange attributes of the target to construct the backend-native message address. The message address must match the format used by subscription(s) created by the BaseDriver.listen_for_notifications() method. Only one copy of the message is delivered in the case of multiple subscribers to the same address. In this case the driver should implement a delivery pattern that distributes messages in a balanced fashion across the multiple subscribers.

There is an exception to the single delivery semantics described above: the pool parameter to the BaseDriver.listen_for_notifications() method may be used to set up shared subscriptions. See BaseDriver.listen_for_notifications() for details.

This method must also honor the retry parameter. See BaseDriver.send() for details regarding implementing the retry process.

version indicates whether or not the message should be encapsulated in an envelope. A value < 2.0 should not envelope the message. See common.serialize_msg() for more detail.

Parameters:

  • target (Target) – The message’s destination address

  • ctxt (dict) – Context metadata provided by sending application which is transferred along with the message.

  • message (dict) – message provided by the caller

  • version (float) – determines the envelope for the message

  • retry (int) – maximum message send attempts permitted

Returns:

None

Raises:

MessagingException