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:

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.publish('mystream', amqp_message)

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',
)

TODO

Documentation
Handle MetadataUpdate and reconnect to another broker on stream configuration changes
AsyncIterator protocol for consumer
Add frame size validation

Release files for rstream 0.4.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.4.0
File	Size	Uploaded
rstream-0.4.0.tar.gz	14.5 kB	Mar 21, 2022	Details

Built distribution (wheel)

Table of built distributions (wheels) for rstream 0.4.0
File	Interpreter	ABI	Platform	Reset
rstream-0.4.0-py3-none-any.whl 17.4 kB Mar 21, 2022	Python 3	none	any	Details

Total release size: 31.9 kB

Release files / rstream-0.4.0.tar.gz

Download URL	rstream-0.4.0.tar.gz
Size	14.5 kB
Tags	Source
SHA-256 checksum How to use checksums	`25c1cfda1e995d2943ae3b0c8eb9d48e880ba1d678430e118362a8d763292c30`
BLAKE2b-256 checksum How to use checksums	`13a86273fe6ab7777d19e217fa6e62b1c3884c26c051a545002ac180bced2261`
Upload date	Mar 21, 2022
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`poetry/1.1.12 CPython/3.9.5 Linux/5.11.0-49-generic`

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

Download URL	rstream-0.4.0-py3-none-any.whl
Size	17.4 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`edb35895108819ea8459bf59b7115471872281eecfd98603f0a200879abccaad`
BLAKE2b-256 checksum How to use checksums	`dcfc3e5ceea4e36bbcd40cadde3dc6450fa5c6e156fa97f12f3468ca203c4451`
Upload date	Mar 21, 2022
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`poetry/1.1.12 CPython/3.9.5 Linux/5.11.0-49-generic`