Как использовать faststream+RMQ с отправкой и получением ответа к N+1 микросервисам, в рамках одного connection

[Spirit412]

Есть.
Микросервисы. Сейчас их чуть больше десяти.
Шина на RMQ
API на асинхронном FastApi (принимает запросы из вне, агрегирует ответы от мекросервисов и отдаёт ответ за запрос). Минимум 3 воркера!
Все общаются через шину асинхронно.

Задача, которую не получается реализовать. Точнее, не понятно с какого бока к ней подойти (
На API приходит куча запросов на разные роутеры. Для обработки каждого запроса задействовано N+1 микросервиса.
Т.е. пришел запрос, нужно отправить сообщение нескольким микросервисам. Полученный ответ от них, и эти ответы должны быть именно на этот запрос... Ответы агрегируются и отправляются ответом из апи.

Единственное что приходит на ум, это на каждый запрос к API создавать одноразовые очереди(каналы) с correlation_id....
Можно ли это реализовать в faststream?
Очень сомневаюсь, что данная задача свалилась только на меня и никто раньше её не решал.

[Lancetnik]

FastStream уже поддерживает RPC запросы поверх RMQ без создания одноразовых очередей (за счёт Direct Reply-To). Просто используйте этот механизм.

[Spirit412]

Т е. Каждый запрос в роутер будет из шины получать от микросервисов только свои ответы, и все асинхронно!
Можно пример такого взаимодействия, если учесть что в роутере происходит агрегация ответов от нескольких микросервисов?

[Lancetnik]

Никакой магии:

@app.get("/")
async def handler():
    rmq_response = await broker.publish("message", "queue", "exchange", rpc=True)
    return rmq_response

И потребитель

broker = RabbitBroker()

@broker.subscriber("queue", "exchange")
async def handler(msg):
    return "Response"

Тип exchnage может быть любой, а поднимать вы можете любое количество инстансов - ответ все равно будет приходить именно на то сообщение, которое было отправлено

[Spirit412]

Спасибо .
Буду пробовать.
Библиотека действительно нужная.