RabbitMQ Stream Python Client
A Python asyncio-based client for RabbitMQ Streams
Install
The client is distributed via PIP:
pip install rstream
Publishing messages:
You can publish messages with four different methods:
send: asynchronous, messages are automatically buffered internally and sent at once after a timeout expires.send_batch: synchronous, the user buffers the messages and sends them. This is the fastest publishing method.send_wait: synchronous, the caller wait till the message is confirmed. This is the slowest publishing method.send_sub_entry: asynchronous, allow batch in sub-entry mode. This mode increases throughput at the cost of increased latency and potential duplicated messages even when deduplication is enabled. It also allows using compression to reduce bandwidth and storage if messages are reasonably similar, at the cost of increasing CPU usage on the client side.
- producer using send
- producer using send_wait
- producer using send_batch
- producer using sub_entry_batch
Publishing with confirmation
The Send method takes as parameter an handle function that will be called asynchronously when the message sent will be notified from the server to have been published.
Example:
With send_wait instead will wait until the confirmation from the server is received.
Consuming messages:
See consumer examples for basic consumer and consumers with different offsets.
Server-side offset tracking
RabbitMQ Streams provides server-side offset tracking for consumers. This features allows a consuming application to restart consuming where it left off in a previous run. You can use the store_offset (to store an offset in the server) and query_offset (to query it) methods of the consumer class like in this example:
Superstreams
A super stream is a logical stream made of individual, regular streams. It is a way to scale out publishing and consuming with RabbitMQ Streams: a large logical stream is divided into partition streams, splitting up the storage and the traffic on several cluster nodes.
See the blog post for more info.
You can use superstream_producer and superstream_consumer classes which internally uses producers and consumers to operate on the componsing streams.
See the Super Stream example
Single Active Consumer support:
Single active consumer provides exclusive consumption and consumption continuity on a stream.
See the blog post for more info.
See examples in:
See the single active consumer example
Connecting with SSL:
import ssl
ssl_context = ssl.SSLContext()
ssl_context.load_cert_chain('/path/to/certificate.pem', '/path/to/key.pem')
producer = Producer(
host='localhost',
port=5551,
ssl_context=ssl_context,
username='guest',
password='guest',
)
Load Balancer
In order to handle load balancers, you can use the load_balancer_mode parameter for producers and consumers. This will always attempt to create a connection via the load balancer, discarding connections that are inappropriate for the client type.
Producers must connect to the leader node, while consumers can connect to any, prioritizing replicas if available.
TODO
- Documentation
- Handle
MetadataUpdateand reconnect to another broker on stream configuration changes - AsyncIterator protocol for consumer
- Add frame size validation
2 maintainers
Metadata
2 maintainers
Release files for rstream 0.10.1
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| rstream-0.10.1.tar.gz | 22.6 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| rstream-0.10.1-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 50.4 kB
Release files / rstream-0.10.1.tar.gz
| Download URL | rstream-0.10.1.tar.gz |
|---|---|
| Size | 22.6 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
69c5905621244158e68521171c6dc54253dcf6d203385aa0b9f970b326ebdefa
|
|
BLAKE2b-256 checksum How to use checksums |
3c7a938ca683e98c4517d6f5d6bf80f0be5aef531b46fab717e4ca1b1ef5573d
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
poetry/1.5.1 CPython/3.10.6 Linux/5.15.0-1041-azure
|
Release files / rstream-0.10.1-py3-none-any.whl
| Download URL | rstream-0.10.1-py3-none-any.whl |
|---|---|
| Size | 27.8 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
20e561ed40758ef28c8837f359323dd0a02e1b4b7cde02dcd39faed6f5158846
|
|
BLAKE2b-256 checksum How to use checksums |
e6b407cebc0656c23d4aba9adbe7c472b0ba75ae42be2076d241e36eab15c642
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
poetry/1.5.1 CPython/3.10.6 Linux/5.15.0-1041-azure
|
