Some thoughts on AsyncBufferedReader

[grant-allan-ctct]

I'm pretty new to this library and a learner when it comes to Python and especially asyncio, so there may be not much virtue in these ideas, but y'all seem like a friendly bunch so I'll put pen to paper anyway and see what y'all think.

Background info:

The AsyncBufferedReader holds a self.buffer that's an asyncio.Queue, as can be see here.

Such a queue has an internal counter _unfinished_tasks that's incremented each time that put_nowait is called. Is also has an event-lock thingy called _finished which has its set() method called at queue construction and its clear() method when put_nowait is called. Looking at the other places where _unfinished_tasks and _finished used, we can see how the asyncio.Queue achieves the functionality that's documented for its join() method, viz. waiting until all queued items have been "processed" and then closing shop (see here).

Wonderings:

Allow async for loops to end?

The implementation of __anext__ on the AsyncBufferedReader class is written in such a way that iterating over the collection will never end, and further, the iterator seems willing to wait forever for the next message to arrive. To clarify, I'll mention the following example code snippet is taken from the docstring for AyncBufferedReader here:

        async for msg in reader:
            print(msg)

If we write something like that in our code though, what happens that any lines of code that come afterwards are effectively unreachable. I realize that it's possible that this is exactly what we want, but there could be some other options:

1. keep track of the _unfinished_tasks that the queue is tracking, and have __anext__ raise a StopAsyncIteration if it gets to 0. This will mark the async for loop as being complete, and program execution can then continue. In order to do this, we would need to make a call to self.buffer.task_done() every time we successfully removed an item from the queue using self.buffer.get(), as described here.
2. employ something along the lines of await asyncio.wait_for(self.buffer.get(), timeout) inside __anext__ and check for occurrence of a timeout; if one occurs, then either:
   1. we could raise StopAsyncIteration, or
   2. return None and allow the universe to decide how to respond, e.g. maybe, or maybe not, break out of the loop

Provide an optional timeout for get_message ?

Speaking of timeouts, it might be nice for the AsyncBufferedReader to have a common default timeout that it can share between the __anext__ implementation and the get_message implementation (here). And, if get_message were given an optional timeout argument then __anext__ could just call return await get_message() (depending on outcome of prior questions).

Remove a smidge of DRY violation ?

Actually, seems like having __anext__ just call return await get_message() could happen already 😎 .