primitives/queue.py

   1 # queue.py: adapted from uasyncio V2
   2
   3 # Copyright (c) 2018-2020 Peter Hinch
   4 # Released under the MIT License (MIT) - see LICENSE file
   5
   6 # Code is based on Paul Sokolovsky's work.
   7 # This is a temporary solution until uasyncio V3 gets an efficient official version
   8
   9 import uasyncio as asyncio
  10
  11
  12 # Exception raised by get_nowait().
  13 class QueueEmpty(Exception):
  14     pass
  15
  16
  17 # Exception raised by put_nowait().
  18 class QueueFull(Exception):
  19     pass
  20
  21 class Queue:
  22
  23     def __init__(self, maxsize=0):
  24         self.maxsize = maxsize
  25         self._queue = []
  26         self._evput = asyncio.Event()  # Triggered by put, tested by get
  27         self._evget = asyncio.Event()  # Triggered by get, tested by put
  28
  29     def _get(self):
  30         self._evget.set()  # Schedule all tasks waiting on get
  31         self._evget.clear()
  32         return self._queue.pop(0)
  33
  34     async def get(self):  #  Usage: item = await queue.get()
  35         while self.empty():  # May be multiple tasks waiting on get()
  36             # Queue is empty, suspend task until a put occurs
  37             # 1st of N tasks gets, the rest loop again
  38             await self._evput.wait()
  39         return self._get()
  40
  41     def get_nowait(self):  # Remove and return an item from the queue.
  42         # Return an item if one is immediately available, else raise QueueEmpty.
  43         if self.empty():
  44             raise QueueEmpty()
  45         return self._get()
  46
  47     def _put(self, val):
  48         self._evput.set()  # Schedule tasks waiting on put
  49         self._evput.clear()
  50         self._queue.append(val)
  51
  52     async def put(self, val):  # Usage: await queue.put(item)
  53         while self.full():
  54             # Queue full
  55             await self._evget.wait()
  56             # Task(s) waiting to get from queue, schedule first Task
  57         self._put(val)
  58
  59     def put_nowait(self, val):  # Put an item into the queue without blocking.
  60         if self.full():
  61             raise QueueFull()
  62         self._put(val)
  63
  64     def qsize(self):  # Number of items in the queue.
  65         return len(self._queue)
  66
  67     def empty(self):  # Return True if the queue is empty, False otherwise.
  68         return len(self._queue) == 0
  69
  70     def full(self):  # Return True if there are maxsize items in the queue.
  71         # Note: if the Queue was initialized with maxsize=0 (the default) or
  72         # any negative number, then full() is never True.
  73         return self.maxsize > 0 and self.qsize() >= self.maxsize