Getting Faststream Context in asyncio script

[RoryMMMM]

I'm trying to use asyncio to listen to a remote server, receive data constantly, and then to do some basic parsing and publishing to rabbitmq.

Here's some dummy code:

import asyncio  

from faststream import Context, FastStream, Logger, apply_types
from faststream.rabbit import RabbitBroker, RabbitExchange, RabbitQueue, ExchangeType

exch = RabbitExchange('my_exchange', auto_delete=False, type=ExchangeType.TOPIC)
broker_object = RabbitBroker(host = 'localhost', port = '5432', login='admin', password='password')  
routing_key='some.routing.key'
app = FastStream(broker_object)

@broker_object.publisher("test-queue")
async def process_chunk(data, broker=Context() ): 
    msg_dict = some_function(data.decode) # returns a dict that is built from incoming socket data
    await broker.publish(msg_dict, 
                        routing_key=routing_key, 
                        exchange=exch
                        ) 
# ===========================================================================     
class ClientReader(asyncio.Protocol):
    '''
    https://docs.python.org/3/library/asyncio-protocol.html#streaming-protocols
    '''
    def __init__(self, on_con_lost): 
        self.on_con_lost = on_con_lost

    def connection_made(self, transport):
        print('Connection made.')

    def data_received(self, data): 
        print('Data received...')
        asyncio.create_task(process_chunk(data=data))

    def connection_lost(self, exc):
        print('The server closed the connection')
        self.on_con_lost.set_result(True)

async def main(logger: Logger): 
    print("Setting up AIS asyncio...")
    loop = asyncio.get_running_loop()
    on_con_lost = loop.create_future() 
    transport, protocol = await loop.create_connection(lambda: ClientReader(on_con_lost), '172.0.0.1', 1000)
    try:
        await on_con_lost
    finally:
        transport.close()
# ===========================================================================
asyncio.run(main(Logger))
# ===========================================================================

I keep seeing that the recommendation is to use Context and decorators to pass the broker object into functions/classes that are outside the scope of the faststream code but I'm having a hard time implementing this. When I run this code (well, the code that actually has all the config and functions) I get:

Setting up AIS asyncio...
Connection made. 
Data received...
<big chunk of data>

 Task exception was never retrieved
 future: <Task finished name='Task-2' coro=<process_chunk() done, defined at /project/./ais_i_mov/application.py:43> exception=AttributeError("'Context' object has no attribute 'publish'")>
Traceback (most recent call last):
 File "/project/./ais_i_mov/application.py", line 61, in process_chunk a await broker.publish(msg_dict,
AttributeError: 'Context' object has no attribute 'publish'

Am I misunderstanding something from the context documentation? What am I actually doing when I write broker=Context() in the def?

This is my first stab at asynchronous code so I might be missing something basic here...

[Lancetnik]

1. @broker_object.publisher("test-queue") should be @broker_object.subscriber("test-queue")
2. I don't see where you start the broker_object
3. instead of using broker.publish at the and of function you should use @broker.publisher(...) decorator

The following script works fine

import asyncio

from faststream import Context, FastStream
from faststream.rabbit import ExchangeType, RabbitBroker, RabbitExchange

exch = RabbitExchange("my_exchange", auto_delete=False, type=ExchangeType.TOPIC)
broker_object = RabbitBroker()
routing_key = "some.routing.key"
app = FastStream(broker_object)

@broker_object.subscriber("test-queue")
async def process_chunk(data, broker=Context()):
    msg_dict = data
    await broker.publish(msg_dict, routing_key=routing_key, exchange=exch)


async def main():
    loop = asyncio.get_running_loop()
    on_con_lost = loop.create_future()
    await broker_object.start()
    await broker_object.publish("hi", "test-queue")
    try:
        await on_con_lost
    finally:
        await broker_object.close()


asyncio.run(main())

But the following one is preffered

@broker_object.subscriber("test-queue")
@broker_object.publisher(routing_key=routing_key, exchange=exch)
async def process_chunk(data):
    return data

[RoryMMMM]

Okay, I've got a couple more questions and to be a little pedantic:

• I see in all the examples the publisher functions are decorated with the @broker_object.subscriber("test-queue"): why? Surely this is a little against the idea of the topic exchange in rabbitmq where you shouldn't need any subscriber to exist in order to publish a message.
• Nowhere in the tutorials is the command broker_object.start() ever mentioned.
• The code is running without error, and the function that sends to rabbit is called, BUT nothing is arriving in rabbitmq...

Here's the complete, running code example. You should be able to get the messages from the public AIS_HOST server. I'm using a docker image of rabbitmq:3.8.2-management from dockerhub. Also asyncio 3.4.3, faststream 0.4.6

I test this in a Jupyter notebook so the last line is a little different:

import asyncio 

from faststream import Context, FastStream, Logger, apply_types
from faststream.rabbit import RabbitBroker, RabbitExchange, RabbitQueue, ExchangeType

# Source of RAW AIS messages
AIS_HOST="153.44.253.27"
AIS_PORT=5631

# RabbitMQ connection details
RABBIT_MSG_PORT=5672
RABBIT_MANAGE_PORT=15672
RABBIT_HOST="10.0.0.10"
RABBITMQ_DEFAULT_USER="test"
RABBITMQ_DEFAULT_PASS="test" 

# Routing Config
EXCHANGE="ais_test_exchange"
ENCODED_ROUTING_KEY="ais_test_key"

exch = RabbitExchange(EXCHANGE, durable=True, auto_delete=False, type=ExchangeType.TOPIC)
broker_object = RabbitBroker(host = RABBIT_HOST, port = RABBIT_MSG_PORT, login=RABBITMQ_DEFAULT_USER, password=RABBITMQ_DEFAULT_PASS)  
app = FastStream(broker_object)

@broker_object.subscriber("ais_test_key",exchange=exch) 
@broker_object.publisher(routing_key=ENCODED_ROUTING_KEY, exchange=exch)
async def send_to_rabbit(msg_dict): 
    print(msg_dict)
    return msg_dict

async def wait_for_data():
    loop = asyncio.get_running_loop()
    # Register the open socket to wait for data. 
    reader, writer = await asyncio.open_connection(AIS_HOST, AIS_PORT) 
    await broker_object.start()
    await broker_object.publish("hi", "test-queue")
    # Wait for data
    while True:
        data = await reader.readline()
        if not data:
            # Connection closed.
            break
        msg = data.decode()
        # print("Received:", data.decode())
        await send_to_rabbit(msg)
    writer.close()
    await writer.wait_closed()

await wait_for_data()
# asyncio.run(main())

[Lancetnik]

1. @broker.publisher(...) is just a structural way to write data-processing pipelines: you consume message from one subject and publish response to another one. It is just 1 of 4 ways to publish messages in FastStream and it is a most suitable way for your case. You can find all other publish options in the docs https://faststream.airt.ai/latest/getting-started/publishing/
2. broker.start is a method to connect & start all brokers' subscribers. You can find examples in integration doc section (your case)
3. Seems like reader.readline in your case blocks event-loop and it can't process any other tasks in the same time

[RoryMMMM]

Aaah! That's a bit of a bummer. Readline is so handy for not dealing with partial messages but I guess it was too easy. I'll have to figure something else out then...

Thanks for the assistance!