A queue, useful for coordinating producer and consumer coroutines. If maxsize is less than or equal to zero, the queue size is infinite. If it is an integer greater than 0, then "await put()" will block when the queue reaches maxsize, until an item is removed by get(). Unlike
| 19 | |
| 20 | |
| 21 | class Queue(mixins._LoopBoundMixin): |
| 22 | """A queue, useful for coordinating producer and consumer coroutines. |
| 23 | |
| 24 | If maxsize is less than or equal to zero, the queue size is infinite. If it |
| 25 | is an integer greater than 0, then "await put()" will block when the |
| 26 | queue reaches maxsize, until an item is removed by get(). |
| 27 | |
| 28 | Unlike the standard library Queue, you can reliably know this Queue's size |
| 29 | with qsize(), since your single-threaded asyncio application won't be |
| 30 | interrupted between calling qsize() and doing an operation on the Queue. |
| 31 | """ |
| 32 | |
| 33 | def __init__(self, maxsize=0): |
| 34 | self._maxsize = maxsize |
| 35 | |
| 36 | # Futures. |
| 37 | self._getters = collections.deque() |
| 38 | # Futures. |
| 39 | self._putters = collections.deque() |
| 40 | self._unfinished_tasks = 0 |
| 41 | self._finished = locks.Event() |
| 42 | self._finished.set() |
| 43 | self._init(maxsize) |
| 44 | |
| 45 | # These three are overridable in subclasses. |
| 46 | |
| 47 | def _init(self, maxsize): |
| 48 | self._queue = collections.deque() |
| 49 | |
| 50 | def _get(self): |
| 51 | return self._queue.popleft() |
| 52 | |
| 53 | def _put(self, item): |
| 54 | self._queue.append(item) |
| 55 | |
| 56 | # End of the overridable methods. |
| 57 | |
| 58 | def _wakeup_next(self, waiters): |
| 59 | # Wake up the next waiter (if any) that isn't cancelled. |
| 60 | while waiters: |
| 61 | waiter = waiters.popleft() |
| 62 | if not waiter.done(): |
| 63 | waiter.set_result(None) |
| 64 | break |
| 65 | |
| 66 | def __repr__(self): |
| 67 | return f'<{type(self).__name__} at {id(self):#x} {self._format()}>' |
| 68 | |
| 69 | def __str__(self): |
| 70 | return f'<{type(self).__name__} {self._format()}>' |
| 71 | |
| 72 | __class_getitem__ = classmethod(GenericAlias) |
| 73 | |
| 74 | def _format(self): |
| 75 | result = f'maxsize={self._maxsize!r}' |
| 76 | if getattr(self, '_queue', None): |
| 77 | result += f' _queue={list(self._queue)!r}' |
| 78 | if self._getters: |
no outgoing calls
no test coverage detected