45 lines
1.0 KiB
Python
45 lines
1.0 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
"""SSE 事件总线 —— 兼容旧通道（前端 MQTT 不可用时回退）
|
||
|
|
所有 MQTT 控制指令同时推送到 SSE /api/events
|
||
|
|
"""
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import json
|
||
|
|
|
||
|
|
|
||
|
|
class EventBus:
|
||
|
|
def __init__(self):
|
||
|
|
self._queues = set()
|
||
|
|
self._loop = None
|
||
|
|
|
||
|
|
def set_loop(self, loop):
|
||
|
|
self._loop = loop
|
||
|
|
|
||
|
|
def subscribe(self):
|
||
|
|
q = asyncio.Queue(maxsize=256)
|
||
|
|
self._queues.add(q)
|
||
|
|
return q
|
||
|
|
|
||
|
|
def unsubscribe(self, q):
|
||
|
|
self._queues.discard(q)
|
||
|
|
|
||
|
|
def emit(self, payload: dict):
|
||
|
|
"""线程安全：从任意线程广播事件"""
|
||
|
|
if not self._loop:
|
||
|
|
return
|
||
|
|
try:
|
||
|
|
asyncio.run_coroutine_threadsafe(self._emit(payload), self._loop)
|
||
|
|
except Exception: # noqa: BLE001
|
||
|
|
pass
|
||
|
|
|
||
|
|
async def _emit(self, payload):
|
||
|
|
data = json.dumps(payload, ensure_ascii=False)
|
||
|
|
for q in list(self._queues):
|
||
|
|
try:
|
||
|
|
q.put_nowait(data)
|
||
|
|
except asyncio.QueueFull:
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
bus = EventBus()
|