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.
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.
oslo_messaging._drivers.base.
IncomingMessage
(ctxt, message)¶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: |
|
---|
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 |
---|
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 |
---|
oslo_messaging._drivers.base.
RpcIncomingMessage
(ctxt, message)¶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.
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 |
---|
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: |
|
---|---|
Raises: | Does not raise an exception |
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: |
|
---|
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.
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.
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: |
|
---|---|
Raises: | Does not raise an exception. |
Returns: | A list of up to batch_size IncomingMessage objects. |
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: |
|
---|
cleanup
()¶Release all resources used by the driver. This method must block until the cleanup is complete.
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:
For example, given a target with exchange ‘my-exchange’, topic ‘my-topic’, and server ‘my-server’, the driver would create subscriptions for:
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: |
|
---|---|
Returns: | None |
Raises: |
|
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: |
|
---|---|
Returns: | None |
Raises: |
|
require_features
(requeue=False)¶The driver must raise a ‘NotImplementedError’ if any of the feature flags passed as True are not supported.
send
(target, ctxt, message, wait_for_reply=None, timeout=None, call_monitor_timeout=None, retry=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 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: |
|
---|---|
Returns: | A reply message or None if no reply expected |
Raises: |
|
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: |
|
---|---|
Returns: | None |
Raises: |
|
Except where otherwise noted, this document is licensed under Creative Commons Attribution 3.0 License. See all OpenStack Legal Documents.