oslomessaging(1)

OSLOMESSAGING(1)

oslo.messaging

OSLOMESSAGING(1)

NAME

oslomessaging - oslo.messaging 16.1.0

The Oslo messaging API supports RPC and notifications over a number of different messaging transports.

CONTRIBUTING TO OSLO.MESSAGING

Contributing

If you would like to contribute to the development of oslo's libraries, first you must take a look to this page:

https://specs.openstack.org/openstack/oslo-specs/specs/policy/contributing.html

If you would like to contribute to the development of OpenStack, you must follow the steps in this page:

https://docs.openstack.org/infra/manual/developers.html

Once those steps have been completed, changes to OpenStack should be submitted for review via the Gerrit tool, following the workflow documented at:

https://docs.openstack.org/infra/manual/developers.html#development-workflow

Pull requests submitted through GitHub will be ignored.

Bugs should be filed on Launchpad, not GitHub:

https://bugs.launchpad.net/oslo.messaging

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

Supported Messaging Drivers

RabbitMQ may not be sufficient for the entire community as the community grows. Pluggability is still something we should maintain, but we should have a very high standard for drivers that are shipped and documented as being supported.

This document defines a very clear policy as to the requirements for drivers to be carried in oslo.messaging and thus supported by the OpenStack community as a whole. We will deprecate any drivers that do not meet the requirements, and announce said deprecations in any appropriate channels to give users time to signal their needs. Deprecation will last for two release cycles before removing the code. We will also review and update documentation to annotate which drivers are supported and which are deprecated given these policies

Policy

Testing

Must have unit and/or functional test coverage of at least 60% as reported by coverage report. Unit tests must be run for all versions of python oslo.messaging currently gates on.
Must have integration testing including at least 3 popular oslo.messaging dependents, preferably at the minimum a devstack-gate job with Nova, Cinder, and Neutron.
All testing above must be voting in the gate of oslo.messaging.

Documentation

•: Must have a reasonable amount of documentation including documentation in the official OpenStack deployment guide.

Support

•: Must have at least two individuals from the community committed to triaging and fixing bugs, and responding to test failures in a timely manner.

Prospective Drivers

•: Drivers that intend to meet the requirements above, but that do not yet meet them will be given one full release cycle, or 6 months, whichever is longer, to comply before being marked for deprecation. Their use, however, will not be supported by the community. This will prevent a chicken and egg problem for new drivers.

NOTE:

This work is licensed under a Creative Commons Attribution 3.0 Unported License. http://creativecommons.org/licenses/by/3.0/legalcode

Oslo Messaging Simulator

This guide explains how to set up and run the oslo messaging simulator for testing different messaging scenarios.

Prerequisites

Python 3.x
virtualenv
wget (for Kafka scenarios)

Environment Setup

This assumes you have git cloned the oslo.messaging repository and are in the root directory of the repository.

Create and activate a virtual environment:

python -m venv .venv
source .venv/bin/activate

Install required packages:

pip install pifpaf
pip install -e .

Running the Simulator

The simulator supports different scenarios for testing messaging patterns. Below are the common usage patterns.

Basic Setup

Before running the simulator, set up the messaging environment:

./tools/setup-scenario-env.sh

Available Scenarios

The simulator supports two main scenarios:

Scenario 01 (RabbitMQ only)

This scenario uses RabbitMQ for both RPC and notifications:

export SCENARIO=scenario01
./tools/setup-scenario-env.sh

Scenario 02 (RabbitMQ + Kafka)

This scenario uses RabbitMQ for RPC and Kafka for notifications:

export SCENARIO=scenario02
./tools/setup-scenario-env.sh

Running the Simulator

RPC Server Example

To start the RPC server:

python tools/simulator.py --url rabbit://pifpaf:secret@127.0.0.1:5682/ rpc-server

RPC Client Example

To start the RPC client:

python tools/simulator.py --url rabbit://pifpaf:secret@127.0.0.1:5682/ rpc-client --exit-wait 15000 -p 64 -m 64

Optional Configuration

You can generate a sample configuration file using oslo-config-generator:

oslo-config-generator --namespace oslo.messaging > oslo.messaging.conf

For reference on all available configuration options, visit: https://docs.openstack.org/oslo.messaging/latest/configuration/opts.html

To use a configuration file with the simulator, use the --config-file option:

python tools/simulator.py --config-file oslo.messaging.conf [other options]

Command Line Options

The simulator supports various command line options:

--url URL: The transport URL for the messaging service
--config-file PATH: Path to a configuration file
-d, --debug: Enable debug mode
-p PROCESSES: Number of processes (for client)
-m MESSAGES: Number of messages (for client)
--exit-wait MILLISECONDS: Wait time before exit (for client)

Cleanup

To clean up the environment, you can terminate the running processes:

pkill -f "RABBITMQ"

Notes

The default scenario is scenario01 if not specified
Kafka setup is automatic when using scenario02
The simulator uses pifpaf to manage the message broker processes
Installing with pip install -e . allows for development mode installation
Configuration options can be referenced in the official documentation

CONFIGURATION

Configuration Options

oslo.messaging uses oslo.config to define and manage configuration options to allow the deployer to control how an application uses the underlying messaging system.

DEFAULT

executor_thread_pool_size

Type: integer
Default: 64

Size of executor thread pool when executor is threading or eventlet.

Deprecated Variations

Group	Name
DEFAULT	rpc_thread_pool_size

rpc_response_timeout

Type: integer
Default: 60

Seconds to wait for a response from a call.

transport_url

Type: string
Default: rabbit://

The network address and optional user credentials for connecting to the messaging backend, in URL format. The expected format is:

driver://[user:pass@]host:port[,[userN:passN@]hostN:portN]/virtual_host?query

Example: rabbit://rabbitmq:password@127.0.0.1:5672//

For full details on the fields in the URL see the documentation of oslo_messaging.TransportURL at https://docs.openstack.org/oslo.messaging/latest/reference/transport.html

control_exchange

Type: string
Default: openstack

The default exchange under which topics are scoped. May be overridden by an exchange name specified in the transport_url option.

rpc_ping_enabled

Type: boolean
Default: False

Add an endpoint to answer to ping calls. Endpoint is named oslo_rpc_server_ping

oslo_messaging_kafka

kafka_max_fetch_bytes

Type: integer
Default: 1048576

Max fetch bytes of Kafka consumer

kafka_consumer_timeout

Type: floating point
Default: 1.0

Default timeout(s) for Kafka consumers

consumer_group

Type: string
Default: oslo_messaging_consumer

Group id for Kafka consumer. Consumers in one group will coordinate message consumption

producer_batch_timeout

Type: floating point
Default: 0.0

Upper bound on the delay for KafkaProducer batching in seconds

producer_batch_size

Type: integer
Default: 16384

Size of batch for the producer async send

compression_codec

Type: string
Default: none
Valid Values: none, gzip, snappy, lz4, zstd

The compression codec for all data generated by the producer. If not set, compression will not be used. Note that the allowed values of this depend on the kafka version

enable_auto_commit

Type: boolean
Default: False

Enable asynchronous consumer commits

max_poll_records

Type: integer
Default: 500

The maximum number of records returned in a poll call

security_protocol

Type: string
Default: PLAINTEXT
Valid Values: PLAINTEXT, SASL_PLAINTEXT, SSL, SASL_SSL

Protocol used to communicate with brokers

sasl_mechanism

Type: string
Default: PLAIN

Mechanism when security protocol is SASL

ssl_cafile

Type: string
Default: ''

CA certificate PEM file used to verify the server certificate

ssl_client_cert_file

Type: string
Default: ''

Client certificate PEM file used for authentication.

ssl_client_key_file

Type: string
Default: ''

Client key PEM file used for authentication.

ssl_client_key_password

Type: string
Default: ''

Client key password file used for authentication.

oslo_messaging_notifications

driver

Type: multi-valued
Default: ''

The Drivers(s) to handle sending notifications. Possible values are messaging, messagingv2, routing, log, test, noop

transport_url

Type: string
Default: <None>

A URL representing the messaging driver to use for notifications. If not set, we fall back to the same configuration used for RPC.

topics

Type: list
Default: ['notifications']

AMQP topic used for OpenStack notifications.

retry

Type: integer
Default: -1

The maximum number of attempts to re-send a notification message which failed to be delivered due to a recoverable error. 0 - No retry, -1 - indefinite

oslo_messaging_rabbit

amqp_durable_queues

Type: boolean
Default: False

Use durable queues in AMQP. If rabbit_quorum_queue is enabled, queues will be durable and this value will be ignored.

amqp_auto_delete

Type: boolean
Default: False

Auto-delete queues in AMQP.

rpc_conn_pool_size

Type: integer
Default: 30
Minimum Value: 1

Size of RPC connection pool.

conn_pool_min_size

Type: integer
Default: 2

The pool size limit for connections expiration policy

conn_pool_ttl

Type: integer
Default: 1200

The time-to-live in sec of idle connections in the pool

ssl

Type: boolean
Default: False

Connect over SSL.

ssl_version

Type: string
Default: ''

SSL version to use (valid only if SSL enabled). Valid values are TLSv1 and SSLv23. SSLv2, SSLv3, TLSv1_1, and TLSv1_2 may be available on some distributions.

ssl_key_file

Type: string
Default: ''

SSL key file (valid only if SSL enabled).

ssl_cert_file

Type: string
Default: ''

SSL cert file (valid only if SSL enabled).

ssl_ca_file

Type: string
Default: ''

SSL certification authority file (valid only if SSL enabled).

ssl_enforce_fips_mode

Type: boolean
Default: False

Global toggle for enforcing the OpenSSL FIPS mode. This feature requires Python support. This is available in Python 3.9 in all environments and may have been backported to older Python versions on select environments. If the Python executable used does not support OpenSSL FIPS mode, an exception will be raised.

heartbeat_in_pthread

Type: boolean
Default: False

(DEPRECATED) It is recommend not to use this option anymore. Run the health check heartbeat thread through a native python thread by default. If this option is equal to False then the health check heartbeat will inherit the execution model from the parent process. For example if the parent process has monkey patched the stdlib by using eventlet/greenlet then the heartbeat will be run through a green thread. This option should be set to True only for the wsgi services.

WARNING:

This option is deprecated for removal. Its value may be silently ignored in the future.

Reason: The option is related to Eventlet which will be removed. In addition this has never worked as expected with services using eventlet for core service framework.

kombu_reconnect_delay

Type: floating point
Default: 1.0
Minimum Value: 0.0
Maximum Value: 4.5

How long to wait (in seconds) before reconnecting in response to an AMQP consumer cancel notification.

kombu_reconnect_splay

Type: floating point
Default: 0.0
Minimum Value: 0.0

Random time to wait for when reconnecting in response to an AMQP consumer cancel notification.

kombu_compression

Type: string
Default: <None>

EXPERIMENTAL: Possible values are: gzip, bz2. If not set compression will not be used. This option may not be available in future versions.

kombu_missing_consumer_retry_timeout

Type: integer
Default: 60

How long to wait a missing client before abandoning to send it its replies. This value should not be longer than rpc_response_timeout.

Deprecated Variations

Group	Name
oslo_messaging_rabbit	kombu_reconnect_timeout

kombu_failover_strategy

Type: string
Default: round-robin
Valid Values: round-robin, shuffle

Determines how the next RabbitMQ node is chosen in case the one we are currently connected to becomes unavailable. Takes effect only if more than one RabbitMQ node is provided in config.

rabbit_login_method

Type: string
Default: AMQPLAIN
Valid Values: PLAIN, AMQPLAIN, EXTERNAL, RABBIT-CR-DEMO

The RabbitMQ login method.

rabbit_retry_interval

Type: integer
Default: 1
Minimum Value: 1

How frequently to retry connecting with RabbitMQ.

rabbit_retry_backoff

Type: integer
Default: 2
Minimum Value: 0

How long to backoff for between retries when connecting to RabbitMQ.

rabbit_interval_max

Type: integer
Default: 30
Minimum Value: 1

Maximum interval of RabbitMQ connection retries.

rabbit_ha_queues

Type: boolean
Default: False

Try to use HA queues in RabbitMQ (x-ha-policy: all). If you change this option, you must wipe the RabbitMQ database. In RabbitMQ 3.0, queue mirroring is no longer controlled by the x-ha-policy argument when declaring a queue. If you just want to make sure that all queues (except those with auto-generated names) are mirrored across all nodes, run: "rabbitmqctl set_policy HA '^(?!amq.).*' '{"ha-mode": "all"}' "

rabbit_quorum_queue

Type: boolean
Default: False

Use quorum queues in RabbitMQ (x-queue-type: quorum). The quorum queue is a modern queue type for RabbitMQ implementing a durable, replicated FIFO queue based on the Raft consensus algorithm. It is available as of RabbitMQ 3.8.0. If set this option will conflict with the HA queues (rabbit_ha_queues) aka mirrored queues, in other words the HA queues should be disabled. Quorum queues are also durable by default so the amqp_durable_queues option is ignored when this option is enabled.

rabbit_transient_quorum_queue

Type: boolean
Default: False

Use quorum queues for transients queues in RabbitMQ. Enabling this option will then make sure those queues are also using quorum kind of rabbit queues, which are HA by default.

rabbit_quorum_delivery_limit

Type: integer
Default: 0

Each time a message is redelivered to a consumer, a counter is incremented. Once the redelivery count exceeds the delivery limit the message gets dropped or dead-lettered (if a DLX exchange has been configured) Used only when rabbit_quorum_queue is enabled, Default 0 which means dont set a limit.

rabbit_quorum_max_memory_length

Type: integer
Default: 0

By default all messages are maintained in memory if a quorum queue grows in length it can put memory pressure on a cluster. This option can limit the number of messages in the quorum queue. Used only when rabbit_quorum_queue is enabled, Default 0 which means dont set a limit.

Deprecated Variations

Group	Name
oslo_messaging_rabbit	rabbit_quroum_max_memory_length

rabbit_quorum_max_memory_bytes

Type: integer
Default: 0

By default all messages are maintained in memory if a quorum queue grows in length it can put memory pressure on a cluster. This option can limit the number of memory bytes used by the quorum queue. Used only when rabbit_quorum_queue is enabled, Default 0 which means dont set a limit.

Deprecated Variations

Group	Name
oslo_messaging_rabbit	rabbit_quroum_max_memory_bytes

rabbit_transient_queues_ttl

Type: integer
Default: 1800
Minimum Value: 0

Positive integer representing duration in seconds for queue TTL (x-expires). Queues which are unused for the duration of the TTL are automatically deleted. The parameter affects only reply and fanout queues. Setting 0 as value will disable the x-expires. If doing so, make sure you have a rabbitmq policy to delete the queues or you deployment will create an infinite number of queue over time.In case rabbit_stream_fanout is set to True, this option will control data retention policy (x-max-age) for messages in the fanout queue rather then the queue duration itself. So the oldest data in the stream queue will be discarded from it once reaching TTL Setting to 0 will disable x-max-age for stream which make stream grow indefinitely filling up the diskspace

rabbit_qos_prefetch_count

Type: integer
Default: 0

Specifies the number of messages to prefetch. Setting to zero allows unlimited messages.

heartbeat_timeout_threshold

Type: integer
Default: 60

Number of seconds after which the Rabbit broker is considered down if heartbeat's keep-alive fails (0 disables heartbeat).

heartbeat_rate

Type: integer
Default: 3

How often times during the heartbeat_timeout_threshold we check the heartbeat.

direct_mandatory_flag

Type: boolean
Default: True

(DEPRECATED) Enable/Disable the RabbitMQ mandatory flag for direct send. The direct send is used as reply, so the MessageUndeliverable exception is raised in case the client queue does not exist.MessageUndeliverable exception will be used to loop for a timeout to lets a chance to sender to recover.This flag is deprecated and it will not be possible to deactivate this functionality anymore

WARNING:

This option is deprecated for removal. Its value may be silently ignored in the future.

Reason: Mandatory flag no longer deactivable.

enable_cancel_on_failover

Type: boolean
Default: False

Enable x-cancel-on-ha-failover flag so that rabbitmq server will cancel and notify consumerswhen queue is down

use_queue_manager

Type: boolean
Default: False

Should we use consistant queue names or random ones

hostname

Type: string
Default: node1.example.com

This option has a sample default set, which means that its actual default value may vary from the one documented above.

Hostname used by queue manager. Defaults to the value returned by socket.gethostname().

processname

Type: string
Default: nova-api

This option has a sample default set, which means that its actual default value may vary from the one documented above.

Process name used by queue manager

rabbit_stream_fanout

Type: boolean
Default: False

Use stream queues in RabbitMQ (x-queue-type: stream). Streams are a new persistent and replicated data structure ("queue type") in RabbitMQ which models an append-only log with non-destructive consumer semantics. It is available as of RabbitMQ 3.9.0. If set this option will replace all fanout queues with only one stream queue.

API

oslo_messaging.opts.list_opts()

Return a list of oslo.config options available in the library.

The returned list includes all oslo.config options which may be registered at runtime by the library.

Each element of the list is a tuple. The first element is the name of the group under which the list of elements in the second element will be registered. A group name of None corresponds to the [DEFAULT] group in config files.

This function is also discoverable via the 'oslo_messaging' entry point under the 'oslo.config.opts' namespace.

The purpose of this is to allow tools like the Oslo sample config file generator to discover the options exposed to users by this library.

Returns: a list of (group_name, opts) tuples

Testing Configurations

class oslo_messaging.conffixture.ConfFixture(conf, transport_url=None)

Tweak configuration options for unit testing.

oslo.messaging registers a number of configuration options, but rather than directly referencing those options, users of the API should use this interface for querying and overriding certain configuration options.

An example usage:

self.messaging_conf = self.useFixture(messaging.ConfFixture(cfg.CONF))
self.messaging_conf.transport_url = 'fake:/'

Parameters

conf (oslo.config.cfg.ConfigOpts) -- a ConfigOpts instance
transport_url (str) -- override default transport_url value

property response_timeout: Default number of seconds to wait for a response from a call.

setUp()

Prepare the Fixture for use.

This should not be overridden. Concrete fixtures should implement _setUp. Overriding of setUp is still supported, just not recommended.

After setUp has completed, the fixture will have one or more attributes which can be used (these depend totally on the concrete subclass).

Raises: MultipleExceptions if _setUp fails. The last exception captured within the MultipleExceptions will be a SetupError exception.
Returns: None.
Changed in 1.3: The recommendation to override setUp has been reversed - before 1.3, setUp() should be overridden, now it should not be.
Changed in 1.3.1: BaseException is now caught, and only subclasses of Exception are wrapped in MultipleExceptions.

property transport_url: The transport url

DEPLOYMENT GUIDE

Available Drivers

fake

Fake driver used for testing.

This driver passes messages in memory, and should only be used for unit tests.

kafka

Kafka Driver

See Kafka Driver Deployment Guide for details.

kombu

RabbitMQ Driver

The rabbit driver is the default driver used in OpenStack's integration tests.

The driver is aliased as kombu to support upgrading existing installations with older settings.

rabbit

RabbitMQ Driver

The rabbit driver is the default driver used in OpenStack's integration tests.

The driver is aliased as kombu to support upgrading existing installations with older settings.

Kafka Driver Deployment Guide

Introduction

The Kafka Driver is an experimental messaging transport backend in oslo.messaging. The driver maps the base oslo.messaging capabilities for notification message exchange onto v2.0 of the Apache Kafka distributed streaming platform. More detail regarding the Apache Kafka server is available from the Apache Kafka website.

More detail regarding the driver's implementation is available from the adding kafka driver specification and the update kafka driver specification.

Overview

The Kafka driver only supports use for sending and receiving oslo.messaging notifications. Specifically, the Kafka driver does not support oslo.messaging RPC transfers. Communications between the driver and Kafka server backend uses a binary protocol over TCP that defines all APIs as request response message pairs. The Kafka driver integrates the confluent-kafka Python client for full protocol support and utilizes the Producer API to publish notification messages and the Consumer API for notification listener subscriptions. The driver is able to work with a single instance of a Kafka server or a clustered Kafka server deployment.

Hybrid Messaging Deployment

Oslo.messaging provides a mechanism to configure separate messaging backends for RPC and notification communications. This is supported through the definition of separate RPC and notification transport urls in the service configuration. When the Kafka driver is deployed for oslo.messaging notifications, a separate driver and messaging backend must be deployed for RPC communications. For these hybrid messaging configurations, the rabbit drivers can be deployed for oslo.messaging RPC.

Topics and vhost Support

The Kafka topic is the feed name to which records are published. Topics in Kafka are multi-subscriber such that a topic can have zero, one or many consumers that subscribe to the data written to it. In oslo.messaging, a notification listener subscribes to a topic in a supplied target that is directly mapped by the driver to the Kafka topic. The Kafka server architecture does not natively support vhosts. In order to support the presence of a vhost in the transport url provided to the driver, the topic created on the Kafka server will be appended with the virtual host name. This creates a unique topic per virtual host but note there is otherwise no access control or isolation provided by the Kafka server.

Listener Pools

The Kafka driver provides support for listener pools. This capability is realized by mapping the listener pool name to a Kafka server consumer group name. Each record published to a topic will be delivered to one consumer instance within each subscribing pool (e.g. consumer group). If a listener pool name is not assigned to the notification listener, a single default consumer group will be used by the Kafka driver and all listeners will be assigned to that group and the messages will effectively be load balanced across the competing listener instances.

Synchronous Commit

A primary functional difference between a Kafka server and a classic broker queue is that the offset or position of the message read from the commit log is controlled by the listener (e.g. consumer). The driver will advance the offset it maintains linearly as it reads message records from the server. To ensure that duplicate messages are not generated during downtime or communication interruption, the driver will synchronously commit the consumed messages prior to the notification listener dispatch. Due to this, the driver does not support the re-queue operation and the driver can not replay messages from a Kafka partition.

Prerequisites

In order to run the driver the confluent-kafka Python client must be installed. The Kafka driver integrates a Python client based on librdkafka for full protocol support and utilizes the Producer API to publish notification messages and the Consumer API for notification listener subscriptions.

Source packages for the confluent-kafka library are available via PyPI. Since the Kafka driver is an optional extension to oslo.messaging these packages are not installed by default. Use the kafka extras tag when installing oslo.messaging in order to pull in these extra packages:

$ python -m pip install oslo.messaging[kafka]

Configuration

Transport URL Enable

In oslo.messaging, the transport_url parameters define the OpenStack service backends for RPC and Notify. The URL is of the form:

transport://user:pass@host1:port[,hostN:portN]/virtual_host

Where the transport value specifies the RPC or notification backend as one of rabbit, kafka, etc. To specify and enable the Kafka driver for notifications, in the section [oslo_messaging_notifications] of the service configuration file, specify the transport_url parameter:

[oslo_messaging_notifications]
transport_url = kafka://username:password@kafkahostname:9092

Note, that if a transport_url parameter is not specified in the [oslo_messaging_notifications] section, the value of [DEFAULT] transport_url will be used for both RPC and notification backends.

Driver Options

It is recommended that the default configuration options provided by the Kafka driver be used. The configuration options can be modified in the oslo_messaging_kafka section of the service configuration file.

Notification Listener Options

oslo_messaging_kafka.kafka_max_fetch_bytes
oslo_messaging_kafka.kafka_consumer_timeout
oslo_messaging_kafka.consumer_group
oslo_messaging_kafka.enable_auto_commit
oslo_messaging_kafka.max_poll_records

Notifier Options

oslo_messaging_kafka.producer_batch_timeout
oslo_messaging_kafka.producer_batch_size

compression_codec: The compression codec for all data generated by the producer, valid values are: none, gzip, snappy, lz4, zstd. Note that the legal option of this depends on the kafka version, please refer to kafka documentation.

Security Options

oslo_messaging_kafka.security_protocol
oslo_messaging_kafka.sasl_mechanism
oslo_messaging_kafka.ssl_cafile

DevStack Support

The plugin for the Kafka oslo.messaging driver is supported by DevStack. As the Kafka driver can only be deployed for notifications, the plugin supports the deployment of several message bus configurations. In the [localrc] section of local.conf, the devstack-plugin-kafka plugin repository must be enabled. For example:

[[local|localrc]]
enable_plugin kafka https://opendev.org/openstack/devstack-plugin-kafka

Set the Kafka and Scala version and location variables if needed for the configuration

KAFKA_VERSION=2.0.0
KAFKA_BASEURL=http://archive.apache.org/dist/kafka
SCALA_VERSION=2.12
SCALA_BASEURL=http://www.scala-lang.org/riles/archive

The RPC_ and NOTIFY_ variables will define the message bus configuration that will be used. The hybrid configurations will allow for the rabbit drivers to be used for the RPC transports while the kafka driver will be used for the notification transport. The setting of the service variables will select which messaging intermediary is enabled for the configuration:

	RPC	NOTIFY
SERVICE	PORT	SERVICE	PORT
Config 1	rabbit	5672	kafka	9092

RabbitMq Driver Deployment Guide

Introduction

The RabbitMQ Driver is a messaging transport backend in oslo.messaging. The driver maps the base oslo.messaging capabilities for notification message exchange onto the RabbitMQ distributed message broker. More detail regarding the RabbitMQ server is available from the RabbitMQ website.

Abstract

The RabbitMQ Driver is a messaging transport backend supported in oslo.messaging. Communications between the driver and RabbitMQ server backend uses the AMQP 0-9-1 protocol (Advanced Message Queuing Protocol) which is an open application layer that allows clients applications to communicate with messaging middleware brokers in a standard way. AMQP defines all APIs to request messages from a message queue or to publishes messages to an exchange. The RabbitMQ driver integrates the kombu python client for full protocol support and utilizes the Producer API to publish notification messages and the Consumer API for notification listener subscriptions. The driver is able to work with a single instance of a RabbitMQ server or a clustered RabbitMQ server deployment.

Exchange

Exchange is a AMQP mechanism that it designed to dispatch the messages like a proxy wrapping.

Messages are always published to exchanges.

An exchange can:

receives messages from producers
push messages to queues

Exchanges can distribute message copies to one or many queues using rules called bindings.

AMQP protocol defines different types of exchanges:

direct
topic
headers
fanout

An exchange can live without any binding. By default, no exception is raised if the message is not redirected to any queue, unless the mandatory flag is used.

oslo.messaging allow you to send and consume messages in a related manner through the Connection class.

With mandatory flag RabbitMQ raises a callback if the message is not routed to any queue. This callback will be used to loop for a timeout and let's a chance to sender to recover.

Queue

The AMQP queue is the messages store, it can store the messages in memory or persist the messages to the disk.

The queue is bound to the exchange through one or more bindings.

Consumers can consume messages from the queue.

Queues have names so that applications can reference them.

Queues have properties that define how they behave.

Routing-Key

The routing key is part of the AMQP envelop of the message. The routing key is set by the producer to route the sended message. When a message is received, the exchange will try to match the message routing-key with the binding key of all the queues bound to it. If no match exist the message will be ignored else the message will be routed to the corresponding queue who binding key is matched.

Exchange types

direct

A direct exchange is an exchange which route messages to queues based on message routing key. Message will be directly delivered to the queue that correspond to the routing-key. direct is a type of exchange so RabbitMQ backend does not store the data.

topic

The RabbitMQ topic is an exchange type. In RabbitMQ messages sent to a topic exchange can't have an arbitrary routing_key - it must be a list of words, delimited by dots. The words can be anything, but usually they specify some features connected to the message. A few valid routing key examples: "blue.orange.yellow", "cars.bikes", "quick.orange.rabbit". There can be as many words in the routing key as you like, up to the limit of 255 bytes.. In oslo.messaging, a notification listener subscribes to a topic in a supplied target that is directly mapped by the driver to the RabbitMQ topic.

fanout

The fanout exchange will broadcasts all messages it receives to all the queues it knows.

headers

An headers exchange will route messages to queues by using message header content instead of routing by using the routing key like described previously. In this case producer will adds message header values as key-value pair in and he will sends by using the headers exchange.

The exchange will try to match all or any (based on the value of “x-match”) header value of the received message with the binding value of all the queues bound to it.

The exchange will determine which header to use to match the binding queues by using the value of the "x-match" header entry.

If match is not found, the message will be ignored else it route the message to the queue whose binding value is matched.

Health check with heartbeat frames

The RabbitMQ driver of oslo.messaging allow you to detect dead TCP connections with heartbeats and TCP keepalives. The heartbeat function from the driver is build over the heartbeat_check feature of kombu client and over the AMQP 0.9.1 heartbeat feature implemented by RabbitMQ. Heartbeating is a technique designed to undo one of TCP/IP's features, namely its ability to recover from abroken physical connection by closing only after a quite long time-out. In some scenarios we need to knowvery rapidly if a peer is disconnected or not responding for other reasons (e.g. it is looping). Since heart-beating can be done at a low level, AMQP 0.9.1 implement this as a special type of frame that peers exchange at the transport level, rather than as a class method. Heartbeats also defend against certain network equipment which may terminate "idle" TCP connections when there is no activity on them for a certain period of time. The driver will always run the heartbeat in a native python thread and avoid to inherit the execution model from the parent process to avoid to use green threads.

Prerequisites

In order to run the driver the kombu python client must be installed. The RabbitMQ driver integrates a Python client based on kombu and on py-amqp for full protocol support and utilizes the Producer API to publish notification messages and the Consumer API for notification listener subscriptions.

Source packages for the kombu library are available via PyPI. Since the RabbitMQ driver is not an optional extension to oslo.messaging these packages installed by default.

Configuration

Transport URL Enable

In oslo.messaging, the transport_url parameter defines the OpenStack service backends for RPC and notifications. The URL is of the form:

transport://user:pass@host1:port[,hostN:portN]/virtual_host

Where the transport value specifies the RPC or notification backend as one of amqp, rabbit, kafka, etc. To specify and enable the RabbitMQ driver for notifications, in the section [oslo_messaging_notifications] of the service configuration file, specify the transport_url parameter:

[oslo_messaging_notifications]
transport_url = rabbit://username:password@kafkahostname:9092

Note, that if a transport_url parameter is not specified in the [oslo_messaging_notifications] section, the [DEFAULT] transport_url option will be used for both RPC and notifications backends.

Note about Quorum and Stream queues

Recent RabbitMQ releases (3.8 and above) introduced Quorum and Stream queue types which are going to replace the classic HA queues.

It's highly recommended that you switch whenever possible to quorum and stream queues because classic HA queues are going to be deprecated.

The recommended options are the following:

[oslo_messaging_rabbit] rabbit_quorum_queue = True rabbit_transient_quorum_queue = True use_queue_manager = True rabbit_stream_fanout = True

Driver Options

It is recommended that the default configuration options provided by the RabbitMQ driver be used. The configuration options can be modified in the oslo_messaging_rabbit section of the service configuration file.

Publishing Options

oslo_messaging_rabbit.kombu_compression
oslo_messaging_rabbit.kombu_missing_consumer_retry_timeout

Consuming Options

oslo_messaging_rabbit.rabbit_ha_queues
oslo_messaging_rabbit.rabbit_quorum_queue
oslo_messaging_rabbit.rabbit_quorum_delivery_limit
oslo_messaging_rabbit.rabbit_quorum_max_memory_length
oslo_messaging_rabbit.rabbit_quorum_max_memory_bytes
oslo_messaging_rabbit.rabbit_transient_queues_ttl
oslo_messaging_rabbit.rabbit_transient_quorum_queue
oslo_messaging_rabbit.use_queue_manager
oslo_messaging_rabbit.hostname
oslo_messaging_rabbit.processname
oslo_messaging_rabbit.rabbit_stream_fanout

Connection Options

oslo_messaging_rabbit.kombu_reconnect_delay
oslo_messaging_rabbit.kombu_failover_strategy
oslo_messaging_rabbit.rabbit_retry_interval
oslo_messaging_rabbit.rabbit_retry_backoff
oslo_messaging_rabbit.rabbit_interval_max
oslo_messaging_rabbit.rabbit_qos_prefetch_count

Heartbeat Options

oslo_messaging_rabbit.heartbeat_timeout_threshold
oslo_messaging_rabbit.heartbeat_rate

Security Options

oslo_messaging_rabbit.ssl
oslo_messaging_rabbit.ssl_version
oslo_messaging_rabbit.ssl_key_file
oslo_messaging_rabbit.ssl_cert_file
oslo_messaging_rabbit.rabbit_login_method
oslo_messaging_rabbit.ssl_enforce_fips_mode

USING OSLO.MESSAGING

Frequently Asked Questions

I don't need notifications on the message bus. How do I disable them?

Notification messages can be disabled using the noop notify driver. Set driver = noop in your configuration file under the [oslo_messaging_notifications] section.

Why does the notification publisher create queues, too? Shouldn't the subscriber do that?

The notification messages are meant to be used for integration with external services, including services that are not part of OpenStack. To ensure that the subscriber does not miss any messages if it starts after the publisher, oslo.messaging ensures that subscriber queues exist when notifications are sent.

How do I change the queue names where notifications are published?

Notifications are published to the configured exchange using a topic built from a base value specified in the configuration file and the notification "level". The default topic is notifications, so an info-level notification is published to the topic notifications.info. A subscriber queue of the same name is created automatically for each of these topics. To change the queue names, change the notification topic using the topics configuration option in [oslo_messaging_notifications]. The option accepts a list of values, so it is possible to publish to multiple topics.

What are the other choices of notification drivers available?

messaging Send notifications using the 1.0 message format.
messagingv2 Send notifications using the 2.0 message format (with a message envelope).
routing Configurable routing notifier (by priority or event_type).
log Publish notifications via Python logging infrastructure.
test Store notifications in memory for test verification.
noop Disable sending notifications entirely.

CHANGES

16.1.0

Skip installation to speed up pep8
Fixed: use self.start_time in QManager shm
Added Docs for Simulator
rabbit: validate input for retry options

16.0.0

Deprecate the Eventlet executor
Remove deprecated aliases of notification options
Remove remaining functional test code for AMQP1 driver
Fix outdated job description
rabbit: Clean up deprecated options
rabbit: Fix wrong section to render connection pool options
kafka: Remove unused connection pool options
Remove pyngus from dependencies
Add a kombu_reconnect_splay setting
Revert "Remove explicit pbr dependency"
reno: Update master for unmaintained/2023.1

15.0.0

Remove deprecated AMQP1 driver
Remove explicit pbr dependency
Add note about requirements lower bounds
Skip grenade job on doc updates
Skip functional tests on pre-commit config update
Run pyupgrade to clean up Python 2 syntaxes
pre-commit: Bump versions
Remove Python 3.8 support
[rabbit] Ignore stream offset header when stream queue is not used
Declare Python 3.12 support
Use oslo.utils implementation to parse/escape server address
Add a note about quorum and stream queues
Fix queue_manager in a containerized environment
Update master for stable/2024.2
[doc] Better document rabbit_transient_queues_ttl option
Use more sensible "default" for [oslo_messaging_rabbit] processname
doc: Fix rendering in the Notifier reference
Remove comment reference to removed function

14.9.0

Reduce log level to DEBUG for new messages
Deprecate the option heartbeat_in_pthread
Add RPC incoming and reply log

14.8.1

Handle NotFound exception when declaring a queue
rabbit: Make PreconditionFailed warning log more informative

14.8.0

reno: Update master for unmaintained/zed
Remove old excludes
Make oslo.messaging, magnum, and zaqar reproducible
Fix incorrect desc of rabbit_stream_fanout option
Update master for stable/2024.1
reno: Update master for unmaintained/xena
reno: Update master for unmaintained/wallaby
reno: Update master for unmaintained/victoria
kafka: Fix invalid hostaddr format for IPv6 address
Use StopWatch timer when waiting for message
Restore read stream queues from last known offset

14.7.0

Display coverage report
reno: Update master for unmaintained/yoga
Display the reply queue's name in timeout logs
Bump hacking (again)
Bump hacking
Remove scenario 03/04 tests from CI
Drop unused function from scenario test script
Utilize the new RequestContext redacted_copy method
cleanup amqp1 scenarios remnants setups
Add an option to use rabbitmq stream for fanout queues
Update the python search path for extra qdrouter modules
Update python classifier in setup.cfg

14.6.0

Remove translation sections from setup.cfg
Fix clearing of the fake RPC Exchange
Add QManager to amqp driver
Enable use of quorum queues for transient messages
Auto-delete the failed quorum rabbit queues
Allow creating transient queues with no expire

14.5.0

Add some logs when sending RPC messages
Move integration jobs to Ubuntu 22.04 (Jammy)
Imported Translations from Zanata
test py311 job
Add is_admin to safe fields list for notifications
Update master for stable/2023.2

14.4.0

Add Python3 antelope unit tests
Only allow safe context fields in notifications
Set default heartbeat_rate to 3
Imported Translations from Zanata
Bump bandit and make oslo.messaging compatible with latest rules

14.3.1

•: Increase ACK_REQUEUE_EVERY_SECONDS_MAX to exceed default kombu_reconnect_delay

14.3.0

Disable greenthreads for RabbitDriver "listen" connections
Update master for stable/2023.1
Fix typo in quorum-related variables for RabbitMQ

14.2.0

Support overriding class for get_rpc_* helper functions
tox cleanups

14.1.0

Implement get_rpc_client function
Warn when we force creating a non durable exchange
Deprecate the amqp1 driver and Remove qpid functional tests
Update master for stable/zed

14.0.0

Change default value of "heartbeat_in_pthread" to False
Remove logging from ProducerConnection._produce_message

13.0.0

update hacking pin to support flake8 3.8.3
Drop python3.6/3.7 support in testing runtime

12.14.0

Add EXTERNAL as rabbit login method
Add quorum queue control configurations
Bump bandit
tests: Fix test failures with kombu >= 5.2.4
Add Python3 zed unit tests
Update master for stable/yoga

12.13.0

•: Adding support for rabbitmq quorum queues

12.12.0

[rabbit] use retry parameters during notification sending
Update python testing classifier
Force creating non durable control exchange when a precondition failed
Reproduce bug 1917645

12.11.1

amqp1: fix race when reconnecting
Add a new option to enforce the OpenSSL FIPS mode

12.11.0

•: Remove deprecation of heartbeat_in_pthread

12.10.0

rabbit: move stdlib_threading bits into _utils
Add Python3 yoga unit tests
Update master for stable/xena
use message id cache for RPC listener

12.9.1

amqp1: Do not reuse _socket_connection on reconnect
amqp1: re-organize TestFailover to be reused by TestSSL
Revert "Disable AMQP 1.0 SSL unit tests"

12.9.0

limit maximum timeout in the poll loop
Add Support For oslo.metrics
Changed minversion in tox to 3.18.0
Upgrade the pre-commit-hooks version
setup.cfg: Replace dashes with underscores
Remove the oslo_utils.fnmatch

12.8.0

Remove references to 'sys.version_info'
Fix formatting of release list
Move flake8 as a pre-commit local target
Add Python3 xena unit tests
Update master for stable/wallaby

12.7.1

Remove lower constraints
Correctly handle missing RabbitMQ queues

12.7.0

•: Deprecate the mandatory flag

12.6.1

remove unicode from code
Remove six
Fix type of direct_mandatory_flag opt
Dropping lower constraints testing
Move jobs to py38
fix variable name
Fix doc title rendering
Use TOX_CONSTRAINTS_FILE
Use py3 as the default runtime for tox

12.6.0

Python 3.9: isAlive is removed
add min of 1 to rpc_conn_pool_size
Adding pre-commit
Add Python3 wallaby unit tests
Update master for stable/victoria

12.5.0

•: [goal] Migrate testing to ubuntu focal

12.4.0

Run rabbitmq heartbeat in python thread by default
Add a ping endpoint to RPC dispatcher

12.3.0

Cancel consumer if queue down
Bump bandit version

12.2.2

Move legacy grenade jobs to Zuul v3
Catch ConnectionForced Exception

12.2.1

tests: Resolves issues with kombu > 4.6.8
Simplify tools/test-setup.sh
Drop a python 2 exception management
Fix pygments style
bindep: Add 'librdkafka-dev' dependency

12.2.0

Fix hacking min version to 3.0.1
Switch to newer openstackdocstheme and reno versions
Remove the unused coding style modules
Print warning message when connection running out
Remove six usage
Remove monotonic usage
Align contributing doc with oslo's policy
Bump default tox env from py37 to py38
Add py38 package metadata
Add release notes links to doc index
Imported Translations from Zanata
Add Python3 victoria unit tests
Update master for stable/ussuri
Fix some typos

12.1.0

•: Update hacking for Python3

12.0.0

Remove the deprecated blocking executor
remove outdated header
reword releasenote for py27 support dropping
Setup backend scenarios for functional tests

11.0.0

[ussuri][goal] Drop python 2.7 support and testing
Don't log NoSuchMethod for special non-existing methods
Add support for kafka SSL autentication
Adding debug logs on AMQPListener poll
tox: Trivial cleanup

10.5.0

Ignore releasenote cache within git untracked files
Do not use threading.Event
Removed unused variable pools

10.4.1

Revert "Add RPC incoming and reply log"
Remove telemetry checks

10.4.0

Migrate grenade jobs to py3
Make sure minimum amqp is 2.5.2

10.3.0

Remove unused variable WAKE_UP
Switch to Ussuri jobs
tox: Keeping going with docs
Add RPC incoming and reply log
Modify some comments to make them clickable
Fix spacing in help message
Update the constraints url
Align message serialization between drivers
Update master for stable/train
Fix help text for heartbeat_in_pthread option

10.2.0

•: Add the mandatory flag for direct send

10.1.0

Introduce RabbitMQ driver documentation
Allow users run the rabbitmq heartbeat inside a standard pthread
Fix nits on kafka compression help text

10.0.0

Correcting typo in acknowledge spelling
Bump the openstackdocstheme extension to 1.20

9.8.0

Add Python 3 Train unit tests
Blacklist sphinx 2.1.0 (autodoc bug)
Use default exchange for direct messaging
doc: Cleanup admin docs
Implement mandatory flag for RabbitMQ driver
Implement the transport options
Add the "transport_options" parameter to the amqp1 and kafka drivers
Support kafka message compression
fix typos
Add transport_options parameter
Download kafka from archive.apache.org

9.7.2

9.7.1

Add thread name to the RabbitMQ heartbeat thread
Add help msg to payload for CLI notifier

9.7.0

Cap Bandit below 1.6.0 and update Sphinx requirement
Fix switch connection destination when a rabbitmq cluster node disappear
Replace git.openstack.org URLs with opendev.org URLs
Remove log translation and i18n

9.6.0

OpenDev Migration Patch
Dropping the py35 testing
Consider the topic parameter as an array in client-notify
Handle collections.abc deprecations
Retry to declare a queue after internal error
Unmark RabbitMQ heartbeat as experimental
Replace openstack.org git:// URLs with https://
Explain why Listener connections cannot be pooled
Update master for stable/stein
Clarify the documentation for pooled Notification Listeners
Update messaging intermediaries for amqp1 tests

9.5.0

Handle unexpected failures during call monitor heartbeat
add python 3.7 unit test job
Mark telemetry tests nv and remove from gate
Bump amqp requirement version to >=2.4.1
Change python3.5 job to python3.7 job on Stein+

9.4.0

Bump amqp requirement version to >= 2.4.0
Kafka driver deployment guide
Update hacking version

9.3.1

•: Avoid unnecessary use of items()

9.3.0

Update mailinglist from dev to discuss
Switch driver to confluent-kafka client library
Don't use monotonic with Python >=3.3

9.2.1

•: Use ensure_connection to prevent loss of connection error logs

9.2.0

•: Add a test for rabbit URLs lacking terminating '/'

9.1.1

•: Use '/' for the vhost if the transport_url has no trailing '/'

9.1.0

Using pip as a python module
doc: Remove crud from conf.py file
Clean up .gitignore references to personal tools
Allow transport_url initialization in ConfFixture constructor

9.0.1

Fix oslo.messaging default transport
always build universal wheels
Use default exchange for direct messaging

9.0.0

Refactor GetTransportSadPathTestCase
Add release note about deprecated option removals
Remove rpc_backend and ConfFixture.transport_driver
Remove deprecated rabbit options
Use templates for cover and lower-constraints
Remove deprecated amqp1 options
Remove rabbit_durable_queues deprecated option
Remove default_{host,port} deprecated options
Remove the deprecated ZeroMQ driver
Fix the coverage tox tests
Avoid logging passwords on connection events
add lib-forward-testing-python3 test job
add python 3.6 unit test job
import zuul job settings from project-config
Call listener stop only if listener is initialized
Update reno for stable/rocky
Remove setting of DEVSTACK_GATE_EXERCISES

8.1.0

Bump py-amqp to >= 2.3.0
Issue blocking ACK for RPC requests from the consumer thread

8.0.0

Do not access the connection's socket during error callback
Fix debug log message - missing argument
py37: deal with Exception repr changes
py37: drop use of 'async' as parameter name
Remove transport aliases support
Moving stestr to correct package order in test-requirements.txt
Switch to stestr
No longer allow redundant calls to server start()
Fix the bandit security linter test

7.0.0

Replace 'raise StopIteration' with 'return'
Remove fake_rabbit configuration option
Add release notes link to README
Add ZeroMQ deprecation release note

6.5.0

Fix oslo messaging gating
Enable RPC call monitoring in AMQP 1.0 driver
Mark the ZeroMQ driver deprecated

6.4.1

fix tox python3 overrides
Add warning output if failed to rebuild exception when deserialize
Correct usage of transport_url in example

6.4.0

•: Add ConfFixture.transport_url

6.3.0

Convert legacy zuul jobs to v3
[rabbitmq] Implement active call monitoring
Make oslo.messaging-devstack-amqp1 job non-voting
Remove stale pip-missing-reqs tox test
Add a skeleton for a v3-native devstack job
Add heartbeat() method to RpcIncomingMessage
Trivial: Update pypi url to new url
Add kafka for python 3 functional test

6.2.0

Move requirements for the optional drivers (amqp1, kafka)
set default python to python3
fix lower constraints and uncap eventlet

6.1.0

Revert "rabbit: Don't prefetch when batch_size is set"
Update kafka and dsvm jobs
add lower-constraints job
remove zmq tests
Updated from global requirements

6.0.0

Remove the deprecated Pika driver
update configuration for qdrouter v1.0.0
Updated from global requirements
Add restart() method to DecayingTimer

5.36.0

Imported Translations from Zanata
Add rabbitmq-server for platform:rpm into bindep.txt
Restore devstack project name in amqp1 test
Switch from pip_missing_reqs to pip_check_reqs
Add kafka config options for security (ssl/sasl)
Zuul: Remove project name
Modify grammatical errors
Fixed telemetry integration zuul jobs
Zuul: Remove project name
Updated from global requirements
Imported Translations from Zanata
Update reno for stable/queens
Updated from global requirements
Imported Translations from Zanata
Add support for synchronous commit
Update telemetry integration playbooks
Follow the new PTI for document build

5.35.0

Add kafka driver vhost emulation
Updated from global requirements
Create doc/requirements.txt
Update kafka functional test
Imported Translations from Zanata
Updated from global requirements

5.34.1

Imported Translations from Zanata
Avoid tox_install.sh for constraints support
rabbitmq: don't wait for message ack/requeue
Provide bindep_profile in openstack-tox job setup
Updated from global requirements
Add zmq packages that are no longer in bindep-fallback
don't convert generator to list unless required
sort when using groupby

5.34.0

Remove setting of version/release from releasenotes
Updated from global requirements
Updated from global requirements
Catch socket.timeout when doing heartbeat_check
Updated from global requirements
fix batch handling
Remove stable/newton from zuul settings
Zuul: add file extension to playbook path

5.33.1

Move legacy zuulv3 tests into oslo.messaging repo
Imported Translations from Zanata
Flesh out transport_url help
Fix typo in contributor docs title

5.33.0

Fix default value of RPC dispatcher access_policy
Fix wrong transport warnings in functional tests
Updated from global requirements

5.32.0

Updated from global requirements
Warn when wrong transport instance is used
Fix some reST field lists in docstrings
Remove pbr version from setup.py
Suppress excessive debug logs when consume rabbit
Fix use of print function on python3

5.31.0

Remove envelope argument from driver send() interface
Imported Translations from Zanata
Updated from global requirements
Update amqp 1.0 driver deployment guide
Prevent rabbit from raising unexpected exceptions
Updated from global requirements
Remove unnecessary setUp function in testcase
Add licenses and remove unused import in doc/source/conf.py
Ensure RPC endpoint target attribute is correct
Fix a typo
Update links in README
Updated from global requirements
Class-level _exchanges in FakeExchangeManager
fix 'configration' typo
Update reno for stable/pike
Add support for virtual hosts
Remove the test that counts kombu connect calls

5.30.0

Updated from global requirements
Update URLs in documents according to document migration
Add monkey_patch to demo code

5.29.0

switch from oslosphinx to openstackdocstheme
update the docs url in the readme
rearrange content to fit the new standard layout
Updated from global requirements
Enable some off-by-default checks

5.28.0

Updated from global requirements
Add kafka_driver directory

5.27.0

Updated from global requirements
Fix html_last_updated_fmt for Python3
Add note for blocking executor deprecation
Fix rabbitmq driver with blocking executor
Build universal wheels
Updated from global requirements
Fix serializer tests
deprecated blocking executor

5.26.0

Updated from global requirements
Clean up the TransportURL documentation
Mark the Pika driver as deprecated

5.25.0

Updated from global requirements
Updated from global requirements
Add missing {posargs:} to AMQP 1.0 functional tests
rabbit: restore synchronous ack/requeue

5.24.2

Updated from global requirements
[AMQP 1.0] Properly shut down test RPC server

5.24.1

Updated from global requirements
Fix the amqp1 SSL test CA certificate
Add get_rpc_transport call
Disable AMQP 1.0 SSL unit tests

5.24.0

5.23.0

Fix notification tests not unmocking logging
Remove use of mox stubs
Fix aliases deprecation
tests: fix MultiStrOpt value
Retry support for oslo_messaging_notifications driver

5.22.0

[AMQP 1.0] Add default SASL realm setting
Updated from global requirements
Remove usage of parameter enforce_type

5.21.0

Optimize the link address
[AMQP 1.0] if RPC call is configured as presettled ignore acks
Mock 'oslo_messaging.notify._impl_routing.LOG' in notifier tests
Updated from global requirements
Add "ssl" option for amqp driver
Refactor logic of getting exector's executor_thread_pool_size
remove all kombu<4.0.0 workarounds

5.20.0

serializer: remove deprecated RequestContextSerializer
Try to fix TestSerializer.test_call_serializer failed randomly
Updated from global requirements
Deprecate username/password config options in favor of TRANSPORT_URL
Add HACKING.rst
Break user credentials from host at the rightmost '@'
[zmq] Prevent access to rpc_response_timeout
[zmq] pass a dummy TransportURL to register_opts
Fix simulator's use of Notifier - use 'topics' not 'topic'
Trivial: Add executor 'threading' in docstring
Deprecate parameter aliases
Use Sphinx 1.5 warning-is-error
tox: Build docs with Python 2.7

5.19.0

Updated from global requirements
Remove self.mox
Move decorator updated_kwarg_default_value to right place

5.18.0

Remove old messaging notify driver alias
[Fix gate]Update test requirement
Updated from global requirements
Allow checking if notifier is enabled
RabbitMQ: Standardize SSL parameter names
drop topic keyword from Notifier
Validate the transport url query string
drivers: use common.ConfigOptsProxy everywhere
Stop using oslotest.mockpatch
tests: don't run functional tests in parallel
rabbit: make ack/requeue thread-safe
Fix releasenotes
Remove mox3 from test-requirements.txt
Updated from global requirements
[zmq] Update configurations documentation
Fix type of the kafka_consumer_timeout option
[zmq] Dynamic connections send failure
support kombu4
Test:Use unittest.mock on Python 3
Fix the typo
pbr.version.VersionInfo needs package name (oslo.xyz and not oslo_xyz)
[zmq] Properly analyse `use_dynamic_connections` option
[zmq] Dummy add value aging mechanism
kafka: skip multiple servers tests
kafka: ensure topics are created
kafka: fix python3 exception
kafka: Remove testing hack for kafka
[zmq] Failure of dynamic connections fanout
Update reno for stable/ocata
Return list of addresses for IPV4 and IPV6

5.17.0

[zmq] Dynamic connections failover
[zmq] Fix py35 gate
[zmq] Use more stable configuration in voting job
Remove references to Python 3.4
[AMQP 1.0] Fix SSL client authentication
[zmq] Support py35 testenv
[zmq] Distinguish Round-Robin/Fanout socket sending mode
tests: cleanup monkey path
[AMQP 1.0] Resend messages that are released or modified
gate: Remove useless files
[zmq] Redis TTL for values
eventlet is no more a hard dependency
[AMQP 1.0] Propagate authentication errors to caller
ensure we set channel in lock
tox: use already installed kafka if present
kafka: remove no really implemented feature
kafka: return to poller when timeout is reach
kafka: Don't hide unpack/unserialize exception
kafka: timeout is in milliseconds
kafka: disable batch for functional tests
kafka: Remove Producer singleton
Moving driver to new kafka-python version
tox: rename zeromq target
tests: make rabbit failover failure more helpful
[zmq] Refactor make `zmq_address.target_to_key` a universal method
Updated from global requirements
[zmq] Restore static direct connections
reject when skipping failed messages
fix one typo
[AMQP 1.0] Setup the amqp1 test environment on ubuntu
test_rabbitmq: remove dead code

5.16.0

Updated from global requirements
Replace mox with mock
tests: fix test-setup.sh
tests: remove useless debug
[rabbit] Log correct topic on cast/call

5.15.0

Updated from global requirements
kafka separate unit/functionnal tests
Add bindep.txt/test-setup.sh to prepare the system
[zmq] Matchmaker redis available time

5.14.0

[AMQP 1.0] Simplify the I/O event loop code
[zmq] Support message versions for rolling upgrades
[zmq] Fix non voting gate jobs
Fix transport url with empty port
Remove ordering assumption from functional test
Periodically purge sender link cache

5.13.0

Remove small job timeout
Register opts if we're going to check conf.transport_url in parse()
[doc] Fix three typos
[zmq] Fix zmq-specific f-tests from periodic hangs
[zmq] Fix functional gates proxy/pub-sub
Show team and repo badges on README
[zmq] Send fanouts without pub/sub in background
Use assertGreater(len(x), 0) instead of assertTrue(len(x) > 0)
Add Constraints support
Replace six.iteritems() with .items()
[zmq] Fix configuration for functional gate job
Document the transport backend driver interface
Fix a docstring typo in impl_pika.py
[sentinel] Move master/slave discovering from __init__
rabbit: on reconnect set socket timeout after channel is set
Updated from global requirements
[zmq] Don't create real matchmaker in unit tests
update srouce doc pika_driver.rst the charactor then to than
Remove useless logging import statements
rabbit: Avoid busy loop on epoll_wait with heartbeat+eventlet
[zmq] Refactor receivers
[zmq] Cleanup changes to zmq-specific f-tests
Updated from global requirements
This patch cleans up the 'notification_listener.rst' documetion by removing some class which don't exist and adding some function which exist in current source
Remove nonexistent functions from documentation
Replace retrying with tenacity

5.12.0

Updated from global requirements
Updated from global requirements
Remove the temporary hack in code
Using assertIsNone() instead of assertEqual(None)
Change assertTrue(isinstance()) by optimal assert
[zmq] Don't fallback to topic if wrong server specified
[TrivialFix] Replace old style assertions with new style assertions
[TrivialFix] Fix typo in oslo.messaging
[simulator] Fix transport_url usage
[simulator] Fix a message length generator usage
Update .coveragerc after the removal of respective directory
[sentinels] Fix hosts extracting and slaves usage
[zmq] SUB-PUB local proxy

5.11.0

Fix typos in addressing.py and setup.cfg
Updated from global requirements
Record length of queues for ReplyWaiters
rabbit: Don't prefetch when batch_size is set
[AMQP 1.0] Avoid unnecessary thread switch on ack
[zmq] Fix issues with broken messages on proxies
[zmq] Maintain several redis hosts
Removed redundant 'the'
Fix a typo in server.py
[document] The example which is written in the developer guide of 'Notification Listener' doesn't work
Enable release notes translation
cast() and RPC replies should not block waiting for endpoint to ack
[simulator] Automatic stopping of rpc-servers
Fix whitespace formatting issue
Properly deserializes built-in exceptions
[zmq] Fix send_cast in AckManager
Remove debug logs from fast path
[zmq] Routing table refactoring, dynamic direct connections
Fix simulator bool command line args
Replace 'the' with 'to' in docstring
Remove default=None when set value in Config
[zmq] Add acks from proxy for PUB/SUB messages
[zmq] Refactor consumers and incoming messages
[zmq] Make second ROUTER socket optional for proxy
Use method fetch_current_thread_functor from oslo.utils
[zmq] Fix ZmqSocket.send_string
[zmq] Remove unused methods from executors
[zmq] Added a processing to handle ImportError in Redis plugin of Matchmaker
modify the home-page info with the developer documentation
Set the valid choices for the rabbit login methods
[zmq] Unify delimeters
[zmq] Fix fanout without PUB/SUB
[zmq] Send immediate ack after message receiving
Corrects documentation typo
[zmq] Remove unnecessary subscriptions from SubConsumer
Fixups to the inline documentation
Fix consuming from unbound reply queue
Add configurable serialization to pika
[zmq] Remove ZmqSocket.close_linger attribute
[zmq] Make ZMQ TCP keepalive options configurable
[zmq] Fix TestZmqAckManager periodic failure
[zmq] Make ThreadingPoller work with ZmqSocket
Fix notify filter when data item is None
[zmq] Rename rpc_cast_timeout option
[AMQP 1.0] Update setup test environment dispatch router backend
Allow dispatcher to restrict endpoint methods
[AMQP 1.0] Add Acknowledgement and Batch Notification Topics
Update reno for stable/newton
[kafka] invoke TypeError exception when 'listen()' method of KafkaDriver is called
[zmq] Proxy has to skip broken multi-part message
Add Documentation String for PikaDriver
[zmq] Implement retries for unacknowledged CALLs

5.10.0

[AMQP 1.0] Make the default settlement behavior configurable
[zmq] Eliminate GreenPool from GreenPoller
Avoid sending cast after server shutdown in functional test
[zmq] Update ZMQ-driver documentation
Updated from global requirements

5.9.0

[zmq] Add --log-file option to zmq-proxy
Updated from global requirements
[zmq] Host name and target in socket identity

5.8.0

[zmq] Make zmq_immediate configurable
Fix calculating of duration in simulator.py
[zmq] Redis unavailability is not critical
[zmq] Discover new publisher proxy
Clean outdated docstring and comment
[AMQP 1.0] small fixes to improve timer scalability
Add docstring for get_notification_transport
Add warning when credential is not specified for each host
Updated from global requirements
[zmq] Implement retries for unacknowledged CASTs
Fix the help info format

5.7.0

Move zmq driver options into its own group
Log a warning when connected to a routable message bus
Updated from global requirements
[AMQP 1.0] Add link credit configuration options
Updated from global requirements
[AMQP 1.0] AMQP 1.0 Driver User Guide Document update
AMQP 1.0 Driver Architecture Overview Document
Remove the max_send_retries option

5.6.0

Fix pika functional tests
[zmq] Use zmq.IMMEDIATE option for round-robin
fix a typo in impl_rabbit.py
Updated from global requirements
[AMQP 1.0] Cancel response treatment for detached link
Fix syntax error on notification listener docs
Delete fanout queues on gracefully shutdown
Set the default link property to allow message acks
Properly cleanup listener and driver on simulator exit
Fix a timer leak in the AMQP 1.0 driver
[zmq] Let proxy serve on a static port numbers
Introduce TTL for idle connections
Fix parameters of assertEqual are misplaced
Fix misstyping issue
Updated from global requirements
Updated from global requirements
notify: add a CLI tool to manually send notifications
Add deprecated relnote for max_retries rabbit configuration option
[zmq] Add py34 configuration for functional tests
[zmq] Merge publishers
Add Python 3.5 classifier and venv
Replace assertEqual(None, *) with assertIsNone in tests
Updated from global requirements
[zmq] Use json/msgpack instead of pickle
[AMQP 1.0] Add configuration parameters for send message deadline
[zmq] Refactor publishers
Re-factor the AMQP 1.0 addressing semantics
Add Python 3.4 functional tests for AMQP 1.0 driver
tests: allow to override the functionnal tests suite args
[zmq] Additional configurations for f-tests
Remove discover from test-requirements
tests: rabbitmq failover tests
[AMQP 1.0] Add acknowledge and requeue handling for incoming message
Imported Translations from Zanata
Updated from global requirements
Remove rabbitmq max_retries
Config: no need to set default=None

5.5.0

[zmq] Fix message sending when using proxy and not using PUB/SUB
AMQP 1.0 - create only one Cyrus SASL configuration for the tests
Updated from global requirements
Refactor AMQP 1.0 command task to support timers
[zmq] Remove redundant Envelope class
[zmq] Properly stop ZmqServer
Refactor link management to support link recovery
[Trival] fix a typo nit
[zmq] Fix backend router port for proxy

5.4.0

[zmq] Remove unused Request.close method
Add query paramereters to TransportURL
Fix temporary problems with pika unit tests
[zmq] Periodic updates of endpoints connections

5.3.0

Improve the impl_rabbit logging
Modify info of default_notification_exchange
Imported Translations from Zanata
[zmq] Remove rpc_zmq_concurrency option
[zmq] Fix timeout in ThreadingPoller.poll
Fix typo: 'olso' to 'oslo'
Updated from global requirements
[zmq] Don't skip non-direct message types
[zmq] Refactoring of zmq client
[impl_rabbit] Remove deprecated get_expiration method

5.2.0

Updated from global requirements
[AMQP 1.0] Randomize host list connection attempts
Modify the TransportURL's docstrings
Fix problems after refactoring RPC client
deprecate usage of transport aliases
Documents recommended executor
kafka: Deprecates host, port options
Updated from global requirements
Add reno for releasenotes management
Remove logging from serialize_remote_exception
[kafka] Add several bootstrap servers support
Add the proper branch back to .gitreview
Fix consuming from missing queues
Fix bug with version_cap and target.version in RPCClient
Make TransportURL.parse aware of transport_url
rabbit: Deprecates host, port, auth options
Remove deprecated localcontext
zeromq: Deprecates host, port options
Reorganize the AMQP 1.0 driver source files
Implements configurable connection factory
The need to wait for a given time is no longer valid in 3.2+
[zmq] Reduce object serialization on router proxy
Updated from global requirements
[zmq] Add backend ROUTER to increase bandwidth
[zmq] Add Sentinel instructions to deployment guide
Rabbit driver: failure of rpc-calls with float timeout

5.1.0

Use eventletutils to check is_monkey_patched
remove feature branch from master .gitreview
[zmq] Second router proxy doesn't dispatch messages properly
Add parse.unquote to transport_url
Fix simulator stat printing
Use single producer and to avoid an exchange redeclaration
[zmq] Redesign router proxy
Add feature branch to .gitreview file
Remove Beta development status from classifiers

5.0.0

Updated from global requirements
Fixes sumulator.py signal_handler logic
Refactor RPC client
Send notify if notify=True passed
Improves exception handling and logging
Implements pika thread safe connection
Fix incorrect parameters order in assertIn call
Update the RPC cast() documentation
Fix unstable work of cast func tests
[zmq] Reduce threading from python proxy
Imported Translations from Zanata
use thread safe fnmatch
Refactor base interfaces
Gracefully handle missing TCP_USER_TIMEOUT
Simulator: handle SIGINT and SIGTERM signals
Updated from global requirements
Log the unique_id in listener than msg_id
serializer: deprecate RequestContextSerializer
Don't set html_last_updated_fmt without git
Amqp driver send method temporary work-around
Updated from global requirements
Updated from global requirements
Allow simulator to be launched from arbitrary directory
[zmq] Fix cast message loss in simulator
Make transport_url config option secret
Fix oslo.messaging for Mac OS X
Refactor driver's listener interface
[kafka] Do not remove kafka_client during reset
Updated from global requirements
Replace expriration_time by timer
[zmq] Reduce number of connections
Move server related logic from dispatchers
Fix typos in Oslo.messaging files
Fix Break in Windows platforms
[py34] replace file() with open()
Claim python3 compatability for Newton onwards
Simulator: collect error stats
Simulator: make parameter wait_after_msg float
Update CheckForLoggingIssues hacking rule from keystone
Simulator: align stats to whole seconds
Support python3 in simulator.py
Fix typo passend should be passenv
Always set all socket timeouts
Add a py34 functional test for rabbit
Small fixes
Use only unique topics for the Kafka driver
[zmq] Refactoring consumer side
[Kafka] Ensure a topics before consume messages
Fix problems during unstable network
Missing version parameter in can_send_version()
Bump rabbit_transient_queues_ttl to 30 mins
Explicitly exclude tests from bandit scan
Fix Notification listener blocking behavior
Pika: fix sending fanout messages
Revert "Ensure the json result type is bytes on Python 3"
Replace deprecated LOG.warn with LOG.warning
Simulator: store results in JSON format
Simulator: calculate message latency statistics
Fix the driver shutdown/failover logic
Always delete exc_info tuple, even if reply fails
Do not leak Listeners on failover
Simulator: always use random messages for time-bound tests
Fallback if git is absent
Simulator: implement own random generator instead of scipy
Simulator: fix batch-notify-server command
Work with kombu from upstream
Fail quickly if there on bad password
[zmq] Dynamic port range is ignored
[zmq] Implement Response and Envelope classes
[kafka] Use notification priority
Make simulator more asynchronous
Adds exhange declaration on sender's side
Updated from global requirements

4.5.0

amqp: log time elapsed between receiving a message and replying
[zmq] Matchmaker redis set instead of list
Allow Notifier to have multiple topics
Fix a minor syntax error in a log statement
Use PortOpt on kafka_default_port
Added duration to notify server/client
Ensure the json result type is bytes on Python 3
Improves logging
Use more efficient mask_dict_password to mask password
Improves poller's stop logic
Typos of 'recieve' instead of 'receive'
[zmq] Support transport URL
Get kafka notifications to work with kafka-python 0.9.5
Move server's logic from executors
Avoid hardcoding the notification topic and specify driver
[zmq] Fix cinder create volume hangs
Py3: Replace filter()/map() if a list is needed
Py3: Switch json to oslo_serialization
Updated from global requirements

4.4.0

Updated from global requirements
Option rpc_response_timeout should not be used in zmq driver
Remove duplicate requirements
Reduce number of rabbitmq consumer tag used
Documents the mirror queue policy of RabbitMQ 3.0
fix override_pool_size
Remove executor callback
Log format change in simulator.py
Fix kombu accept different TTL since version 3.0.25
.testr.conf: revert workaround of testtools bug
Remove aioeventlet executor

4.3.0

simulator.py improvements
rabbit: improvements to QoS
Updated from global requirements
Remove server queue creating if target's server is empty
Updated from global requirements
Correctly set socket timeout for publishing
Updated from global requirements
Use more secure yaml.safe_load() instead of yaml.load()
[kombu] Implement experimental message compression
[zmq] Multithreading access to zmq sockets
[zmq] ZMQ_LINGER default value
Remove matchmaker_redis configs from [DEFAULT]
Refactors base classes

4.2.0

Switches pika driver to eager connection to RabbitMQ
Remove bandit.yaml in favor of defaults
[zmq] Use PUSH/PULL for direct CAST
Updated from global requirements
support ability to set thread pool size per listener
Fix misspellings
[zmq] RPC timeout for CAST
Enable pep8 on oslo_messaging/tests

4.1.0

[zmq] Fix slow down
Update translation setup
Let PikaDriver inherit base.BaseDriver
Improve simulator.py
Fixed some warnings about imports and variable
test: Don't test message's reply timeout
Updated from global requirements
Adds document and configuration guide
[zmq] Support KeyboardInterrupt for broker
[zmq] Reduce proxy for direct messaging
Fixed a couple of pep8 errors/warnings
assertEquals is deprecated, use assertEqual
Updated from global requirements
Updated from global requirements
Trivial: Remove unused logging import
replace string format arguments with function parameters
Adds params field to BlockingConnection object
Python 3 deprecated the logger.warn method in favor of warning
Fix URL in warning message
[zmq] Implement background redis polling from the client-side
rabbit: Add option to configure QoS prefetch count
rabbit: making interval_max configurable
Imported Translations from Zanata
Updated from global requirements
Logging rpc client/server targets
Updated from global requirements
Topic/server arguments changed in simulator.py
[zmq] Update zmq-guide with new options
[zmq] Listeners management cleanup
Drop H237,H402,H904 in flake8 ignore list
Replace deprecated library function os.popen() with subprocess
py3: Replaces xrange() with six.moves.range()
Kombu: make reply and fanout queues expire instead of auto-delete
fix .gitreview - bad merge from pika branch
Explicitly add pika dependencies
Add duration option to simulator.py
[zmq] Added redis sentinel HA implementation to zmq driver
rabbit: set interval max for auto retry
[zmq] Add TTL to redis records
Updated from global requirements
make enforce_type=True in CONF.set_override
Use assertTrue/False instead of assertEqual(T/F)
Improvement of logging acorrding to oslo.i18n guideline
Updated from global requirements
rabbit: fix unit conversion error of expiration
list_opts: update the notification options group
rabbit: Missing to pass parameter timeout to next
Fix formatting of code blocks in zmq docs
Adds unit tests for pika_poll module
Updated from global requirements
[zmq] Switch notifications to PUB/SUB pattern
Optimize sending of a reply in RPC server
Optimize simulator.py for better throughput
Remove stale directory synced from oslo-incubator
Fix wrong bugs report URL in CONTRIBUTING
zmq: Don't log error when can't import zmq module

4.0.0

assertIsNone(val) instead of assertEqual(None,val)
Adds tests for pika_message.py
[zmq] PUB-SUB pipeline
Updated from global requirements
Fixes conflicts after merging master
Updated from global requirements
Move to debug a too verbose log
Cleanup parameter docstrings
Removes MANIFEST.in as it is not needed explicitely by PBR
Revert "default of kombu_missing_consumer_retry_timeout"
Don't trigger error_callback for known exc
Adds comment for pika_pooler.py
Improves comment
Fix reconnection when heartbeat is missed
Revert "serializer: deprecate RequestContextSerializer"
Fix notifier options registration
notif: Check the driver features in dispatcher
batch notification listener
Updated from global requirements
Adds comment, updates pika-pool version
Preparations for configurable serialization
creates a dispatcher abstraction
Remove unnecessary quote
Fix multiline strings with missing spaces
Properly skip zmq tests without ZeroMQ being installed
kombu: remove compat of folsom reply format
Follow the plan about the single reply message

3.1.0

default of kombu_missing_consumer_retry_timeout
rename kombu_reconnect_timeout option
Skip Cyrus SASL tests if proton does not support Cyrus SASL
setUp/tearDown decorator for set/clear override
Adds comments and small fixes
Support older notifications set_override keys
Don't hold the connection when reply fail
doc: explain rpc call/cast expection
Add a driver for Apache Kafka
Option group for notifications
Move ConnectionPool and ConnectionContext outside amqp.py
Use round robin failover strategy for Kombu driver
Revert "serializer: remove deprecated RequestContextSerializer"
Updated from global requirements
[zmq] Random failure with ZmqPortRangeExceededException
[zmq] Driver optimizations for CALL
Updated from global requirements
Use oslo_config new type PortOpt for port options
serializer: remove deprecated RequestContextSerializer
Add log info for AMQP client
Updated from global requirements
Provide missing parts of error messages
Add Warning when we cannot notify
ignore .eggs directory
serializer: deprecate RequestContextSerializer
middleware: remove oslo.context usage
Removes additional select module patching
Fix delay before host reconnecting

3.0.0

Remove qpidd's driver from the tree
Provide alias to oslo_messaging.notify._impl_messaging
make pep8 faster
Updated from global requirements
Robustify locking in MessageHandlingServer
Updated from global requirements
cleanup tox.ini

2.9.0

[zmq] Add config options to specify dynamic ports range
[zmq] Make bind address configurable
[zmq][matchmaker] Distinguish targets by listener types
[zmq] Update zmq-deployment guide according to the new driver
Implements more smart retrying
Make "Connect(ing|ed) to AMQP server" log messages DEBUG level
Updated from global requirements
Decouple transport for RPC and Notification
Fixing the server example code Added server.stop() before server.wait()

2.8.1

Revert "Robustify locking in MessageHandlingServer"
Splits pika driver into several files
Fixes and improvements after testing on RabbitMQ cluster:
Move supported messaging drivers in-tree

2.8.0

Add a "bandit" target to tox.ini
Fix fanout exchange name pattern
Updated from global requirements
Remove a useless statement
Robustify locking in MessageHandlingServer
Use "secret=True" for password-related options
Imported Translations from Zanata
Modify simulator.py tool
Fix target resolution mismatch in neutron, nova, heat
Use yaml.safe_load instead of yaml.load
Trivial locking cleanup in test_listener
Remove unused event in ServerThreadHelper
Fix a race calling blocking MessageHandlingServer.start()
Fix assumptions in test_server_wait_method
Rename MessageHandlingServer._executor for readability
Implements rabbit-pika driver
bootstrap branch
Updated from global requirements

2.7.0

Updated from global requirements
Some executors are not async so update docstring to reflect that
Updated from global requirements
Updated from global requirements
Small grammar messaging fix
Use a condition (and/or a dummy one) instead of a lock
Updated from global requirements

2.6.1

•: Fix failures when zmq is not available

2.6.0

AMQP1.0: Turn off debug tracing when running tox
Fix typo in rpc/server.py and notify/listener.py
Fix a typo in server.py
Use the hostname from the Transport for GSSAPI Authentication
Adapt functional tests to pika-driver
ConfFixture should work even when zmq/redis is not present
Added matchmaker timeouts and retries
AMQP 1.0: Properly initialize AMQP 1.0 configuration options
Updated from global requirements
Workaround test stream corruption issue
Skip Redis specific tests when it is not installed
Port the AMQP 1.0 driver to Python 3
rabbit: shuffle hosts before building kombu URL
Updated from global requirements
Remove unnecessary rpc_zmq_port option
Non-blocking outgoing queue was implemented
Allow custom notification drivers
Fix the home-page value with Oslo wikipage
Include changelog/history in docs
Fix spelling typo in output
Change ignore-errors to ignore_errors
Unsubscribe target listener when leaving
Add SASL configuration options for AMQP 1.0 driver
Updated from global requirements
Fix a few leaks in the AMQP 1.0 driver
Disable ACL if authentication cannot be performed
Imported Translations from Zanata
Enhance start/stop concurrency race condition fix
Updated from global requirements
Extend logging in amqpdriver
Remove useless additional requirement file
Fix AMQP 1.0 functional and unit test failures
Provide the executor 'wait' function a timeout and use it

2.5.0

Imported Translations from Transifex
Update path to subunit2html in post_test_hook
Fix typos in a document and a comment
Updated from global requirements
Imported Translations from Transifex
Updated from global requirements
Port the AMQP1 driver to new Pyngus SASL API
Updated from global requirements
Imported Translations from Transifex
Updated from global requirements
Add config options to the documentation
Updated from global requirements

2.4.0

Mask passwords when logging messages
Updated from global requirements
Use proper translating helper for logging
Improve simulator.py

2.3.0

Imported Translations from Transifex
Added trace logging for debuggability
Log warning instead of raising RuntimeError
Use pickle instead of jsonutils for serialization
Updated from global requirements
Acknowledgements implementation
Replace 'M' with 'Mitaka'
Add if condition for random.shuffle
Fix message missing after duplicated message error
Fix fork-related issues
FIx CPU time consuming in green_poller poll()
Documenting main driver classes
Notifier implementation
Imported Translations from Transifex
Fix BaseDriver.listen_for_notifications() signature
ZMQ: Minor matchmaker improvement
Imported Translations from Transifex
Updated from global requirements
Add unit tests for zmq_async

2.2.0

Imported Translations from Transifex
ZMQ: `Lazify` driver code
Ensures that some assumptions are true
Remove oslo namespace package
Register matchmaker_redis_opts in RedisMatchMaker
Imported Translations from Transifex
Updated from global requirements
ZMQ: Removed unused code and tests
ZMQ: Run more functional tests
Get rid of proxy process in zmq
Fully use futurist code-base to abstract concurrent.futures away

2.1.0

Imported Translations from Transifex
Updated from global requirements
Close sockets properly
add plugin documentation for executors and notifiers
Allows to change defaults opts
Target direct usage
Move zmq tests into a subdirectory

2.0.0

Allow a forward slash as a part of the user/password
Update 'impl_eventlet' docstring to reflect actual impl
Updated from global requirements
tests: adjusts an expected time for gate
Updated from global requirements
Ensure callback variable capture + cleanup is done correctly
Remove oslo namespace package
ZMQ: Initial matchmaker implementation
Updated from global requirements
Fix threading zmq poller and proxy
Don't install pyngus on Python 3
Fix amqp connection pool leak in ConnectionContext
Executor docstring & attribute tweaks

1.17.1

Use the warn_eventlet_not_patched util function
Drop use of 'oslo' namespace package

1.17.0

Updated from global requirements
Add unit tests for zmq_serializer
Updated from global requirements
Fix work with timeout in CallRequest.receive_reply()
Fix mock use for mock 1.1.0
Make heartbeat the default
ZMQ: Allow to raise remote exception
Local Fanout implementation
Drop use of 'oslo' namespace package
Use oslo.log in the zmq receiver
Imported Translations from Transifex
Remove usage of contentmanager for executors
Verify that version in 'prepare' is valid

1.16.0

Fix qpid's functional gate
Don't reply when we known that client is gone
Remove py26 artefacts from oslo.messaging code
Remove 2.6 classifier
Imported Translations from Transifex
Add WebOb and greenlet to requirements
Use ServiceBase from oslo.service as a parent class
Manual update the requirements
Deprecated impl_qpid
Add a missed `raise` statement
Remove qpid-config call
Initial commit for new zmq driver implementation
Add tox target to find missing requirements
Fix qpid's functional gate
Imported Translations from Transifex
fix typo
Correct RPCVersionCapError message

1.15.0

Drop use of 'oslo' namespace package
Update .gitreview for feature/zmq
Use `inferred=True` by default
Enable amqp's protocol unit tests everywhere
Switch badges from 'pypip.in' to 'shields.io'
Don't use devstack to setup our functional env
Switch to warnings module instead of versionutils
Updated from global requirements
Get mox from mox3, not from six.moves
rabbit: Add logging on blocked connection
Provide better detection of failures during message send

1.14.0

Reduce `magic` conf attribute usage
Imported Translations from Transifex
Remove leftover oslo.config reference
replace rpc_response_timeout use in rabbit driver
Enable `fanout_target` scenarios in test_impl_rabbit
Add drivers to the documentation

1.13.0

Ensure rpc_response_timeout is registered before using it
rabbit: test for new reply behavior

1.12.0

Fix condition in _publish_and_retry_on_missing_exchange()
Set places to 0 in self.assertAlmostEqual()
Allow to remove second _send_reply() call
Don't create a new channel in RabbitMQ Connection.reset()
Imported Translations from Transifex
Adding Publisher Acknowledgements/confirms
Fix deprecated_group of rpc_conn_pool_size
Refactor processing reply in ReplyWaiter
rabbit: doc fixes
consumer connections not closed properly

1.11.0

rabbit: smart timeout on missing exchange
rabbit: Fix message ttl not work
rabbit: remove publisher classes
rabbit: Set timeout on the underlying socket
Remove stale copy of context.py
Add one more functional test for MessagingTimeout
Fix list_opts test to not check all deps
make it possible to import amqp driver without dependencies
Remove outdated release notes
rabbit: smarter declaration of the notif. queue
rabbit: redeclare consumers when ack/requeue fail
Bump kombu and amqp requirements
Updated from global requirements
rabbit: fix exception path in queue redeclaration
rabbit: fix consumers declaration
rabbit: remove unused consumer interfaces
rabbit: remove unused code
rabbit: Remove unused stuffs from publisher
Remove support for Python 3.3
Updated from global requirements
Add RequestContextSerializer
Updated from global requirements
rabbit: fixes a logging issue
rabbit/qpid: simplify the consumer loop
Updated from global requirements
Imported Translations from Transifex
Fix missing space in help text
zmq: Add support for ZmqClient pooling
Enable eventlet dependency on Python 3
Add JsonPayloadSerializer serializer
Fix test_matchmaker_redis on Python 3
Disable and mark heartbeat as experimental

1.10.0

Uncap library requirements for liberty
Port ZMQ driver to Python 3
Use unittest.mock on Python 3
Enable redis test dependency on Python 3
Remove amqp driver 'unpacked content' logging
Updated from global requirements
Add pypi download + version badges
Fix TypeError caused by err_msg formatting
Fix typo in oslo_messaging/_drivers/protocols/amqp/opts.py
Document notification_driver possible values
Do not skip functional test for amqp driver
Add functional test for notify.logger
Properly deserialize received AMQP 1.0 messages
Make notify driver messaging play well with publish_errors
Imported Translations from Transifex

1.9.0

Use the oslo_utils stop watch in decaying timer
Updated from global requirements
Remove 'UNIQUE_ID is %s' logging
Sync with latest oslo-incubator
rabbit: fix ipv6 support
Create a unique transport for each server in the functional tests
Publish tracebacks only on debug level
Add pluggability for matchmakers
Make option [DEFAULT]amqp_durable_queues work
Reconnect on connection lost in heartbeat thread
Don't raise Timeout on no-matchmaker results
Imported Translations from Transifex
cleanup connection pool return
rabbit: Improves logging
fix up verb tense in log message
rabbit: heartbeat implementation
Fix changing keys during iteration in matchmaker heartbeat
Minor improvement
ZeroMQ deployment guide
Fix a couple typos to make it easier to read
Tiny problem with notify-server in simulator
Fix coverage report generation
Add support for multiple namespaces in Targets
tools: add simulator script
Deprecates the localcontext API
Update to oslo.context
Remove obsolete cross tests script
Fix the bug redis do not delete the expired keys

1.8.0

Updated from global requirements
NotifyPublisher need handle amqp_auto_delete
Fix matchmaker_redis ack_alive fails with KeyError
Properly distinguish between server index zero and no server

1.7.0

Add FAQ entry for notifier configuration
rabbit: Fix behavior of rabbit_use_ssl
amqp1: fix functional tests deps
Skip functional tests that fail due to a qpidd bug
Use import of zmq package for test skip
Remove unnecessary log messages from amqp1 unit tests
Include missing parameter in call to listen_for_notifications
Fix the import of the driver by the unit test
Add a new aioeventlet executor
Add missing unit test for a recent commit
Add the threading executor setup.cfg entrypoint
Move each drivers options into its own group
Refactor the replies waiter code
Imported Translations from Transifex
Fix notifications broken with ZMQ driver
Gate functionnal testing improvements
Treat sphinx warnings as errors
Move gate hooks to the oslo.messaging tree
Set the password used in gate
Update README.rst format to match expectations

1.6.0

Declare DirectPublisher exchanges with passive=True
Updated from global requirements
Expose _impl_test for designate
Update Oslo imports to remove namespace package
Speedup the rabbit tests
Fix functionnal tests
kombu: fix driver loading with kombu+qpid scheme
Fixed docstring for Notifier
zmq: Refactor test case shared code
Add more private symbols to the old namespace package
Updated from global requirements
Adjust tests for the new namespace
Fixes test_two_pools_three_listener
Add TimerTestCase missing tests case
Ensure kombu channels are closed
fix qpid test issue with eventlet monkey patching
Make setup.cfg packages include oslo.messaging
Upgrade to hacking 0.10
Implements notification-dispatcher-filter
Add oslo.messaging._drivers.common for heat tests
Port zmq driver to Python 3
Make sure zmq can work with redis
fix qpid test issue with eventlet monkey patching
Move files out of the namespace package
Add a info log when a reconnection occurs
rabbit: fix timeout timer when duration is None
Don't log each received messages
Fix some comments in a backporting review session
Enable IPv6-support in libzmq by default
Add a thread + futures executor based executor
safe_log Sanitize Passwords in List of Dicts
Updated from global requirements
rabbit: add some tests when rpc_backend is set
Warns user if thread monkeypatch is not done
Add functional and unit 0mq driver tests
The executor doesn't need to set the timeout
qpid: honor iterconsume timeout
rabbit: more precise iterconsume timeout
Workflow documentation is now in infra-manual
Touch up grammar in warning messages

1.5.1

Reintroduces fake_rabbit config option
Make the RPCVersionCapError message clearer
Doc: 'wait' releases driver connection, not 'stop'
Don't allow call with fanout target
Imported Translations from Transifex
Add an optional executor callback to dispatcher

1.5.0

Rabbit: Fixes debug message format
Rabbit: iterconsume must honor timeout
Don't use oslo.cfg to set kombu in-memory driver
Don't share connection pool between driver object
Show what the threshold is being increased to
Wait for expected messages in listener pool test
Dispath messages in all listeners in a pool
Reduces the unit tests run times
Set correctly the messaging driver to use in tests
Always use a poll timeout in the executor
Have the timeout decrement inside the wait() method
Warn user if needed when the process is forked
Renamed PublishErrorsHandler
Fix reconnect race condition with RabbitMQ cluster
Create a new connection when a process fork has been detected
Add more TLS protocols to rabbit impl
Remove the use of PROTOCOL_SSLv3
Add qpid and amqp 1.0 tox targets
Updated from global requirements
Imported Translations from Transifex
rabbit: uses kombu instead of builtin stuffs
Allows to overriding oslotest environ var
Create ZeroMQ Context per socket
Remove unuseful param of the ConnectionContext
Updated from global requirements
Add basic tests for 0mq matchmakers
Notification listener pools
Updated from global requirements
Fix tiny typo in server.py
Switch to oslo.middleware
Updated from global requirements
Activate pep8 check that _ is imported
Enable user authentication in the AMQP 1.0 driver
Documentation anomaly in TransportURL parse classmethod
Don't put the message payload into warning log
Updated from global requirements
Fix incorrect attribute name in matchmaker_redis
Add pbr to installation requirements
Updated from global requirements
Add driver independent functional tests
Imported Translations from Transifex
zmq: Remove dead code
Updated from global requirements
Finish transition to oslo.i18n
Imported Translations from Transifex
Imported Translations from Transifex
qpid: Always auto-delete queue of DirectConsumer
Updated from global requirements
Imported Translations from Transifex
Enable oslo.i18n for oslo.messaging
Switch to oslo.serialization
Cleanup listener after stopping rpc server
Updated from global requirements
Track the attempted method when raising UnsupportedVersion
fix memory leak for function _safe_log
Stop using importutils from oslo-incubator
Add missing deprecated group amqp1
Updated from global requirements
Stop using intersphinx
Add documentation explaining how to use the AMQP 1.0 driver
Imported Translations from Transifex
Construct ZmqListener with correct arguments
Message was send to wrong node with use zmq as rpc_backend
Work toward Python 3.4 support and testing
Ensure the amqp options are present in config file
Add contributing page to docs
Import notifier middleware from oslo-incubator
Let oslotest manage the six.move setting for mox

1.4.1

Imported Translations from Transifex
Add square brackets for ipv6 based hosts
An initial implementation of an AMQP 1.0 based messaging driver
warn against sorting requirements
Improve help strings
Switch to oslo.utils
Fix Python 3 testing
Import oslo-incubator context module
Import oslo-incubator/middleware/base
Should not send replies for cast messages
Port to Python 3
Sync jsonutils from oslo-incubator
Add parameter to customize Qpid receiver capacity
Make tests pass with random python hashseed
Set sample_default for rpc_zmq_host
Enable PEP8 check E714
Enable PEP8 check E265
Enable PEP8 check E241
Fix error in example of an RPC server
Replace lambda method _
Enable check for E226
Updated from global requirements
Add release notes for 1.4.0.0a4
Add release notes for stable/icehouse 1.3.1 release

1.4.0.0a4

Enabled hacking checks H305 and H307
Bump hacking to version 0.9.2
Fixes incorrect exchange lock in fake driver
Imported Translations from Transifex

1.4.0.0a3

Add release notes for 1.4.0.0a2/a3
Fix AMQPListener for polling with timeout
Replaced 'e.g.' with 'for example'
Use assertEqual instead of assertIs for strings

1.4.0.0a2

Fix structure of unit tests in oslo.messaging (part 3 last)
Fix structure of unit tests in oslo.messaging (part 2)
Fix slow notification listener tests
encoding error in file
Fix info method of ListenerSetupMixin

1.4.0.0a1

Add release notes for 1.4.0.0a1
Fix formatting of TransportURL.parse() docs
Remove duplicate docs for MessageHandlingServer
Add missing docs for list_opts()
Add 'docs' tox environment
Replace usage of str() with six.text_type
Fix structure of unit tests in oslo.messaging (part 1)
Synced jsonutils and its dependencies from oslo-incubator
Ensures listener queues exist in fake driver
RPC server doc: use the blocking executor
Fix the notifier example
Removes the use of mutables as default args
Set correct group for matchmaker_redis options
replace string format arguments with function parameters
Removes contextlib.nested
Transport reconnection retries for notification
Disable connection pool in qpid interfaces tests
Updated from global requirements
Add check credentials to log message if rabbmitmq closes socket
Fix the notify method of the routing notifier
Handle unused allowed_remote_exmods in _multi_send
rabbit/qpid: remove the args/kwargs from ensure()
Add an example usage of RPCClient retry parameter
Add transport reconnection retries
Add an optional timeout parameter to Listener.poll
Bump hacking to 0.9.x series
Removes unused config option
fixed pep8 issue E265
Setup for translation
Updated from global requirements
Remove amqp default exchange hack
remove default=None for config options
Cleaned up references to executor specific RPCServer types
Make the TransportUrl hashable
debug level logs should not be translated
Explicitly name subscription queue for responses
Fix passing envelope variable as timeout
Updated from global requirements
Synced jsonutils from oslo-incubator
Remove str() from LOG.* and exceptions
Remove dependent module py3kcompat
Enable log messages to handle exceptions containing unicode
Updated from global requirements
Fix typo in docstring of notify/notifier
Full support of multiple hosts in transport url
Logical error in blockless fanout of zmq
Select AMQP message broker at random
Use a for loop to set the defaults for __call__ params
Update ensure()/reconnect() to catch MessagingError
Remove old drivers dead code
Import run_cross_tests.sh from oslo-incubator
Remove rendundant parentheses of cfg help strings
zmq: switch back to not using message envelopes
Trival:Fix assertEqual arguments order
Oslo-messaging-zmq-receiver cannot recive any messages

1.3.0

Add release notes for 1.3.0
Ensure routing key is specified in the address for a direct producer
Fix wrong parameter description in docstring
Fixed inconsistent EventletContextManagerSpawnTest failures
Use messaging_conf fixture configuration by default
Fixed possible pep8 failure due to pyflakes bug
Refactor AMQP message broker selection
Add unit test to check the order of Qpid hosts on reconnect
Fixed the issue for pop exception
Clean up for qpid tests
Add kombu driver library to requirements.txt
Use driver's notify_send() method again
Remove vim header
Updated from global requirements
Fixed spelling error - runnung to running
Build log_handler documentation
Add release notes up to 1.3.0a9

1.3.0a9

•: Remove use of sslutils

1.3.0a8

Expose PublishErrorsHandler through oslo.messaging
Use mock's call assert methods over call_args_list
notify listener: document the metadata callback parameter
Add missing data into the notif. endpoint callback
notification listener: add allow_requeue param
Adds unit test cases to impl_qpid
Do not leak _unique_id out of amqp drivers
Add multiple exchange per listerner in fake driver
Allow to requeue the notification message
Slow down Kombu reconnect attempts
Don't run python 3 tests by default
Gracefully handle consumer cancel notifications
Updated from global requirements
Convert to oslo.test
Add log_handler to oslo.messaging
Add a link to the docs from the README
Pass the session to QpidMessage constructor
User a more accurate max_delay for reconnects
Make the dispatcher responsible of the message ack
Don't reply to notification message
Abstract the acknowledge layer of a message
Implements notification listener and dispatcher
Switch over to oslosphinx
Improve help strings
Update ExpectedException handling
Ignore any egg and egg-info directories
Qpid: advance thru the list of brokers on reconnect
RabbitMQ: advance thru the list of brokers on reconnect

1.3.0a7

Make the dispatcher responsible to listen()
Allow fake driver to consume multiple topics
Allow different login methods to be used with kombu connections

1.3.0a6

Use stevedore's make_test_instance
Expose an entry point to list all config options
Fix test case name typo
Fix UnboundLocalError error

1.3.0a5

Fix help strings
Add release notes for 1.3.0a3
python3: Switch to mox3 instead of mox
Remove dependencies on pep8, pyflakes and flake8
Routing notifier

1.3.0a4

Removes use of timeutils.set_time_override
Fix spelling errors in comments
Fix test_notifier_logger for Python 3
Minor Python 3 fixes
Remove copyright from empty files
Fix duplicate topic messages for Qpid topology=2
Replace dict.iteritems() with six.iteritems()
Remove unused eventlet/greenlet from qpid/rabbit
fix test_rabbit for Python 3
Fix try/except syntax for Python 3
Fix exception deserialiation on Python 3
Add Sample priority
sysnchronize oslo-incubator modules
Remove eventlet related code in amqp driver
Fix syntax of relative imports for Python3
Updated from global requirements
Updated from global requirements
Unify different names between Python2 and Python3
Replace data structures' attribute with six module
Avoid creating qpid connection twice in initialization
Use six.moves.queue instead of Queue
Add transport aliases
Remove the partial implementation of ack_on_error
Fixed misspellings of common words
Add release notes for 1.3.0a2
Unify different names between Python2/3 with six.moves
Remove vim header
Ensure context type is handled when using to_dict
Refactors boolean returns

1.3.0a2

Simplify common base exception prototype
Properly reconnect subscribing clients when QPID broker restarts
Remove useless global vars / import
Avoid storing configuration in notifier
Implement a log handler using notifier
notifier: add audit level
Add 'warning' as an alias to 'warn'
Decouple from Oslo uuidutils module
Supply missing argument to raise_invalid_topology_version()
Support a new qpid topology
Remove hosts as property in TransportURL
Remove property on virtual_host in TransportURL
Updated from global requirements
Fix some typos and adjust capitalization
Changes driver method for notifications

1.3.0a1

Properly handle transport URL config on the client
Updated from global requirements
Updated from global requirements
Replace assertEquals with assertEqual
Properly handle transport:///vhost URL
Updated from global requirements
Make rpc_backend default to 'rabbit'
Apply six for metaclass
Add third element to RPC versions for backports
Fix rpc client docs
Updated from global requirements
Remove cruft from setup.cfg
Updated from global requirements
Fixes a typo in the address string syntax
Implement the server side of ZmqDriver
Add zmq-receiver
Implement the client side of ZmqDriver
Import zmq driver code with minimal modifications

1.2.0a11

Fix race-condition in rabbit reply processing
Fix error message if message handler fails
Don't include msg_id or reply_q in casts
Remove check_for_lock support in RPCClient

1.2.0a10

•: Add a Notifier.prepare() method

1.2.0a9

•: Fix dictionary changed size during iteration

1.2.0a8

•: Fix transport URL parsing bug

1.2.0a7

•: Fix rabbit driver handling of None, etc. replies

1.2.0a6

Remove ConfFixture from toplevel public API
Fix fake driver handling of failure replies
Bumps hacking to 0.7.0
Fix transport URL ipv6 parsing support

1.2.0a5

•: Fix handling of None, etc. replies

1.2.0a4

1.2.0a3

Add a unit testing configuration fixture
Add a TransportURL class to the public API

1.2.0a2

•: Ensure namespace package is installed

1.2.0a1

Add transport URL support to rabbit driver
Kill ability to specify exchange in transport URL
Fix capitalization, it's OpenStack
Fix handling expected exceptions in rabbit driver
Add thread-local store of request context
Add a context serialization hook
Removes a redundant version_is_compatible function
Document how call() handles remote exceptions
Add a per-transport allow_remote_exmods API
Expose RemoteError exception in the public API
Implement failure replies in the fake driver
Add API for expected endpoint exceptions
Add a driver method specifically for sending notifications
Enforce target preconditions outside of drivers
Add comments to ReplyWaiter.wait()
Remove some FIXMEs and debug logging
Remove unused IncomingMessage.done()
Implement wait_for_reply timeout in rabbit driver
Use testtools.TestCase assertion methods
Implement failure replies in rabbit driver
Add test with multiple waiting sender threads
Fix race condition in ReplyWaiters.wake_all()
Add rabbit unit test for sending and receiving replies
Add some docs on target version numbers
Add tests for rabbit driver wire protcol
Pop _unique_id when checking for duplicates
Add a transport cleanup() method
Remove my notes and test scripts
Add initial qpid driver
Move most new rabbit driver code into amqpdriver
Move rpc_conn_pool_size into amqp
Add simple rabbit driver unit test
Temporarily add eventlet to requirements
Add missing gettextutils
Add unit tests for object pool
Remove only_free param to Pool.get()
Connection pool bugfix
Remove unused file
Add exception serialization tests
Don't call consume() each time iterconsume() is called
Add test code for the rabbit driver
Remove use of gettextutils
Add initial rabbit driver
Remove use of openstack.common.local
Use stdlib logging
Don't register options with cfg.CONF at module import
Port away from some eventlet infrastructure
Adjust imports in rabbit/qpid drivers
Import some needed modules from oslo-incubator
Add oslo-incubator code unmodified
Make executor threads more robust
Allow use of hacking 0.6.0 and fix min version
Include docstrings in published docs
Use oslo.sphinx and remove local copy of doc theme
Add some notes
Unit tests for notifier
Make test notifier useful
Use lowercase priority in log notifier
Use lowercase priority in notifications topic
Handle case where no drivers configured
Fix buglet in v2 messaging notifier
Make LOG private in notifier
Require a transport to construct a Notifier
Add serializer support to notifier
Rename context to ctxt in serializer API
Rename context to ctxt in notify API
Make Notifier public at top-level
Docstrings for notifier API
Fix notify drivers namespace
Remove backwards compat entry point aliases
Simplify public symbol exports
Use assertEqual() rather than assertEquals()
Remove accidental use of messaging.rpc_server
Make exchange_from_url() use parse_url()
Unit tests for URL parsing code
Fix parse_urls() buglets
Move url utils into messaging._urls
Don't use common logging
Update example scripts for recent API changes
Fix fake driver with eventlet
Use log.warning() instead of log.warn()
Fix some pep8 issues
Don't translate exception messages
Knock off a few TODOs
Add can_send_version() to RPCClient
Check start() does nothing on a running server
Remove unused statements in base serializer
Fix thinko in exchange_from_url()
Call wait() in server tests
Add docstrings for base executor class
Remove a fixed fixme
Add a client call timeout test
Don't raise a driver-specific error on send
Add some docstrings to driver base
Test a bunch more RPC server scenarios
Make it possible to call prepare() on call context
Rework how queues get created in fake driver
Use testscenarios
Move files to new locations for oslo.messaging
Import stuff from oslo-incubator
Add oslo.messaging project infrastructure
Add some RPC server tests
More gracefully handle "no listeners" in fake driver
Better error handling in server.start()
Re-work server API to eliminate server subclasses
Add license header to _executors/__init__.py
Add RPCDispatcher tests
Check for return value in client serializer test
Add note about can_send_version()
More client unit tests
Make RPCClient.check_for_lock a callable
Apply version cap check when casting
Make RPCVersionCapError extend base exception
Remove a bogus param from client.prepare() docs
pep8 fixes for serializer code
Simple RPCClient test
Unit tests
Move some stuff into doc/
Implement Target.__eq__()
Fix bug in exchange_from_url()
pep8 fixes for fake driver
Make utils.parse_url() docstring pep8 compliant
Don't translate exceptions
Misc pep8 fixes
pep8 fixes for toplevel package
Some error handling improvements
Recommend wrapping the client class rather than subclassing
Document how to use RPCClient directly
Document the public RPC API
Fix defaults for client.prepare() args
Fix client.cast() typo
Fix version_cap typo
Allow all target attributes in client.prepare()
Expose Serializer from top-level namespace
Allow specifying a serializer when creating a server
Make endpoint.target optional
Dispatch methods in their own greenthreads
Make rpc.dispatcher private
Make the base RPCServer class private
Fix typo with the serializer work
Update use of stevedore
Require topics and target in notify driver constructors
Add generic serialization support
Support namespace in RPCClient.prepare()
Add parse_url to _utils
Remove entry point lists from the public API
Support capping message versions in the client
Fix RPCClient check_for_lock()
First cut at the notifier API
Add some notes
Add IncomingMessage abstraction
Pass a context dict
Fix docstring
Implement a fake driver
Adding reply infrastructure
Add some exceptions
Fix buglet with default timeout
Fix target/namespace target buglet
Fix rpc client buglets
Fix 'Blockinging' typos
Missing self parameter to server start()
Fix default_exchange typo
Add forgotten piece of eventlet executor
It's _executors not _executor
Make poll() just return the message
Make drivers list public again
Add top-level convenience aliases
Prefix the executors module with underscore
Prefix the messaging.server module with an underscore
Prefix the drivers module with an underscore
Make transport methods private
Fix little typo in server exception class name
Add missing utils module
Add convenience RPC server classes
Update changes.txt for recent API changes
Use : for loading classes in entry_points
Split the dispatcher from the executor and server
Make driver and transport methods public
Pass the driver instance to the listener instead of config
Try out replacing "executor" for "dispatcher"
Fix host vs server typo
Initial framework

REFERENCE

Transport

class oslo_messaging.Transport(driver)

A messaging transport.

This is a mostly opaque handle for an underlying messaging transport driver.

RPCs and Notifications may use separate messaging systems that utilize different drivers, access permissions, message delivery, etc. To ensure the correct messaging functionality, the corresponding method should be used to construct a Transport object from transport configuration gleaned from the user's configuration and, optionally, a transport URL.

The factory method for RPC Transport objects:

def get_rpc_transport(conf, url=None,


                      allowed_remote_exmods=None)

If a transport URL is supplied as a parameter, any transport configuration contained in it takes precedence. If no transport URL is supplied, but there is a transport URL supplied in the user's configuration then that URL will take the place of the URL parameter.

The factory method for Notification Transport objects:

def get_notification_transport(conf, url=None,


                               allowed_remote_exmods=None)

If no transport URL is provided, the URL in the notifications section of the config file will be used. If that URL is also absent, the same transport as specified in the user's default section will be used.

The Transport has a single 'conf' property which is the cfg.ConfigOpts instance used to construct the transport object.

class oslo_messaging.TransportURL(conf, transport=None, virtual_host=None, hosts=None, query=None): A parsed transport URL.
Transport URLs take the form:

driver://[user:pass@]host:port[,[userN:passN@]hostN:portN]/virtual_host?query

where:

driver: Specifies the transport driver to use. Typically this is rabbit for the RabbitMQ broker. See the documentation for other available transport drivers.
[user:pass@]host:port: Specifies the network location of the broker. user and pass are the optional username and password used for authentication with the broker.

user and pass may contain any of the following ASCII characters:

Alphabetic (a-z and A-Z)
Numeric (0-9)
Special characters: & = $ - _ . + ! * ( )

user may include at most one @ character for compatibility with some implementations of SASL.

All other characters in user and pass must be encoded via '%nn'

You may include multiple different network locations separated by commas. The client will connect to any of the available locations and will automatically fail over to another should the connection fail.

virtual_host: Specifies the "virtual host" within the broker. Support for virtual hosts is specific to the message bus used.
query: Permits passing driver-specific options which override the corresponding values from the configuration file.

Parameters

conf (oslo.config.cfg.ConfigOpts) -- a ConfigOpts instance
transport (str) -- a transport name for example 'rabbit'
virtual_host (str) -- a virtual host path for example '/'
hosts (list) -- a list of TransportHost objects
query (dict) -- a dictionary of URL query parameters

classmethod parse(conf, url=None): Parse a URL as defined by TransportURL and return a TransportURL object.
Assuming a URL takes the form of:

transport://user:pass@host:port[,userN:passN@hostN:portN]/virtual_host?query

then parse the URL and return a TransportURL object.

Netloc is parsed following the sequence bellow:

It is first split by ',' in order to support multiple hosts
All hosts should be specified with username/password or not at the same time. In case of lack of specification, username and password will be omitted:

user:pass@host1:port1,host2:port2
[


  {"username": "user", "password": "pass", "host": "host1:port1"},


  {"host": "host2:port2"}
]

If the url is not provided conf.transport_url is parsed instead.

Parameters

conf (oslo.config.cfg.ConfigOpts) -- a ConfigOpts instance
url (str) -- The URL to parse

Returns: A TransportURL

class oslo_messaging.TransportHost(hostname=None, port=None, username=None, password=None): A host element of a parsed transport URL.

oslo_messaging.set_transport_defaults(control_exchange): Set defaults for messaging transport configuration options.

Parameters: control_exchange (str) -- the default exchange under which topics are scoped

Forking Processes and oslo.messaging Transport objects

oslo.messaging can't ensure that forking a process that shares the same transport object is safe for the library consumer, because it relies on different 3rd party libraries that don't ensure that. In certain cases, with some drivers, it does work:

•: rabbit: works only if no connection have already been established.

Executors

Executors control how a received message is scheduled for processing by a Server. This scheduling can be synchronous or asynchronous.

A synchronous executor will process the message on the Server's thread. This means the Server can process only one message at a time. Other incoming messages will not be processed until the current message is done processing. For example, in the case of an RPCServer only one method call will be invoked at a time. A synchronous executor guarantees that messages complete processing in the order that they are received.

An asynchronous executor will process received messages concurrently. The Server thread will not be blocked by message processing and can continue to service incoming messages. There are no ordering guarantees - message processing may complete in a different order than they were received. The executor may be configured to limit the maximum number of messages that are processed at once.

Available Executors

eventlet

Executor that uses a green thread pool to execute calls asynchronously.

See: https://docs.python.org/dev/library/concurrent.futures.html and http://eventlet.net/doc/modules/greenpool.html for information on how this works.

It gathers statistics about the submissions executed for post-analysis...

threading

Executor that uses a thread pool to execute calls asynchronously.

It gathers statistics about the submissions executed for post-analysis...

See: https://docs.python.org/dev/library/concurrent.futures.html

Target

class oslo_messaging.Target(exchange=None, topic=None, namespace=None, version=None, server=None, fanout=None, legacy_namespaces=None)

Identifies the destination of messages.

A Target encapsulates all the information to identify where a message should be sent or what messages a server is listening for.

Different subsets of the information encapsulated in a Target object is relevant to various aspects of the API:

an RPC Server's target:: topic and server is required; exchange is optional
an RPC endpoint's target:: namespace and version is optional
an RPC client sending a message:: topic is required, all other attributes optional
a Notification Server's target:: topic is required, exchange is optional; all other attributes ignored
a Notifier's target:: topic is required, exchange is optional; all other attributes ignored

Its attributes are:

Parameters

exchange (str) -- A scope for topics. Leave unspecified to default to the control_exchange configuration option.
topic (str) -- A name which identifies the set of interfaces exposed by a server. Multiple servers may listen on a topic and messages will be dispatched to one of the servers selected in a best-effort round-robin fashion (unless fanout is True).
namespace (str) -- Identifies a particular RPC interface (i.e. set of methods) exposed by a server. The default interface has no namespace identifier and is referred to as the null namespace.
version (str) -- RPC interfaces have a major.minor version number associated with them. A minor number increment indicates a backwards compatible change and an incompatible change is indicated by a major number bump. Servers may implement multiple major versions and clients may require indicate that their message requires a particular minimum minor version.
server (str) -- RPC Clients can request that a message be directed to a specific server, rather than just one of a pool of servers listening on the topic.
fanout (bool) -- Clients may request that a copy of the message be delivered to all servers listening on a topic by setting fanout to True, rather than just one of them.
legacy_namespaces (list of strings) -- A server always accepts messages specified via the 'namespace' parameter, and may also accept messages defined via this parameter. This option should be used to switch namespaces safely during rolling upgrades.

Target Versions

Target version numbers take the form Major.Minor. For a given message with version X.Y, the server must be marked as able to handle messages of version A.B, where A == X and B >= Y.

The Major version number should be incremented for an almost completely new API. The Minor version number would be incremented for backwards compatible changes to an existing API. A backwards compatible change could be something like adding a new method, adding an argument to an existing method (but not requiring it), or changing the type for an existing argument (but still handling the old type as well).

If no version is specified it defaults to '1.0'.

In the case of RPC, if you wish to allow your server interfaces to evolve such that clients do not need to be updated in lockstep with the server, you should take care to implement the server changes in a backwards compatible and have the clients specify which interface version they require for each method.

Adding a new method to an endpoint is a backwards compatible change and the version attribute of the endpoint's target should be bumped from X.Y to X.Y+1. On the client side, the new RPC invocation should have a specific version specified to indicate the minimum API version that must be implemented for the method to be supported. For example:

def get_host_uptime(self, ctxt, host):


    cctxt = self.client.prepare(server=host, version='1.1')


    return cctxt.call(ctxt, 'get_host_uptime')

In this case, version '1.1' is the first version that supported the get_host_uptime() method.

Adding a new parameter to an RPC method can be made backwards compatible. The endpoint version on the server side should be bumped. The implementation of the method must not expect the parameter to be present.:

def some_remote_method(self, arg1, arg2, newarg=None):


    # The code needs to deal with newarg=None for cases


    # where an older client sends a message without it.


    pass

On the client side, the same changes should be made as in example 1. The minimum version that supports the new parameter should be specified.

RPC Server

An RPC server exposes a number of endpoints, each of which contain a set of methods which may be invoked remotely by clients over a given transport.

To create an RPC server, you supply a transport, target and a list of endpoints.

A transport can be obtained simply by calling the get_rpc_transport() method:

transport = messaging.get_rpc_transport(conf)

which will load the appropriate transport driver according to the user's messaging configuration. See get_rpc_transport() for more details.

The target supplied when creating an RPC server expresses the topic, server name and - optionally - the exchange to listen on. See Target for more details on these attributes.

Multiple RPC Servers may listen to the same topic (and exchange) simultaneously. See RPCClient for details regarding how RPC requests are distributed to the Servers in this case.

Each endpoint object may have a target attribute which may have namespace and version fields set. By default, we use the 'null namespace' and version 1.0. Incoming method calls will be dispatched to the first endpoint with the requested method, a matching namespace and a compatible version number.

The first parameter to method invocations is always the request context supplied by the client. The remaining parameters are the arguments supplied to the method by the client. Endpoint methods may return a value. If so the RPC Server will send the returned value back to the requesting client via the transport.

The executor parameter controls how incoming messages will be received and dispatched. Refer to the Executor documentation for descriptions of the types of executors.

Note: If the "eventlet" executor is used, the threading and time library need to be monkeypatched. The Eventlet executor is deprecated and the threading executor will be the only available executor.

The RPC reply operation is best-effort: the server will consider the message containing the reply successfully sent once it is accepted by the messaging transport. The server does not guarantee that the reply is processed by the RPC client. If the send fails an error will be logged and the server will continue to processing incoming RPC requests.

Parameters to the method invocation and values returned from the method are python primitive types. However the actual encoding of the data in the message may not be in primitive form (e.g. the message payload may be a dictionary encoded as an ASCII string using JSON). A serializer object is used to convert incoming encoded message data to primitive types. The serializer is also used to convert the return value from primitive types to an encoding suitable for the message payload.

RPC servers have start(), stop() and wait() methods to begin handling requests, stop handling requests, and wait for all in-process requests to complete after the Server has been stopped.

A simple example of an RPC server with multiple endpoints might be:

# NOTE(changzhi): We are using eventlet executor and
# time.sleep(1), therefore, the server code needs to be
# monkey-patched.
import eventlet
eventlet.monkey_patch()
from oslo_config import cfg
import oslo_messaging
import time
class ServerControlEndpoint(object):


    target = oslo_messaging.Target(namespace='control',


                                   version='2.0')


    def __init__(self, server):


        self.server = server


    def stop(self, ctx):


        if self.server:


            self.server.stop()
class TestEndpoint(object):


    def test(self, ctx, arg):


        return arg
transport = oslo_messaging.get_rpc_transport(cfg.CONF)
target = oslo_messaging.Target(topic='test', server='server1')
endpoints = [


    ServerControlEndpoint(None),


    TestEndpoint(),
]
server = oslo_messaging.get_rpc_server(transport, target, endpoints,


                                       executor='eventlet')
try:


    server.start()


    while True:


        time.sleep(1)
except KeyboardInterrupt:


    print("Stopping server")
server.stop()
server.wait()

oslo_messaging.get_rpc_server(transport, target, endpoints, executor=None, serializer=None, access_policy=None, server_cls=<class 'oslo_messaging.rpc.server.RPCServer'>): Construct an RPC server.

Parameters

transport (Transport) -- the messaging transport
target (Target) -- the exchange, topic and server to listen on
endpoints (list) -- a list of endpoint objects
executor (str) -- (DEPRECATED) name of message executor - available values are 'eventlet' and 'threading'. The Eventlet executor is also deprecated.
serializer (Serializer) -- an optional entity serializer
access_policy (RPCAccessPolicyBase) -- an optional access policy. Defaults to DefaultRPCAccessPolicy
server_cls (class) -- The server class to instantiate

class oslo_messaging.RPCAccessPolicyBase: Determines which endpoint methods may be invoked via RPC

class oslo_messaging.LegacyRPCAccessPolicy: The legacy access policy allows RPC access to all callable endpoint methods including private methods (methods prefixed by '_')

class oslo_messaging.DefaultRPCAccessPolicy: The default access policy prevents RPC calls to private methods (methods prefixed by '_')
NOTE:

LegacyRPCAdapterPolicy currently needs to be the default while we have projects that rely on exposing private methods.

class oslo_messaging.ExplicitRPCAccessPolicy: Policy which requires decorated endpoint methods to allow dispatch

class oslo_messaging.RPCDispatcher(endpoints, serializer, access_policy=None)

A message dispatcher which understands RPC messages.

A MessageHandlingServer is constructed by passing a callable dispatcher which is invoked with context and message dictionaries each time a message is received.

RPCDispatcher is one such dispatcher which understands the format of RPC messages. The dispatcher looks at the namespace, version and method values in the message and matches those against a list of available endpoints.

Endpoints may have a target attribute describing the namespace and version of the methods exposed by that object.

The RPCDispatcher may have an access_policy attribute which determines which of the endpoint methods are to be dispatched. The default access_policy dispatches all public methods on an endpoint object.

class oslo_messaging.MessageHandlingServer(transport, dispatcher, executor=None): Server for handling messages.
Connect a transport to a dispatcher that knows how to process the message using an executor that knows how the app wants to create new tasks.

reset(): Reset service.
Called in case service running in daemon mode receives SIGHUP.

start(override_pool_size=None)

Start handling incoming messages.

This method causes the server to begin polling the transport for incoming messages and passing them to the dispatcher. Message processing will continue until the stop() method is called.

The executor controls how the server integrates with the applications I/O handling strategy - it may choose to poll for messages in a new process, thread or co-operatively scheduled coroutine or simply by registering a callback with an event loop. Similarly, the executor may choose to dispatch messages in a new thread, coroutine or simply the current thread.

stop(): Stop handling incoming messages.
Once this method returns, no new incoming messages will be handled by the server. However, the server may still be in the process of handling some messages, and underlying driver resources associated to this server are still in use. See 'wait' for more details.

wait()

Wait for message processing to complete.

After calling stop(), there may still be some existing messages which have not been completely processed. The wait() method blocks until all message processing has completed.

Once it's finished, the underlying driver resources associated to this server are released (like closing useless network connections).

oslo_messaging.expected_exceptions(*exceptions)

Decorator for RPC endpoint methods that raise expected exceptions.

Marking an endpoint method with this decorator allows the declaration of expected exceptions that the RPC server should not consider fatal, and not log as if they were generated in a real error scenario.

Note that this will cause listed exceptions to be wrapped in an ExpectedException, which is used internally by the RPC sever. The RPC client will see the original exception type.

oslo_messaging.expose(func): Decorator for RPC endpoint methods that are exposed to the RPC client.
If the dispatcher's access_policy is set to ExplicitRPCAccessPolicy then endpoint methods need to be explicitly exposed.:

# foo() cannot be invoked by an RPC client
def foo(self):


    pass
# bar() can be invoked by an RPC client
@rpc.expose
def bar(self):


    pass

exception oslo_messaging.ExpectedException: Encapsulates an expected exception raised by an RPC endpoint
Merely instantiating this exception records the current exception information, which will be passed back to the RPC client without exceptional logging.

RPC Client

class oslo_messaging.RPCClient(transport, target, timeout=None, version_cap=None, serializer=None, retry=None, call_monitor_timeout=None, transport_options=None, _manual_load=True)

A class for invoking methods on remote RPC servers.

The RPCClient class is responsible for sending method invocations to and receiving return values from remote RPC servers via a messaging transport.

The class should always be instantiated by using the get_rpc_client function and not constructing the class directly.

Two RPC patterns are supported: RPC calls and RPC casts.

An RPC cast is used when an RPC method does not return a value to the caller. An RPC call is used when a return value is expected from the method. For further information see the cast() and call() methods.

The default target used for all subsequent calls and casts is supplied to the RPCClient constructor. The client uses the target to control how the RPC request is delivered to a server. If only the target's topic (and optionally exchange) are set, then the RPC can be serviced by any server that is listening to that topic (and exchange). If multiple servers are listening on that topic/exchange, then one server is picked using a best-effort round-robin algorithm. Alternatively, the client can set the Target's server attribute to the name of a specific server to send the RPC request to one particular server. In the case of RPC cast, the RPC request can be broadcast to all servers listening to the Target's topic/exchange by setting the Target's fanout property to True.

While the default target is set on construction, target attributes can be overridden for individual method invocations using the prepare() method.

A method invocation consists of a request context dictionary, a method name and a dictionary of arguments.

This class is intended to be used by wrapping it in another class which provides methods on the subclass to perform the remote invocation using call() or cast():

class TestClient(object):


    def __init__(self, transport):


        target = messaging.Target(topic='test', version='2.0')


        self._client = messaging.get_rpc_client(transport, target)


    def test(self, ctxt, arg):


        return self._client.call(ctxt, 'test', arg=arg)

An example of using the prepare() method to override some attributes of the default target:

def test(self, ctxt, arg):


    cctxt = self._client.prepare(version='2.5')


    return cctxt.call(ctxt, 'test', arg=arg)

RPCClient have a number of other properties - for example, timeout and version_cap - which may make sense to override for some method invocations, so they too can be passed to prepare():

def test(self, ctxt, arg):


    cctxt = self._client.prepare(timeout=10)


    return cctxt.call(ctxt, 'test', arg=arg)

However, this class can be used directly without wrapping it another class. For example:

transport = messaging.get_rpc_transport(cfg.CONF)
target = messaging.Target(topic='test', version='2.0')
client = messaging.get_rpc_client(transport, target)
client.call(ctxt, 'test', arg=arg)

but this is probably only useful in limited circumstances as a wrapper class will usually help to make the code much more obvious.

If the connection to the messaging service is not active when an RPC request is made the client will block waiting for the connection to complete. If the connection fails to complete, the client will try to re-establish that connection. By default this will continue indefinitely until the connection completes. However, the retry parameter can be used to have the RPC request fail with a MessageDeliveryFailure after the given number of retries. For example:

client = messaging.get_rpc_client(transport, target, retry=None)
client.call(ctxt, 'sync')
try:


    client.prepare(retry=0).cast(ctxt, 'ping')
except messaging.MessageDeliveryFailure:


    LOG.error("Failed to send ping message")

call(ctxt, method, **kwargs)

Invoke a method and wait for a reply.

The call() method is used to invoke RPC methods that return a value. Since only a single return value is permitted it is not possible to call() to a fanout target.

call() will block the calling thread until the messaging transport provides the return value, a timeout occurs, or a non-recoverable error occurs.

call() guarantees that the RPC request is done 'at-most-once' which ensures that the call will never be duplicated. However if the call should fail or time out before the return value arrives then there are no guarantees whether or not the method was invoked.

Since call() blocks until completion of the RPC method, call()s from the same thread are guaranteed to be processed in-order.

Method arguments must either be primitive types or types supported by the client's serializer (if any). Similarly, the request context must be a dict unless the client's serializer supports serializing another type.

The semantics of how any errors raised by the remote RPC endpoint method are handled are quite subtle.

Firstly, if the remote exception is contained in one of the modules listed in the allow_remote_exmods messaging.get_rpc_transport() parameter, then it this exception will be re-raised by call(). However, such locally re-raised remote exceptions are distinguishable from the same exception type raised locally because re-raised remote exceptions are modified such that their class name ends with the '_Remote' suffix so you may do:

if ex.__class__.__name__.endswith('_Remote'):


    # Some special case for locally re-raised remote exceptions

Secondly, if a remote exception is not from a module listed in the allowed_remote_exmods list, then a messaging.RemoteError exception is raised with all details of the remote exception.

Parameters

ctxt (dict) -- a request context dict
method (str) -- the method name
kwargs (dict) -- a dict of method arguments

Raises: MessagingTimeout, RemoteError, MessageDeliveryFailure

can_send_version(version=<object object>): Check to see if a version is compatible with the version cap.

cast(ctxt, method, **kwargs)

Invoke a method without blocking for a return value.

The cast() method is used to invoke an RPC method that does not return a value. cast() RPC requests may be broadcast to all Servers listening on a given topic by setting the fanout Target property to True.

The cast() operation is best-effort: cast() will block the calling thread until the RPC request method is accepted by the messaging transport, but cast() does not verify that the RPC method has been invoked by the server. cast() does guarantee that the method will be not executed twice on a destination (e.g. 'at-most-once' execution).

There are no ordering guarantees across successive casts, even among casts to the same destination. Therefore methods may be executed in an order different from the order in which they are cast.

Method arguments must either be primitive types or types supported by the client's serializer (if any).

Similarly, the request context must be a dict unless the client's serializer supports serializing another type.

Parameters

ctxt (dict) -- a request context dict
method (str) -- the method name
kwargs (dict) -- a dict of method arguments

Raises: MessageDeliveryFailure if the messaging transport fails to accept the request.

prepare(exchange=<object object>, topic=<object object>, namespace=<object object>, version=<object object>, server=<object object>, fanout=<object object>, timeout=<object object>, version_cap=<object object>, retry=<object object>, call_monitor_timeout=<object object>, transport_options=<object object>): Prepare a method invocation context.
Use this method to override client properties for an individual method invocation. For example:

def test(self, ctxt, arg):


    cctxt = self.prepare(version='2.5')


    return cctxt.call(ctxt, 'test', arg=arg)

Parameters

exchange (str) -- see Target.exchange
topic (str) -- see Target.topic
namespace (str) -- see Target.namespace
version (str) -- requirement the server must support, see Target.version
server (str) -- send to a specific server, see Target.server
fanout (bool) -- send to all servers on topic, see Target.fanout
timeout (int or float) -- an optional default timeout (in seconds) for call()s
version_cap (str) -- raise a RPCVersionCapError version exceeds this cap
retry (int) -- an optional connection retries configuration: None or -1 means to retry forever. 0 means no retry is attempted. N means attempt at most N retries.
transport_options (dictionary) -- additional parameters to configure the driver for example to send parameters as "mandatory" flag in RabbitMQ
call_monitor_timeout (int) -- an optional timeout (in seconds) for active call heartbeating. If specified, requires the server to heartbeat long-running calls at this interval (less than the overall timeout parameter).

exception oslo_messaging.RemoteError(exc_type=None, value=None, traceback=None): Signifies that a remote endpoint method has raised an exception.
Contains a string representation of the type of the original exception, the value of the original exception, and the traceback. These are sent to the parent as a joined string so printing the exception contains all of the relevant info.

Notifier

class oslo_messaging.Notifier(transport, publisher_id=None, driver=None, serializer=None, retry=None, topics=None)

Send notification messages.

The Notifier class is used for sending notification messages over a messaging transport or other means.

Notification messages follow the following format:

{'message_id': str(uuid.uuid4()),


 'publisher_id': 'compute.host1',


 'timestamp': timeutils.utcnow(),


 'priority': 'WARN',


 'event_type': 'compute.create_instance',


 'payload': {'instance_id': 12, ... }}

A Notifier object can be instantiated with a transport object and a publisher ID:

notifier = messaging.Notifier(get_notification_transport(CONF),


                              'compute')

and notifications are sent via drivers chosen with the driver config option and on the topics chosen with the topics config option in [oslo_messaging_notifications] section.

Alternatively, a Notifier object can be instantiated with a specific driver or topic:

transport = notifier.get_notification_transport(CONF)
notifier = notifier.Notifier(transport,


                             'compute.host',


                             driver='messaging',


                             topics=['notifications'])

Notifier objects are relatively expensive to instantiate (mostly the cost of loading notification drivers), so it is possible to specialize a given Notifier object with a different publisher id using the prepare() method:

notifier = notifier.prepare(publisher_id='compute')
notifier.info(ctxt, event_type, payload)

audit(ctxt, event_type, payload): Send a notification at audit level.

Parameters

ctxt (dict) -- a request context dict
event_type (str) -- describes the event, for example 'compute.create_instance'
payload (dict) -- the notification payload

Raises: MessageDeliveryFailure

critical(ctxt, event_type, payload): Send a notification at critical level.

Parameters

ctxt (dict) -- a request context dict
event_type (str) -- describes the event, for example 'compute.create_instance'
payload (dict) -- the notification payload

Raises: MessageDeliveryFailure

debug(ctxt, event_type, payload): Send a notification at debug level.

Parameters

ctxt (dict) -- a request context dict
event_type (str) -- describes the event, for example 'compute.create_instance'
payload (dict) -- the notification payload

Raises: MessageDeliveryFailure

error(ctxt, event_type, payload): Send a notification at error level.

Parameters

ctxt (dict) -- a request context dict
event_type (str) -- describes the event, for example 'compute.create_instance'
payload (dict) -- the notification payload

Raises: MessageDeliveryFailure

info(ctxt, event_type, payload): Send a notification at info level.

Parameters

ctxt (dict) -- a request context dict
event_type (str) -- describes the event, for example 'compute.create_instance'
payload (dict) -- the notification payload

Raises: MessageDeliveryFailure

is_enabled(): Check if the notifier will emit notifications anywhere.

Returns: false if the driver of the notifier is set only to noop, true otherwise

prepare(publisher_id=<object object>, retry=<object object>): Return a specialized Notifier instance.
Returns a new Notifier instance with the supplied publisher_id. Allows sending notifications from multiple publisher_ids without the overhead of notification driver loading.

Parameters

publisher_id (str) -- field in notifications sent, for example 'compute.host1'
retry (int) -- connection retries configuration (used by the messaging driver): None or -1 means to retry forever. 0 means no retry is attempted. N means attempt at most N retries.

sample(ctxt, event_type, payload)

Send a notification at sample level.

Sample notifications are for high-frequency events that typically contain small payloads. eg: "CPU = 70%"

Not all drivers support the sample level (log, for example) so these could be dropped.

Parameters

ctxt (dict) -- a request context dict
event_type (str) -- describes the event, for example 'compute.create_instance'
payload (dict) -- the notification payload

Raises: MessageDeliveryFailure

warn(ctxt, event_type, payload): Send a notification at warning level.

Parameters

ctxt (dict) -- a request context dict
event_type (str) -- describes the event, for example 'compute.create_instance'
payload (dict) -- the notification payload

Raises: MessageDeliveryFailure

warning(ctxt, event_type, payload): Send a notification at warning level.

Parameters

ctxt (dict) -- a request context dict
event_type (str) -- describes the event, for example 'compute.create_instance'
payload (dict) -- the notification payload

Raises: MessageDeliveryFailure

class oslo_messaging.LoggingNotificationHandler(url, publisher_id=None, driver=None, topic=None, serializer=None)

Handler for logging to the messaging notification system.

Each time the application logs a message using the logging module, it will be sent as a notification. The severity used for the notification will be the same as the one used for the log record.

This can be used into a Python logging configuration this way:

[handler_notifier]
class=oslo_messaging.LoggingNotificationHandler
level=ERROR
args=('rabbit:///')

CONF = <oslo_config.cfg.ConfigOpts object>: Default configuration object used, subclass this class if you want to use another one.

emit(record): Emit the log record to the messaging notification system.

Parameters: record -- A log record to emit.

class oslo_messaging.LoggingErrorNotificationHandler(*args, **kwargs)

emit(record): Do whatever it takes to actually log the specified logging record.
This version is intended to be implemented by subclasses and so raises a NotImplementedError.

Available Notifier Drivers

log

Publish notifications via Python logging infrastructure.

messaging

Send notifications using the 1.0 message format.

This driver sends notifications over the configured messaging transport, but without any message envelope (also known as message format 1.0).

This driver should only be used in cases where there are existing consumers deployed which do not support the 2.0 message format.

messagingv2

Send notifications using the 2.0 message format.

noop

Base driver for Notifications

routing

Base driver for Notifications

test

Store notifications in memory for test verification.

Notification Driver

Notification drivers for sending notifications via messaging.

The messaging drivers publish notification messages to notification listeners.

In case of the rabbit backend the driver will block the notifier's thread until the notification message has been passed to the messaging transport. There is no guarantee that the notification message will be consumed by a notification listener.

In case of the kafka backend the driver will not block the notifier's thread but return immediately. The driver will try to deliver the message in the background.

Notification messages are sent 'at-most-once' - ensuring that they are not duplicated.

If the connection to the messaging service is not active when a notification is sent the rabbit backend will block waiting for the connection to complete. If the connection fails to complete, the driver will try to re-establish that connection. By default this will continue indefinitely until the connection completes. However, the retry parameter can be used to have the notification send fail. In this case an error is logged and the notifier's thread is resumed without any error.

If the connection to the messaging service is not active when a notification is sent the kafka backend will return immediately and the backend tries to establish the connection and deliver the messages in the background.

class oslo_messaging.notify.messaging.MessagingDriver(conf, topics, transport, version=1.0)

Send notifications using the 1.0 message format.

This driver sends notifications over the configured messaging transport, but without any message envelope (also known as message format 1.0).

This driver should only be used in cases where there are existing consumers deployed which do not support the 2.0 message format.

class oslo_messaging.notify.messaging.MessagingV2Driver(conf, **kwargs): Send notifications using the 2.0 message format.

class oslo_messaging.notify.notifier.Driver(conf, topics, transport): Base driver for Notifications

abstract notify(ctxt, msg, priority, retry): send a single notification with a specific priority

Parameters

ctxt -- current request context
msg (str) -- message to be sent
priority (str) -- priority of the message
retry (int) -- connection retries configuration (used by the messaging driver): None or -1 means to retry forever. 0 means no retry is attempted. N means attempt at most N retries.

Notification Listener

A notification listener is used to process notification messages sent by a notifier that uses the messaging driver.

A notification listener subscribes to the topic - and optionally exchange - in the supplied target. Notification messages sent by notifier clients to the target's topic/exchange are received by the listener.

A notification listener exposes a number of endpoints, each of which contain a set of methods. Each method's name corresponds to a notification's priority. When a notification is received it is dispatched to the method named like the notification's priority - e.g. info notifications are dispatched to the info() method, etc.

Optionally a notification endpoint can define a NotificationFilter. Notification messages that do not match the filter's rules will not be passed to the endpoint's methods.

Parameters to endpoint methods are: the request context supplied by the client, the publisher_id of the notification message, the event_type, the payload and metadata. The metadata parameter is a mapping containing a unique message_id and a timestamp.

An endpoint method can explicitly return oslo_messaging.NotificationResult.HANDLED to acknowledge a message or oslo_messaging.NotificationResult.REQUEUE to requeue the message. Note that not all transport drivers implement support for requeueing. In order to use this feature, applications should assert that the feature is available by passing allow_requeue=True to get_notification_listener(). If the driver does not support requeueing, it will raise NotImplementedError at this point.

The message is acknowledged only if all endpoints either return oslo_messaging.NotificationResult.HANDLED or None.

NOTE: If multiple listeners subscribe to the same target, the notification will be received by only one of the listeners. The receiving listener is selected from the group using a best-effort round-robin algorithm.

This delivery pattern can be altered somewhat by specifying a pool name for the listener. Listeners with the same pool name behave like a subgroup within the group of listeners subscribed to the same topic/exchange. Each subgroup of listeners will receive a copy of the notification to be consumed by one member of the subgroup. Therefore, multiple copies of the notification will be delivered - one to the group of listeners that have no pool name (if they exist), and one to each subgroup of listeners that share the same pool name.

NOTE WELL: This means that the Notifier always publishes notifications to a non-pooled Listener as well as the pooled Listeners. Therefore any application that uses listener pools must have at least one listener that consumes from the non-pooled queue (i.e. one or more listeners that do not set the pool parameter.

Note that not all transport drivers have implemented support for listener pools. Those drivers that do not support pools will raise a NotImplementedError if a pool name is specified to get_notification_listener().

Each notification listener is associated with an executor which controls how incoming notification messages will be received and dispatched. Refer to the Executor documentation for descriptions of the other types of executors.

Note: If the "eventlet" executor is used, the threading and time library need to be monkeypatched.

Notification listener have start(), stop() and wait() messages to begin handling requests, stop handling requests, and wait for all in-process requests to complete after the listener has been stopped.

To create a notification listener, you supply a transport, list of targets and a list of endpoints.

A transport can be obtained simply by calling the get_notification_transport() method:

transport = messaging.get_notification_transport(conf)

which will load the appropriate transport driver according to the user's messaging configuration. See get_notification_transport() for more details.

A simple example of a notification listener with multiple endpoints might be:

from oslo_config import cfg
import oslo_messaging
class NotificationEndpoint(object):


    filter_rule = oslo_messaging.NotificationFilter(


        publisher_id='^compute.*')


    def warn(self, ctxt, publisher_id, event_type, payload, metadata):


        do_something(payload)
class ErrorEndpoint(object):


    filter_rule = oslo_messaging.NotificationFilter(


        event_type='^instance\..*\.start$',


        context={'ctxt_key': 'regexp'})


    def error(self, ctxt, publisher_id, event_type, payload, metadata):


        do_something(payload)
transport = oslo_messaging.get_notification_transport(cfg.CONF)
targets = [


    oslo_messaging.Target(topic='notifications'),


    oslo_messaging.Target(topic='notifications_bis')
]
endpoints = [


    NotificationEndpoint(),


    ErrorEndpoint(),
]
pool = "listener-workers"
server = oslo_messaging.get_notification_listener(transport, targets,


                                                  endpoints, pool=pool)
server.start()
server.wait()

By supplying a serializer object, a listener can deserialize a request context and arguments from primitive types.

oslo_messaging.get_notification_listener(transport, targets, endpoints, executor=None, serializer=None, allow_requeue=False, pool=None)

Construct a notification listener

The executor parameter controls how incoming messages will be received and dispatched.

If the eventlet executor is used, the threading and time library need to be monkeypatched.

Parameters

transport (Transport) -- the messaging transport
targets (list of Target) -- the exchanges and topics to listen on
endpoints (list) -- a list of endpoint objects
executor (str) -- name of message executor - available values are 'eventlet' and 'threading'
serializer (Serializer) -- an optional entity serializer
allow_requeue (bool) -- whether NotificationResult.REQUEUE support is needed
pool (str) -- the pool name

Raises: NotImplementedError

oslo_messaging.get_batch_notification_listener(transport, targets, endpoints, executor=None, serializer=None, allow_requeue=False, pool=None, batch_size=None, batch_timeout=None)

Construct a batch notification listener

The executor parameter controls how incoming messages will be received and dispatched.

If the eventlet executor is used, the threading and time library need to be monkeypatched.

Parameters

transport (Transport) -- the messaging transport
targets (list of Target) -- the exchanges and topics to listen on
endpoints (list) -- a list of endpoint objects
executor (str) -- name of message executor - available values are 'eventlet' and 'threading'
serializer (Serializer) -- an optional entity serializer
allow_requeue (bool) -- whether NotificationResult.REQUEUE support is needed
pool (str) -- the pool name
batch_size (int) -- number of messages to wait before calling endpoints callacks
batch_timeout (int) -- number of seconds to wait before calling endpoints callacks

Raises: NotImplementedError

Serializer

class oslo_messaging.Serializer: Generic (de-)serialization definition base class.

abstract deserialize_context(ctxt): Deserialize a dictionary into a request context.

Parameters: ctxt -- Request context dictionary
Returns: Deserialized form of entity

abstract deserialize_entity(ctxt, entity): Deserialize something from primitive form.

Parameters

ctxt -- Request context, in deserialized form
entity -- Primitive to be deserialized

Returns: Deserialized form of entity

abstract serialize_context(ctxt): Serialize a request context into a dictionary.

Parameters: ctxt -- Request context
Returns: Serialized form of context

abstract serialize_entity(ctxt, entity): Serialize something to primitive form.

Parameters

ctxt -- Request context, in deserialized form
entity -- Entity to be serialized

Returns: Serialized form of entity

class oslo_messaging.NoOpSerializer: A serializer that does nothing.

Exceptions

exception oslo_messaging.ClientSendError(target, ex): Raised if we failed to send a message to a target.

exception oslo_messaging.DriverLoadFailure(driver, ex): Raised if a transport driver can't be loaded.

exception oslo_messaging.ExecutorLoadFailure(executor, ex): Raised if an executor can't be loaded.

exception oslo_messaging.InvalidTransportURL(url, msg): Raised if transport URL is invalid.

exception oslo_messaging.MessagingException: Base class for exceptions.

exception oslo_messaging.MessagingTimeout: Raised if message sending times out.

exception oslo_messaging.MessagingServerError: Base class for all MessageHandlingServer exceptions.

exception oslo_messaging.NoSuchMethod(method): Raised if there is no endpoint which exposes the requested method.

exception oslo_messaging.RPCDispatcherError: A base class for all RPC dispatcher exceptions.

exception oslo_messaging.RPCVersionCapError(version, version_cap)

exception oslo_messaging.ServerListenError(target, ex): Raised if we failed to listen on a target.

exception oslo_messaging.UnsupportedVersion(version, method=None): Raised if there is no endpoint which supports the requested version.

RELEASE NOTES

Read also the oslo.messaging Release Notes.

INDICES AND TABLES

Index
Module Index
Search Page

AUTHOR

Author name not set

COPYRIGHT

2025, Oslo Contributors

April 2, 2025

16.1.0