Python STOMP client with pleasant API
Project description
stompman
A Python client for STOMP asynchronous messaging protocol that is:
- asynchronous,
- not abandoned,
- has typed, modern, comprehensible API.
How To Use
Before you start using stompman, make sure you have it installed:
pip install stompman
poetry add stompman
uv add stompman
Initialize a client:
async with stompman.Client(
servers=[
stompman.ConnectionParameters(host="171.0.0.1", port=61616, login="user1", passcode="passcode1"),
stompman.ConnectionParameters(host="172.0.0.1", port=61616, login="user2", passcode="passcode2"),
],
# Handlers:
on_error_frame=lambda error_frame: print(error_frame.body),
on_unhandled_message_frame=lambda message_frame: print(message_frame.body),
on_heartbeat=lambda: print("Server sent a heartbeat"),
# Optional parameters with sensible defaults:
heartbeat=stompman.Heartbeat(will_send_interval_ms=1000, want_to_receive_interval_ms=1000),
connect_retry_attempts=3,
connect_retry_interval=1,
connect_timeout=2,
connection_confirmation_timeout=2,
read_timeout=2,
) as client:
...
Sending Messages
To send a message, use the following code:
await client.send(b"hi there!", destination="DLQ", headers={"persistent": "true"})
Or, to send messages in a transaction:
async with client.begin() as transaction:
for _ in range(10):
await transaction.send(body=b"hi there!", destination="DLQ", headers={"persistent": "true"})
await asyncio.sleep(0.1)
Listening for Messages
Now, let's subscribe to a destination and listen for messages:
async def handle_message_from_dlq(message_frame: stompman.MessageFrame) -> None:
print(message_frame.body)
await client.subscribe("DLQ", handle_message_from_dlq, on_suppressed_exception=print)
Entered stompman.Client
will block forever waiting for messages if there are any active subscriptions.
Sometimes it's useful to avoid that:
dlq_subscription = await client.subscribe("DLQ", handle_message_from_dlq, on_suppressed_exception=print)
await dlq_subscription.unsubscribe()
By default, subscription have ACK mode "client-individual". If handler successfully processes the message, an ACK
frame will be sent. If handler raises an exception, a NACK
frame will be sent. You can catch (and log) exceptions using on_suppressed_exception
parameter:
await client.subscribe(
"DLQ",
handle_message_from_dlq,
on_suppressed_exception=lambda exception, message_frame: print(exception, message_frame),
)
You can change the ack mode used by specifying the ack
parameter:
# Server will assume that all messages sent to the subscription before the ACK'ed message are received and processed:
await client.subscribe("DLQ", handle_message_from_dlq, ack="client", on_suppressed_exception=print)
# Server will assume that messages are received as soon as it send them to client:
await client.subscribe("DLQ", handle_message_from_dlq, ack="auto", on_suppressed_exception=print)
Cleaning Up
stompman takes care of cleaning up resources automatically. When you leave the context of async context managers stompman.Client()
, or client.begin()
, the necessary frames will be sent to the server.
Handling Connectivity Issues
-
If multiple servers were provided, stompman will attempt to connect to each one simultaneously and will use the first that succeeds. If all servers fail to connect, an
stompman.FailedAllConnectAttemptsError
will be raised. In normal situation it doesn't need to be handled: tune retry and timeout parameters instompman.Client()
to your needs. -
When connection is lost, stompman will handle it automatically.
stompman.FailedAllConnectAttemptsError
will be raised if all connection attempts fail.stompman.RepeatedConnectionLostError
will be raised if connection succeeds but operation (like sending a frame) leads to connection getting lost.
...and caveats
- stompman supports Python 3.11 and newer.
- It implements STOMP 1.2 — the latest version of the protocol.
- Heartbeats are required, and sent automatically in background (defaults to 1 second).
Also, I want to pointed out that:
- Protocol parsing is inspired by aiostomp (meaning: consumed by me and refactored from).
- stompman is tested and used with Artemis ActiveMQ.
- Specification says that headers in CONNECT and CONNECTED frames shouldn't be escaped for backwards compatibility. stompman escapes headers in CONNECT frame (outcoming), but does not unescape headers in CONNECTED (outcoming).
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.