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

Metadata

Release files for rstream 0.3.1

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.3.1
File	Size	Uploaded
rstream-0.3.1.tar.gz	14.5 kB	Oct 22, 2021	Details

Built distribution (wheel)

Table of built distributions (wheels) for rstream 0.3.1
File	Interpreter	ABI	Platform	Reset
rstream-0.3.1-py3-none-any.whl 17.4 kB Oct 22, 2021	Python 3	none	any	Details

Total release size: 31.9 kB

Release files / rstream-0.3.1.tar.gz

Download URL	rstream-0.3.1.tar.gz
Size	14.5 kB
Tags	Source
SHA-256 checksum How to use checksums	`8dbb631632de88889cfcdbf4161bb3ca0c4c66d1584cd31d3ad02a508e1ce74e`
BLAKE2b-256 checksum How to use checksums	`2c2e69c398fea6279f109684f3da4e4460d896a63d3ae05a080687f8a9accefc`
Upload date	Oct 22, 2021
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`poetry/1.1.11 CPython/3.9.5 Linux/5.11.0-37-generic`

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

Download URL	rstream-0.3.1-py3-none-any.whl
Size	17.4 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`3543356977f23903a8012c92eab825e7cbaff2a4c46e0fb7a521e6a9480b83ec`
BLAKE2b-256 checksum How to use checksums	`8ae9964279244bede6d9b5bde5d92349d101909078e0e5ca1806477e37bce1e5`
Upload date	Oct 22, 2021
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`poetry/1.1.11 CPython/3.9.5 Linux/5.11.0-37-generic`