44 lines
1.2 KiB
Python
44 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import time
|
|
|
|
|
|
class AdminEventBus:
|
|
def __init__(self):
|
|
self._subscribers: set[asyncio.Queue] = set()
|
|
self._counter = 0
|
|
self._lock = asyncio.Lock()
|
|
|
|
async def subscribe(self) -> asyncio.Queue:
|
|
queue: asyncio.Queue = asyncio.Queue(maxsize=64)
|
|
async with self._lock:
|
|
self._subscribers.add(queue)
|
|
return queue
|
|
|
|
async def unsubscribe(self, queue: asyncio.Queue) -> None:
|
|
async with self._lock:
|
|
self._subscribers.discard(queue)
|
|
|
|
def subscriber_count(self) -> int:
|
|
return len(self._subscribers)
|
|
|
|
async def publish(self, event_type: str, payload):
|
|
self._counter += 1
|
|
event = {
|
|
"id": self._counter,
|
|
"type": event_type,
|
|
"ts": time.time(),
|
|
"payload": payload,
|
|
}
|
|
for queue in list(self._subscribers):
|
|
if queue.full():
|
|
try:
|
|
queue.get_nowait()
|
|
except asyncio.QueueEmpty:
|
|
pass
|
|
try:
|
|
queue.put_nowait(event)
|
|
except asyncio.QueueFull:
|
|
pass
|