Open greatvovan opened 6 years ago
There's an asyncio task running the dispatcher that reads messages from the server and calls handler methods; Channel.basic_deliver
is called to deliver messages to your consumer callback (https://github.com/Polyconseil/aioamqp/blob/cbb137920b000db3d99adfa986ad9a7dc9b58876/aioamqp/channel.py#L641). The problem is the callback (your on_message function) is executed in the same task as the dispatcher, so while the callback hasn't returned, no messages will be handled from the server.
Confirm select changes the messaging with the server so for the basic.public messages you send (by calling Channel.publish
), the server replies with an ack message. Publish does not return until it has received that ack. So what happens is that Channel.dispatch_frame
is calling Channel.basic_deliver
, which is stuck waiting for on_message to return, but on_message is calling publish_and_ack which waits for and ack for the server, but Channel.dispatch_frame
is stuck and will cannot read the next message.
If I'm reading the code correctly, having another channel should not help if confirm_select() is used, because the connection's main dispatcher just delegates to the channels, and doesn't process the next message from the server until the channel method returns. The publisher confirms are channel-specific, so it might be that it works because there's not actually any confirms used on the channel.
One workaround is to call loop.create_task for publish_and_ack(), which allows the library's dispatcher to process the next messages. But then you have to clean up those tasks once they finish somewhere else to avoid warnings.
If I'm reading the code correctly, having another channel should not help if confirm_select() is used, because the connection's main dispatcher just delegates to the channels
But it does help. Tried to set up yet another (third) channel, it works with it too. To sum up:
consumer
channel – works,publisher
channel – does not work,yet_another_channel
– works.Looks like it does not works only in that channel which was used to call confirm_select()
.
One workaround is to call loop.create_task for publish_and_ack()
Yes, I mentioned this at second point in the initial post. I don't like it and at the same time another workaround (enabling confirms in a separate channel) work pretty well. Are there any pros to use your workaround, namely?
My service reads messages from one queue and outputs processed messages to another exchange. I want to ack incoming message only after successful publishing of outgoing message. It worked well untill I met some issues with lost messages and I decided to enable 'publisher confirms' feature. After this my program started to hang forever on
publish()
. Here is the simple code to illustrate problem.I tried to play with it and found the following:
await self.publisher.confirm_select()
.publish_and_ack()
in task (e.g.asyncio.ensure_future()
), but I dislike releasing the task to 'free float'.consumer
orchannel
(which are the same) inon_message()
instead ofpublisher
.consumer
or onlypublisher
).confirm_select()
from separate channel (e.g.publisher
) and use another channel everywhere.It seems really weird. Am I doing everything correctly? What is the best practice of using channels with this lib?