FastStream STOMP broker
How To Use
Install the package:
uv add faststream-stomp
poetry add faststream-stomp
Basic usage:
import asyncio
import faststream
import faststream_stomp
import stompman
server = stompman.ConnectionParameters(host="127.0.0.1", port=61616, login="admin", passcode="password")
broker = faststream_stomp.StompBroker(stompman.Client([server]))
@broker.subscriber("first")
@broker.publisher("second")
def _(message: str) -> str:
print(message) # this will print message from startup
return "Hi from first handler!"
@broker.subscriber("second")
def _(message: str) -> None:
print(message) # this will print message from first handler
app = faststream.FastStream(broker)
@app.after_startup
async def send_first_message() -> None:
await broker.connect()
await broker.publish("Hi from startup!", "first")
if __name__ == "__main__":
asyncio.run(app.run())
Also there are StompRouter and TestStompBroker for testing. It works similarly to built-in brokers from FastStream, I recommend to read the original FastStream documentation.
By default, published frames include the STOMP content-length header. You can change this for the whole broker or
override it for a publisher or individual publish call:
broker = faststream_stomp.StompBroker(stompman.Client([server]), add_content_length=False)
publisher = broker.publisher("events", add_content_length=True)
await broker.publish("text message", "events", add_content_length=False)
await publisher.publish("bytes message")
Use reply_add_content_length on subscriber() when automatic replies need a different setting. The same options
are available for batch publishing and delayed StompRoute/StompRoutePublisher registrations. Per-call settings
override publisher or subscriber reply defaults, which in turn override the broker default.
Confirming publication
By default publish() returns once the frame has been written to the socket, which does not mean the broker
accepted the message. Pass receipt_timeout to wait for the broker's STOMP RECEIPT instead. It follows the same precedence as
add_content_length: per-call, then publisher, then broker default.
broker = faststream_stomp.StompBroker(stompman.Client([server]), receipt_timeout=3.0)
publisher = broker.publisher("events", receipt_timeout=5.0)
await broker.publish("text message", "events", receipt_timeout=1.0)
await publisher.publish("another message")
The default is None everywhere, which preserves the existing behavior. A failed confirmation raises
stompman.SendReceiptError from publish(), with reason equal to rejected, timeout, or connection_lost.
timeout and connection_lost are ambiguous outcomes: the broker may have accepted the message even though its
receipt was lost. Nothing is replayed automatically: give each application event a stable ID and deduplicate in the
broker or downstream consumer if you retry. JMSCorrelationID alone does not deduplicate anything. And a receipt only
means the broker took ownership: it does not mean a consumer has processed the message.
receipt_timeout is not available for batch publishing, which publishes inside a transaction: a receipt for a
transactional SEND says nothing about whether the COMMIT succeeded. A broker or publisher default is ignored there
rather than silently claiming confirmation.
Caveats
- When exception is raised in consumer handler, the message will be nacked (FastStream doesn't do this by default)
2 maintainers
Metadata
2 maintainers
Release files for faststream-stomp 2.7.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| faststream_stomp-2.7.0.tar.gz | 10.5 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| faststream_stomp-2.7.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 26.9 kB
Release files / faststream_stomp-2.7.0.tar.gz
| Download URL | faststream_stomp-2.7.0.tar.gz |
|---|---|
| Size | 10.5 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
1632e275b6a0aa36fffa3ca50689bdedd12d2058b2db1a01657106f360ce17cd
|
|
BLAKE2b-256 checksum How to use checksums |
2b82ac1adfd353af4d5ff1d766e3eacfcc93f04d05773427e961fae1f60c50f5
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
uv/0.12.15 {"installer":{"name":"uv","version":"0.12.15","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
Release files / faststream_stomp-2.7.0-py3-none-any.whl
| Download URL | faststream_stomp-2.7.0-py3-none-any.whl |
|---|---|
| Size | 16.4 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
b55f8f3a4b02c3a135ed56a56e9b6ab55e4ba608afb29f674ece2d5ea5c3c6e4
|
|
BLAKE2b-256 checksum How to use checksums |
4fedb2cc14c66390a7ff898efa96b6dde8928c20deb7a43a81bc827a8fdea896
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
uv/0.12.15 {"installer":{"name":"uv","version":"0.12.15","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
