50 lines
1.2 KiB
Python
50 lines
1.2 KiB
Python
"""
|
||
Шина SSE-событий.
|
||
Позволяет webhook_handler.py публиковать события,
|
||
а webhook_server.py — раздавать их браузеру через /stream.
|
||
"""
|
||
|
||
import json
|
||
import queue
|
||
import threading
|
||
from collections import deque
|
||
|
||
_history: deque = deque(maxlen=100)
|
||
_clients: list[queue.Queue] = []
|
||
_lock = threading.Lock()
|
||
|
||
|
||
def broadcast(data: dict):
|
||
"""Публикует событие всем подключённым SSE-клиентам и в историю."""
|
||
_history.append(data)
|
||
with _lock:
|
||
for q in list(_clients):
|
||
try:
|
||
q.put_nowait(data)
|
||
except queue.Full:
|
||
pass
|
||
|
||
|
||
def subscribe() -> queue.Queue:
|
||
"""Подписывает нового SSE-клиента. Возвращает очередь с историей."""
|
||
q: queue.Queue = queue.Queue(maxsize=200)
|
||
with _lock:
|
||
_clients.append(q)
|
||
for item in list(_history):
|
||
try:
|
||
q.put_nowait(item)
|
||
except queue.Full:
|
||
break
|
||
return q
|
||
|
||
|
||
def unsubscribe(q: queue.Queue):
|
||
with _lock:
|
||
try:
|
||
_clients.remove(q)
|
||
except ValueError:
|
||
pass
|
||
|
||
|
||
def get_history() -> list:
|
||
return list(_history)
|