learn/rabbitmq/12-流式日志(Stream)/stream_consumer.py

# -*- coding: utf-8 -*-
import asyncio
from rstream import Consumer, OffsetType

STREAM = 'demo-stream'
USER   = 'user'
PWD    = 'Un2yzriWm7veSDoh'
HOST   = 'rabbitmq.rabbitmq.svc.cluster.local'

async def on_message(msg):
    print(" [x]", msg.data.decode())

async def main():
    consumer = Consumer(
        host=HOST,
        port=5552,
        username=USER,
        password=PWD
    )
    async with consumer:
        await consumer.subscribe(
            STREAM,
            callback=on_message,
            offset_type=OffsetType.FIRST
        )
        await consumer.run()

if __name__ == '__main__':
    asyncio.run(main())