Rabbitmq with no-blocking rpc

[PhoenixNazarov]

Hello, in documentation says, that if rpc=True faststream send blocking RPC request
Example for code:

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),
            send(3, broker),
            send(4, broker),
            send(5, broker),
            send(6, broker),
            send(7, broker),
            send(8, broker),
            send(9, broker),
            send(10, broker),
        )


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

asyncio.run(t())

In queue will always be one task.

Queue have not consumers. The following tasks, where rpc=True, are published due to rpc_timeout

How can I use rpc=True with no blocking? Or should I process the result myself with a result queue

[Lancetnik]

@PhoenixNazarov according RMQ logic, we can't send multiple RPC requests at the same time using the same connection-channel pair. This reason we have client-level lock on sending RPC requests (we can't send next RPC requests while we have no answer on previous one). I thought, it shouldn't be a great problem cuz RPC logic is not targeted behavior for async services. If you want to get answers as a regular behavior, probably, you should take a look at another architecture (GRPC-based or HTTP at least) or use persistent consumer for processing replies with reply_to flag like in the following example

@broker.subscriber("reply")
async def process_reply():
    ...

await broker.publish(..., reply_to="reply")

Using this logic you can connect request with response by correlation_id and such logic is much native for RMQ than RPC.

RPC feature exists to send such request from time to time, but not to create RPC-based services.