Parallel RPC with asyncio.gather

[PhoenixNazarov]

I'm using RabbitMq. And I start the server with one queue, and then I return “123” for all tasks.

When I call

async def p():
  out = await broker.publish(message, queue, exchange,
                                         expiration=60,
                                         rpc_timeout=100,
                                         rpc=True)
  print(out)
  
async def main():
  await asyncio.gather(*[p1(), p1()])

I only receive an answer to one created task, the second one hangs endlessly and does not receive its answer.
I'm expecting two "123" console outputs but I'm only getting only one.
Is this normal behavior?

[Lancetnik]

@PhoenixNazarov, sorry, I can't reproduse your problem

My RPC publish example is working correctly

import asyncio

from faststream import FastStream, Logger
from faststream.rabbit import RabbitBroker

broker = RabbitBroker()
app = FastStream(broker)

@broker.subscriber("test-q")
async def handler(msg, logger: Logger):
    logger.info(msg)
    return msg

@app.after_startup
async def t():
    result = await asyncio.gather(
        broker.publish(1, "test-q", rpc=True),
        broker.publish(2, "test-q", rpc=True),
    )
    assert result == [1, 2]

[PhoenixNazarov]

@Lancetnik, Sorry for not giving more information.

I have start server:

import uvicorn
from fastapi import FastAPI
from faststream.rabbit import RabbitQueue, RabbitExchange
from faststream.rabbit.fastapi import RabbitRouter

broker = RabbitRouter()

exchange = RabbitExchange('ai')
queue = RabbitQueue('test')


@broker.subscriber(queue, exchange)
async def wait(message):
    return message


app = FastAPI(title='test', lifespan=broker.lifespan_context)

if __name__ == '__main__':
    uvicorn.run(app, port=8001)

And my client:

import asyncio
from faststream.rabbit import RabbitBroker, RabbitExchange, RabbitQueue

broker = RabbitBroker()

exchange = RabbitExchange('ai')
queue = RabbitQueue('test')

async def t():
    result = await asyncio.gather(
        send(1),
        send(2),
    )


async def send(num):
    async with broker as _broker:
        out = await broker.publish(num, queue=queue, exchange=exchange, rpc=True)
        print(out)

asyncio.run(t())

Out in console

2
None

or

1
None

[Lancetnik]

@PhoenixNazarov please, do not try to connect broker concurantly: it can leads to race condition.

The following code should works fine

import asyncio
from faststream.rabbit import RabbitBroker, RabbitExchange, RabbitQueue

exchange = RabbitExchange('ai')
queue = RabbitQueue('test')

async def t():
    async with RabbitBroker() as broker:
        result = await asyncio.gather(
            send(1, broker),
            send(2, broker),
        )


async def send(num, broker):
    out = await broker.publish(num, queue=queue, exchange=exchange, rpc=True)
    print(out)

asyncio.run(t())

[PhoenixNazarov]

Thx, @Lancetnik will there be a speed problem if I make connections all the time?
I use fastapi. It would be more efficient to call the broker from app.lifespan?

[Lancetnik]

The preffered way - use broker from Context

from faststream.rabbit.fastapi import RabbitBroker

@router.subscriber("tesst")
async def handler(msg, broker: RabbitBroker):
   ...

[PhoenixNazarov]

Yes, but I need to be able to publish tasks outside of this context

[Lancetnik]

You can get it everywhere right by context.get("broker") or just access to router.broker