Hi, I am working on a sans-io implementation of amqp client. So far I only implemented asyncio adapter - and I faced the issue that if one coroutine does nothing but writing and draining on a non paused connection, instead of yielding to the event loop, drain short-circuits making it impossible for other coroutines to be executed in between.
My use case is publishing lots of messages, and publishing in amqp do not have a reply - so there’s nothing I can wait for. However, if publish fails (for example, because of wrong serialization code), the broker sends an exception asynchronously which is read by another coroutine running as a task in “background” - and since drain short-circuits, this task never executes and I can’t handle the exception until all messages are published.
I’m looking for an idea about how to deal with such situation.
Of course, I can put (await asyncio.sleep(0)) right after (await writer.drain), but it makes everything 2.5-3.5 times slower. Or I can pretend this is not an issue and nobody will ever try to send millions of messages and some of them could be malformed.
I’d like to know if there’s something in between - that can be as fast as the current version and can yield to the event loop somehow to receive messages from the broker.
I thank in advance anyone who share their ideas!