A Python library for interfacing with a message broker. It should be compatible with anything based on ActiveMQ like AmazonMQ or Red Hat AMQ, but I test it only with Red Hat AMQ.

The mb library is a thin wrapper around stomp.py. It covers the underlying protocol and exposes only classes for common use cases like:

• publish/subscribe
• push/pull
• RPC

They can be either synchronous or asynchronous (callback based).

The configuration is based on environment variable because I want to deploy the library into OpenShift (Kubernetes) and I don't want to deal with configuration in the code. It also hides a lot of configuration options and exposes only one way to interface with the messaging infrastructure.

Simple pub/sub use case:

import time
import mb

topic_name = "test"
# Note that no configuration is required because it is done in the
# OpenShift template therefore it would be redundant to do it here
consumer = mb.MbConsumer(mb.MbChannelType.TOPIC, topic_name)
producer = mb.MbProducer(mb.MbChannelType.TOPIC, topic_name)
time.sleep(1.0)
msg = "hello, world!"
producer.publish(msg)
# This is a blocking consumer, you can consume messages in a loop
received_msg = consumer.next_message()
assert received_msg == msg

Find more examples here.

Run:

$ docker-compose build && docker-compose up

This will spin up example collector and message producer together with a broker. From this simulation you can get a gist of how to deploy clients using the library. See the docker-compose.yml file for configuration. This would be configured in an OpenShift template.

To try it locally, run the broker first:

$ docker run -p 61613:61613 -e AMQ_USER=user -e AMQ_PASSWORD=redhat -e AMQ_PROTOCOL=stomp -e ADMIN_PASSWORD=redhat registry.access.redhat.com/amq-broker-7/amq-broker-72-openshift

Then run the tests themselves:

$ pytest tests