37 lines
1.2 KiB
Python
37 lines
1.2 KiB
Python
import _thread
|
|
|
|
class SafeQueue:
|
|
"""A lightweight, thread-safe FIFO queue for MicroPython."""
|
|
def __init__(self, maxsize=20):
|
|
self._queue = []
|
|
self._lock = _thread.allocate_lock()
|
|
self.maxsize = maxsize
|
|
|
|
def put(self, item) -> bool:
|
|
"""Push an item to the end of the queue. Returns False if queue is full."""
|
|
with self._lock:
|
|
if len(self._queue) < self.maxsize:
|
|
self._queue.append(item)
|
|
return True
|
|
else:
|
|
print("[Queue Warning] Buffer full, dropping oldest message.")
|
|
self._queue.pop(0) # Drop oldest to make room
|
|
self._queue.append(item)
|
|
return False
|
|
|
|
def get(self):
|
|
"""Pop and return the oldest item from the queue, or None if empty."""
|
|
with self._lock:
|
|
if self._queue:
|
|
return self._queue.pop(0)
|
|
return None
|
|
|
|
def empty(self) -> bool:
|
|
"""Check if the queue has no items."""
|
|
with self._lock:
|
|
return len(self._queue) == 0
|
|
|
|
def size(self) -> int:
|
|
"""Return current number of queued items."""
|
|
with self._lock:
|
|
return len(self._queue) |