consumer/producer with asyncio

[David Rios]

Hello everyone,

I'm trying to implement a producer/consumer using asyncio, but I have
some doubts about the best way to do it. I already have a working
implementation using threads, and want to port it to asyncio for fun,
to learn and also to avoid some problems that I have using threads,
namely that sometimes the script locks and never finish.

What the script have to do is to connect to a server, stream content
from it, and send it for processing as items arrive. The processing
part is to connect to other servers based on the item, send the item
to process and wait for the response.

The threaded version sets a pool of consumer threads that fetch items
from a queue, while on the main thread the queue is populated.

Special requirements that I have are that the script should be
interruptible and should stop cleanly on errors. When I interrupt it
or any part throws an exception for any reason, the script should stop
producing items, the workers should finish the current job and then
stop processing new items, at which point the script should perform
appropriate housekeeping and exit.

I have a mock implementation using asyncio here:
https://gist.github.com/davidrios/011d044b5e7510f085dd

But I think it is quite ugly, particularly all the nesting.

What do you think?