Use rabbitmq direct reply-to for RPC calls

[bodograumann]

The use case of RPC over MQ is very interesting, especially for testing purposes.
Looking at the code, it seems a dedicated reply queue is created for each request.
I think that we can avoid that by using an existing reply queue or (at least for RabbitMQ) by using the Direct Reply-To feature.

Implementing it should be simple.
broker.publish already accepts a reply_to argument and it is currently applied to the message.
Now if both callback=True and reply_to are set, we could avoid creating a dedicated reply queue and use the one from reply_to.
In particular broker.publish(queue=…, callback=True, reply_to="amq.rabbitmq.reply-to") would be using the Direct Reply-To feature of RabbitMQ.

At the moment, when usingcallback and reply_to at the same time, the latter would override the former, but propan would still be waiting for a reply on the anonymous reply queue it created internally. So the reply can never be received in this situation.

I also noticed that the callback_queue is currently not deleted. This should probably be added as well, when it is created internally. (When given by the user via reply_to it should be kept of course. amq.rabbitmq.reply-to isn't even a real queue and cannot be deleted.)

[Lancetnik]

@bodograumann thanks for notice that callback overrides original reply_to. Obviously, it should be fixed.

it seems a dedicated reply queue is created for each request.

It is exactly how RabbitMQ implements Direct Reply-To feature.
Take a look at python or any other language example.
In all cases anonymous exclusive queue is created.

I also noticed that the callback_queue is currently not deleted.

Queues, declared as exclusive automatically deleting after disconnect. So, if you connect only to send a message, it will be deleted imidiatelly. Another case, it will be deleted at reconnection (every 30 minutes with default RabbitMQ settings).
I will test fake-queue behavior tomorrow, but for now I don't see a problem. I am just following the RabbitMQ RPC tutorial, linked by you.

[bodograumann]

But, you are right: management plugin show that anonymous queue as existed. It was deleted only with channel disconnected. But, for now, I have no ideas how to implement pseudo-queue from the original article. When I was tried to get message by reply_to name direclty without declaring RabbitMQ raised error QueueNotExists

Sorry I didn't see that answer. Hope what I wrote above helps figuring out how to do Direct Reply-To without declaring a completely new queue. :-)

[Lancetnik]

I read this article too. Looks great, but it seems like I can publish message with some uuid as reply_to and RabbitMQ creates pseudo-queue binded to default exchange with the same name. But it doesn't.
I tried a few scenarios:

1. publish message with random reply_to, get the pseudo-queue by name - error "Queue not found"
2. publish message with random reply_to, try to get message from default exchange by queue name - error "Queue not found"
3. declare anonymous queue, publish message with reply_to to it - real queue was declared

Looks like there is no interfaces to implement Direct Reply-To or I can't find it.

[bodograumann]

Have you tried using the exact string "amq.rabbitmq.reply-to" as reply_to when publishing?

[Lancetnik]

Wait a second... Looks like I got it. There are too few materials on the Internet to deal with it so simply.

[Lancetnik]

But I don't undestand how get exactly my response from "amq.rabbitmq.reply-to" yet

[Lancetnik]

Wait: you can't use callback=True and reply_to both the same time. Callback should be used then you need to get response right here. reply_to should be used then you need to get response in long-living queue with handler, declared on your app.
So, it looks lika candidate to ValueError, not a wrong behavior.

[Lancetnik]

Also, please do not use RPC to testing. It's a production feature.
I was working hard on local testing, so, please, use it with pytest locally and in CI.

[bodograumann]

Yes definitely for unit testing and such I will use the testnig facilities which won't require all the dependencies.
However migrating to Propan requires changes and iteration on how the service interacts with RabbitMQ.
That is why I want to have a minimal RPC over MQ client that checks whether the service replies correctly with the real RabbitMQ server.

[Lancetnik]

Sure, thanks for your trust in this young project. We can make it greater together with my job and your production feedback!

[Lancetnik]

@bodograumann thanks for your help. RPC over MQ become better by you.

[bodograumann]

Thank you so much for implementing this. Just got server and client working with propan.
First I was a bit confused, because I did not see any reply queue in rabbitmq,
but then I saw that we are now using the Direct Reply-To method by default — Fantastic! ❤️

[Lancetnik]

I am glad to improve this project together. But, this functionality has some limitations you should know: publish with reply_to is synchronous per RabbitBroker instanse by asyncio.Lock. This limitation is connected with RabbitMQ amqp.reply_to single consumer limitation. So, if you sending messages without creating asyncio.Task for each, It shouldn't be a problem

[Lancetnik]

Well, now we have another problem: miltiple consuming from reply_to is forbidden. So, RPC request is blocked by another request...

[Lancetnik]

So, we can use multiple RPC requests from different connections or from one at sync mode. But if we try to create multiple requests at one connection it caused to error.