feat(huya): 协作台账增加 WebSocket 实时同步
- 后端新增进程内事件总线与 /api/huya/external-orders/ws\n- 所有写接口成功后发布轻量失效事件\n- 前端自动重连、静默刷新、断线兜底轮询
This commit is contained in:
@@ -0,0 +1,104 @@
|
||||
"""虎牙外部单协作台账的进程内实时事件总线。
|
||||
|
||||
写接口提交成功后只发布「台账失效」事件,前端收到后按自己的筛选条件静默重拉,
|
||||
不在事件里携带整行数据。当前生产是单 uvicorn worker,进程内队列足够;
|
||||
如果以后改成多 worker / 多实例,这里再换成 Redis Pub/Sub。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import itertools
|
||||
import threading
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
|
||||
class CollabEventHub:
|
||||
"""维护 revision 与当前 WebSocket 订阅队列。"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._lock = threading.Lock()
|
||||
self._revisions = itertools.count(1)
|
||||
self._current_revision = 0
|
||||
self._loop: asyncio.AbstractEventLoop | None = None
|
||||
self._subscribers: set[asyncio.Queue[dict[str, Any]]] = set()
|
||||
|
||||
@property
|
||||
def revision(self) -> int:
|
||||
with self._lock:
|
||||
return self._current_revision
|
||||
|
||||
def subscribe(self) -> asyncio.Queue[dict[str, Any]]:
|
||||
"""注册一个 WebSocket 订阅者;必须在事件循环线程里调用。"""
|
||||
loop = asyncio.get_running_loop()
|
||||
queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=100)
|
||||
with self._lock:
|
||||
self._loop = loop
|
||||
self._subscribers.add(queue)
|
||||
return queue
|
||||
|
||||
def unsubscribe(self, queue: asyncio.Queue[dict[str, Any]]) -> None:
|
||||
with self._lock:
|
||||
self._subscribers.discard(queue)
|
||||
|
||||
def ready_event(self) -> dict[str, Any]:
|
||||
with self._lock:
|
||||
revision = self._current_revision
|
||||
return {"type": "collab.ready", "revision": revision}
|
||||
|
||||
def ping_event(self) -> dict[str, Any]:
|
||||
with self._lock:
|
||||
revision = self._current_revision
|
||||
return {"type": "collab.ping", "revision": revision}
|
||||
|
||||
def publish(
|
||||
self,
|
||||
*,
|
||||
scope: str,
|
||||
action: str,
|
||||
actor_id: int | None = None,
|
||||
orders: list[dict[str, int]] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""发布一次失效事件;同步接口在线程池里调用也安全。"""
|
||||
with self._lock:
|
||||
revision = next(self._revisions)
|
||||
self._current_revision = revision
|
||||
loop = self._loop
|
||||
subscribers = list(self._subscribers)
|
||||
|
||||
event = {
|
||||
"type": "collab.invalidate",
|
||||
"revision": revision,
|
||||
"scope": scope,
|
||||
"action": action,
|
||||
"actor_id": actor_id,
|
||||
"orders": orders or [],
|
||||
"at": datetime.now(UTC).isoformat(),
|
||||
}
|
||||
if loop is None or not subscribers:
|
||||
return event
|
||||
|
||||
for queue in subscribers:
|
||||
try:
|
||||
loop.call_soon_threadsafe(self._offer, queue, event)
|
||||
except RuntimeError:
|
||||
# 事件循环已经关闭:忽略,下次连接会重新拉全量。
|
||||
continue
|
||||
return event
|
||||
|
||||
@staticmethod
|
||||
def _offer(queue: asyncio.Queue[dict[str, Any]], event: dict[str, Any]) -> None:
|
||||
"""队列满时丢掉旧事件,保留最新一次失效通知。"""
|
||||
while queue.full():
|
||||
try:
|
||||
queue.get_nowait()
|
||||
except asyncio.QueueEmpty:
|
||||
break
|
||||
try:
|
||||
queue.put_nowait(event)
|
||||
except asyncio.QueueFull:
|
||||
pass
|
||||
|
||||
|
||||
collab_events = CollabEventHub()
|
||||
Reference in New Issue
Block a user