GitLocker: The Coding Marketplace

Description:

rabbitmqclient 2.4.0

RabbitMQ client helpers based on pika

This project provides helper classes for using RabbitMQ in Python. It is
based on pika, which is an awesome no-dependency client library for
RabbitMQ. Similarly, this project strives for zero dependencies (except for
dev dependencies).
By using this project, users should be able to get started with RabbitMQ in
Python instantly, by simply instantiating and starting a RMQConsumer or
RMQProducer class.
Consumer
RMQConsumer extends the RMQConnection base class with only one extra
method: consume. Consume can be passed parameters for declaring queues and
exchanges, as well as binding them together, and consume parameters, all of
which have corresponding kwargs in the pika library. The idea is not to
re-invent the wheel, but simply the process of declaring a queue -> declaring an
exchange -> binding the exchange and queue together -> consuming from the queue.
Here is an example:
from rabbitmq_client import RMQConsumer, ConsumeParams, QueueParams

def on_message(msg, ack=None):
...

consumer = RMQConsumer()
consumer.start()
consumer.consume(ConsumeParams(on_message),
queue_params=QueueParams("queue_name"))

The flow of declaring, binding, and consuming is quite
straightforward. The above example will declare a queue with the name
"queue_name" and consume from it.
NOTE! Although the above may look synchronous, it is not. Start is
asynchronous
and any consume started while the consumer is not fully started will simply be
delayed until it is. When a consume has been successfully started, the bound
callback will receive a ConsumeOK object containing the resulting
consumer tag.
Acknowledging received messages
By default, received messages need to be acknowledged when received. By calling
the ack kwarg function a message is acknowledged and won't be sent again.
If a message isn't acknowledged using this function, it will be re-sent by
RabbitMQ on consuming from a queue again.
from rabbitmq_client import RMQConsumer, ConsumeParams, QueueParams

from some_other_module import handle_msg

def on_message(msg, ack=None):
error = handle_msg(msg)

if not error:
ack()

consumer = RMQConsumer()
consumer.start()
consumer.consume(ConsumeParams(on_message),
queue_params=QueueParams("queue_name"))

To enable automatic acknowledgement of messages, pass the auto_ack parameter
to the ConsumeParams, set to True. If auto_ack is True, the
ack kwarg is unset.
Producer
RMQProducer extends the RMQConnection base class with two additional
methods: publish and activate_confirm_mode. Publish is used, as it
sounds, to publish messages towards queues and/or exchanges. The confirm
mode activation method enabled confirm mode so that users can verify that
messages have been delivered successfully.
from rabbitmq_client import RMQProducer, ExchangeParams

def on_confirm(confirmation):
...

producer = RMQProducer()
producer.start()
producer.activate_confirm_mode(on_confirm) # Or don't, depends on your needs

producer.publish(b"body",
exchange_params=ExchangeParams("exchange_name"),
routing_key="some.routing.key")

activate_confirm_mode isn't synchronous either, but you don't have to
worry about that. Calling publish after activate_confirm_mode will lead
to the publish not happening until confirm mode has been activated
successfully. The callback passed to activate_confirm_mode will also
receive a ConfirmModeOK once confirm mode is on. Any publish between
calling activate_confirm_mode and the producer receiving a
confirm_select_ok from RabbitMQ will be buffered and not issues until
confirm mode is on. When confirm mode is on, publish also returns a key that
clients can use to correlate successful delivered with calls to publish.
Once a publish call with key X is confirmed, the callback passed to
activate_confirm_mode will be called with X.
Abstract connection helper
The abstract RMQConnection class can be subclassed to get a head start in
using the pika SelectConnection and Channel objects as it wraps them
and provides an easy-to-use interface as well as event hooks on important
happenings.
RMQConnection lifecycle hooks
Subclassing RMQConnection requires the implementer to override three methods:
on_ready, on_close, and on_error.
on_ready is called when RMQConnection has established a connection and
opened a channel.
on_close is called when either the connection, or the channel closes for
any reason. This means the implementer may receive two calls for one failed
connection, one for the channel and one for the connection itself. This
makes it important that on_close is made idempotent.
on_error is called when a recent action failed, such as an exchange
declaration failure. These hooks are meant to enable the implementer to react
to the connection state and restore an operating state. The RMQConnection
abstract base class is used by the rabbitmq_client project to implement its
RMQConsumer and RMQProducer classes.
RMQConnection interface methods
In addition to the hooks that need to be implemented by implementing classes,
RMQConnection provides three public methods that can be used to interact
with the connection object: start, restart, and stop.
start initiates the connection, establishing a pika.SelectConnection and
if that's successful, opening a pika.Channel for the opened connection. Once
a channel has been opened, RMQConnection will issue a call to on_ready.
Subsequent calls to start have no effect if the connection has already been
started.
restart closes the open connection and ensures that it is started again once
is has been fully closed. restart is only meant to be used on successfully
established connections, it will have no effect on closed connections.
restart is meant to be used as a means to change pika.ConnectionParameters
on the fly.
stop permanently closes an open connection and will have no effect on a
closed connection. A connection for which stop has been called cannot be
re-used. on_close is called once the connection is completely stopped.
Aside from the connection-related methods, the RMQConnection also exposes
interations with the pika.Channel, named similarily. See here for what is
exposed: Pika docs.
Automatic reconnection
RMQConnection will re-establish lost connections, but not lost channels.
Reconnections will not be done for any reason though, among the reasons for
reconnecting are:

pika.exceptions.ConnectionClosedByBroker
pika.exceptions.StreamLostError

These two exceptions cover the cases where the broker has been shut down, either
expectedly or unexpectedly, or when the connection is lost for some other
reason.
Again, if the channel is lost, but the connection remains intact,
RMQConnection will not recover the channel.
Reconnection attempts will be made with an increasing delay between attempts.
The first attempt is instantaneous, the second is delayed by 1 second, the
third by 2 seconds, etc. After the 9th attempt, the following reconnects
will be made at 30 second intervals.
Logging
rabbitmq_client follows Python logging standards and is by default disabled.
To enable logging, attach a handler to rabbitmq_client:
import logging

logging.getLogger("rabbitmq_client").addHandler(logging.StreamHandler())

By default, a logging.NullHandler() is attached to this logger.