# -*- 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()