RabbitMQ Stream Python Client

A Python asyncio-based client for RabbitMQ Streams This is a work in progress

Install

pip install rstream

Quick start

Publishing messages:

You can publish messages with three different methods:

send: asynchronous, messages are automatically buffered internally and sent at once after a timeout expires.
batch_send: 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.

Example Using send:

import asyncio
from rstream import Producer, AMQPMessage

async def publish():
    async with Producer('localhost', username='guest', password='guest') as producer:
        await producer.create_stream('mystream')

        for i in range(100):
            amqp_message = AMQPMessage(
                body='hello: {}'.format(i),
            )
            await producer.send('mystream', amqp_message)

asyncio.run(publish())

send is not thread safe so it must be awaited.

Similarly with the send_wait:

import asyncio
from rstream import Producer, AMQPMessage

async def publish():
    async with Producer('localhost', username='guest', password='guest') as producer:
        await producer.create_stream('mystream')

        for i in range(100):
            amqp_message = AMQPMessage(
                body='hello: {}'.format(i),
            )
            await producer.send_wait('mystream', amqp_message)

asyncio.run(publish())

Eventually using batch_send:

import asyncio
from rstream import Producer, AMQPMessage

async def publish():
    async with Producer('localhost', username='guest', password='guest') as producer:
        await producer.create_stream('mystream')
        list_messages = []

        for i in range(100):
            amqp_message = AMQPMessage(
                body='hello: {}'.format(i),
            )
            list_messages.append(amqp_message)

        await producer.send_batch('mystream',  list_messages) 

asyncio.run(publish())

Consuming messages:

import asyncio
import signal
from rstream import Consumer, amqp_decoder, AMQPMessage

async def consume():
    consumer = Consumer(
        host='localhost',
        port=5552,
        vhost='/',
        username='guest',
        password='guest',
    )

    loop = asyncio.get_event_loop()
    loop.add_signal_handler(signal.SIGINT, lambda: asyncio.create_task(consumer.close()))

    def on_message(msg: AMQPMessage):
        print('Got message: {}'.format(msg.body))

    await consumer.start()
    await consumer.subscribe('mystream', on_message, decoder=amqp_decoder)
    await consumer.run()

asyncio.run(consume())

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 MetadataUpdate and reconnect to another broker on stream configuration changes
AsyncIterator protocol for consumer
Add frame size validation

Metadata

Release files for rstream 0.5.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for rstream 0.5.0
File	Size	Uploaded
rstream-0.5.0.tar.gz	15.8 kB	Mar 8, 2023	Details

Built distribution (wheel)

Table of built distributions (wheels) for rstream 0.5.0
File	Interpreter	ABI	Platform	Reset
rstream-0.5.0-py3-none-any.whl 19.1 kB Mar 8, 2023	Python 3	none	any	Details

Total release size: 34.9 kB

Release files / rstream-0.5.0.tar.gz

Download URL	rstream-0.5.0.tar.gz
Size	15.8 kB
Tags	Source
SHA-256 checksum How to use checksums	`caf6c34b84f9d954153af2012084140f5f307c358b1d1d76edaab339c14cfda5`
BLAKE2b-256 checksum How to use checksums	`4e68984c2ded6584487c7b007d19e454832bffa5e2a9e0729dcab2ad6291a2cc`
Upload date	Mar 8, 2023
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`poetry/1.4.0 CPython/3.10.6 Linux/5.15.0-1033-azure`

Release files / rstream-0.5.0-py3-none-any.whl

Download URL	rstream-0.5.0-py3-none-any.whl
Size	19.1 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`bf485160fd384d5b1fcd40421175abae2c96f87e6792d689aa9d745f5094cf68`
BLAKE2b-256 checksum How to use checksums	`a47a827b3ac968adbc190c67c8e38e0ee4ff22f24ed23c6f2d7dfc6d618bfb18`
Upload date	Mar 8, 2023
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`poetry/1.4.0 CPython/3.10.6 Linux/5.15.0-1033-azure`