feat(channel/events): 新增事件总线模块及相关类型定义
实现了完整的频道事件总线系统，包含： 1. 事件主题枚举类，定义了全量可用事件主题 2. 钩子函数到事件主题的映射字典 3. 异步事件总线实现，支持并行/顺序/首结果发布 4. 订阅、取消订阅、启用禁用订阅等管理能力 5. 订阅优先级排序、超时和错误处理机制 6. 对外暴露的统一导出接口
This commit is contained in:
parent
058513022a
commit
c9d0a26f52
11
backend/package/yuxi/channel/events/__init__.py
Normal file
11
backend/package/yuxi/channel/events/__init__.py
Normal file
@ -0,0 +1,11 @@
|
||||
from yuxi.channel.events.bus import ChannelEventBus, EventHandler, PublishResult, Subscription
|
||||
from yuxi.channel.events.types import HOOK_EVENT_TO_TOPIC, EventTopic
|
||||
|
||||
__all__ = [
|
||||
"ChannelEventBus",
|
||||
"EventHandler",
|
||||
"EventTopic",
|
||||
"HOOK_EVENT_TO_TOPIC",
|
||||
"PublishResult",
|
||||
"Subscription",
|
||||
]
|
||||
221
backend/package/yuxi/channel/events/bus.py
Normal file
221
backend/package/yuxi/channel/events/bus.py
Normal file
@ -0,0 +1,221 @@
|
||||
import asyncio
|
||||
import bisect
|
||||
import fnmatch
|
||||
import logging
|
||||
from collections.abc import Callable, Coroutine
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
EventHandler = Callable[..., Coroutine[Any, Any, None]]
|
||||
_DEFAULT_TIMEOUT = 30.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class Subscription:
|
||||
callback: EventHandler
|
||||
priority: int = 100
|
||||
enabled: bool = True
|
||||
key: str = ""
|
||||
topic: str = ""
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return id(self.callback)
|
||||
|
||||
|
||||
@dataclass
|
||||
class PublishResult:
|
||||
topic: str
|
||||
subscriber_count: int = 0
|
||||
error_count: int = 0
|
||||
errors: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
class ChannelEventBus:
|
||||
def __init__(self) -> None:
|
||||
self._subscriptions: dict[str, list[Subscription]] = {}
|
||||
|
||||
def subscribe(
|
||||
self,
|
||||
topic: str,
|
||||
callback: EventHandler,
|
||||
*,
|
||||
priority: int = 100,
|
||||
key: str = "",
|
||||
) -> Subscription:
|
||||
sub = Subscription(callback=callback, priority=priority, key=key, topic=topic)
|
||||
subs = self._subscriptions.setdefault(topic, [])
|
||||
idx = bisect.bisect_left([s.priority for s in subs], priority)
|
||||
subs.insert(idx, sub)
|
||||
logger.debug("Subscribe: topic=%s key=%s priority=%d", topic, key, priority)
|
||||
return sub
|
||||
|
||||
def unsubscribe(self, topic: str, callback: EventHandler) -> None:
|
||||
subs = self._subscriptions.get(topic)
|
||||
if subs is None:
|
||||
return
|
||||
self._subscriptions[topic] = [s for s in subs if s.callback != callback]
|
||||
if not self._subscriptions[topic]:
|
||||
del self._subscriptions[topic]
|
||||
logger.debug("Unsubscribe: topic=%s remaining=%d", topic, len(self._subscriptions.get(topic, [])))
|
||||
|
||||
def enable(self, topic: str, callback: EventHandler) -> bool:
|
||||
for sub in self._subscriptions.get(topic, []):
|
||||
if sub.callback == callback:
|
||||
sub.enabled = True
|
||||
return True
|
||||
return False
|
||||
|
||||
def disable(self, topic: str, callback: EventHandler) -> bool:
|
||||
for sub in self._subscriptions.get(topic, []):
|
||||
if sub.callback == callback:
|
||||
sub.enabled = False
|
||||
return True
|
||||
return False
|
||||
|
||||
def _match_topics(self, topic: str) -> list[str]:
|
||||
if topic in self._subscriptions:
|
||||
return [topic]
|
||||
|
||||
matched = []
|
||||
for stored in self._subscriptions:
|
||||
if fnmatch.fnmatch(topic, stored) or fnmatch.fnmatch(stored, topic):
|
||||
matched.append(stored)
|
||||
return matched
|
||||
|
||||
async def publish(
|
||||
self,
|
||||
topic: str,
|
||||
*args: Any,
|
||||
timeout: float | None = None,
|
||||
**kwargs: Any,
|
||||
) -> PublishResult:
|
||||
topics = self._match_topics(topic)
|
||||
if not topics:
|
||||
return PublishResult(topic=topic)
|
||||
|
||||
effective_timeout = timeout if timeout is not None else _DEFAULT_TIMEOUT
|
||||
|
||||
all_subs: list[Subscription] = []
|
||||
for t in topics:
|
||||
for sub in self._subscriptions.get(t, []):
|
||||
if sub.enabled:
|
||||
all_subs.append(sub)
|
||||
|
||||
all_subs.sort(key=lambda s: s.priority)
|
||||
|
||||
async def _safe_invoke(sub: Subscription) -> str | None:
|
||||
try:
|
||||
await asyncio.wait_for(sub.callback(*args, **kwargs), timeout=effective_timeout)
|
||||
return None
|
||||
except asyncio.TimeoutError:
|
||||
logger.error("EventBus subscriber timed out after %.1fs: topic=%s key=%s", effective_timeout, topic, sub.key)
|
||||
return f"timeout: {sub.key}"
|
||||
except Exception:
|
||||
logger.exception("EventBus subscriber failed: topic=%s key=%s", topic, sub.key)
|
||||
return f"error: {sub.key}"
|
||||
|
||||
results = await asyncio.gather(*(_safe_invoke(s) for s in all_subs))
|
||||
|
||||
errors = [e for e in results if e is not None]
|
||||
return PublishResult(
|
||||
topic=topic,
|
||||
subscriber_count=len(all_subs),
|
||||
error_count=len(errors),
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
async def publish_sequential(
|
||||
self,
|
||||
topic: str,
|
||||
initial: Any = None,
|
||||
*args: Any,
|
||||
timeout: float | None = None,
|
||||
**kwargs: Any,
|
||||
) -> tuple[Any, PublishResult]:
|
||||
topics = self._match_topics(topic)
|
||||
if not topics:
|
||||
return initial, PublishResult(topic=topic)
|
||||
|
||||
effective_timeout = timeout if timeout is not None else _DEFAULT_TIMEOUT
|
||||
|
||||
all_subs: list[Subscription] = []
|
||||
for t in topics:
|
||||
for sub in self._subscriptions.get(t, []):
|
||||
if sub.enabled:
|
||||
all_subs.append(sub)
|
||||
|
||||
all_subs.sort(key=lambda s: s.priority)
|
||||
|
||||
result = initial
|
||||
errors: list[str] = []
|
||||
for sub in all_subs:
|
||||
try:
|
||||
next_result = await asyncio.wait_for(sub.callback(result, *args, **kwargs), timeout=effective_timeout)
|
||||
if next_result is not None:
|
||||
result = next_result
|
||||
except asyncio.TimeoutError:
|
||||
logger.error("EventBus sequential subscriber timed out: topic=%s key=%s", topic, sub.key)
|
||||
errors.append(f"timeout: {sub.key}")
|
||||
except Exception:
|
||||
logger.exception("EventBus sequential subscriber failed: topic=%s key=%s", topic, sub.key)
|
||||
errors.append(f"error: {sub.key}")
|
||||
|
||||
return result, PublishResult(
|
||||
topic=topic,
|
||||
subscriber_count=len(all_subs),
|
||||
error_count=len(errors),
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
async def publish_first(
|
||||
self,
|
||||
topic: str,
|
||||
*args: Any,
|
||||
timeout: float | None = None,
|
||||
**kwargs: Any,
|
||||
) -> tuple[Any, PublishResult]:
|
||||
topics = self._match_topics(topic)
|
||||
if not topics:
|
||||
return None, PublishResult(topic=topic)
|
||||
|
||||
effective_timeout = timeout if timeout is not None else _DEFAULT_TIMEOUT
|
||||
|
||||
all_subs: list[Subscription] = []
|
||||
for t in topics:
|
||||
for sub in self._subscriptions.get(t, []):
|
||||
if sub.enabled:
|
||||
all_subs.append(sub)
|
||||
|
||||
all_subs.sort(key=lambda s: s.priority)
|
||||
|
||||
for sub in all_subs:
|
||||
try:
|
||||
result = await asyncio.wait_for(sub.callback(*args, **kwargs), timeout=effective_timeout)
|
||||
if result is not None:
|
||||
return result, PublishResult(topic=topic, subscriber_count=1)
|
||||
except asyncio.TimeoutError:
|
||||
logger.error("EventBus first subscriber timed out: topic=%s key=%s", topic, sub.key)
|
||||
except Exception:
|
||||
logger.exception("EventBus first subscriber failed: topic=%s key=%s", topic, sub.key)
|
||||
|
||||
return None, PublishResult(topic=topic, subscriber_count=len(all_subs), error_count=len(all_subs),
|
||||
errors=[f"all_failed: {s.key}" for s in all_subs])
|
||||
|
||||
def subscriber_count(self, topic: str | None = None) -> int:
|
||||
if topic is not None:
|
||||
return len(self._subscriptions.get(topic, []))
|
||||
return sum(len(subs) for subs in self._subscriptions.values())
|
||||
|
||||
def list_subscriptions(self, topic: str | None = None) -> dict[str, list[Subscription]]:
|
||||
if topic is not None:
|
||||
subs = self._subscriptions.get(topic)
|
||||
return {topic: list(subs)} if subs else {}
|
||||
return {k: list(v) for k, v in self._subscriptions.items()}
|
||||
|
||||
def clear(self) -> None:
|
||||
self._subscriptions.clear()
|
||||
|
||||
def topics(self) -> list[str]:
|
||||
return list(self._subscriptions.keys())
|
||||
45
backend/package/yuxi/channel/events/types.py
Normal file
45
backend/package/yuxi/channel/events/types.py
Normal file
@ -0,0 +1,45 @@
|
||||
from enum import StrEnum
|
||||
|
||||
|
||||
class EventTopic(StrEnum):
|
||||
CHANNEL_STARTING = "channel.starting"
|
||||
CHANNEL_STOPPING = "channel.stopping"
|
||||
CHANNEL_RUNNING = "channel.running"
|
||||
CHANNEL_ERROR = "channel.error"
|
||||
CHANNEL_HEALTH = "channel.health"
|
||||
|
||||
GATEWAY_START = "gateway.start"
|
||||
GATEWAY_STOP = "gateway.stop"
|
||||
|
||||
MESSAGE_RECEIVED = "message.received"
|
||||
MESSAGE_SENDING = "message.sending"
|
||||
MESSAGE_SENT = "message.sent"
|
||||
|
||||
BEFORE_AGENT_RUN = "agent.before_run"
|
||||
AFTER_AGENT_RUN = "agent.after_run"
|
||||
AGENT_BOOTSTRAP = "agent.bootstrap"
|
||||
|
||||
CONFIG_CHANGED = "config.changed"
|
||||
CONFIG_RELOADED = "config.reloaded"
|
||||
CONFIG_WRITE = "config.write"
|
||||
|
||||
SESSION_START = "session.start"
|
||||
SESSION_END = "session.end"
|
||||
|
||||
|
||||
HOOK_EVENT_TO_TOPIC: dict[str, str] = {
|
||||
"on_channel_starting": EventTopic.CHANNEL_STARTING,
|
||||
"on_channel_stopping": EventTopic.CHANNEL_STOPPING,
|
||||
"on_channel_running": EventTopic.CHANNEL_RUNNING,
|
||||
"on_channel_error": EventTopic.CHANNEL_ERROR,
|
||||
"on_gateway_start": EventTopic.GATEWAY_START,
|
||||
"on_gateway_stop": EventTopic.GATEWAY_STOP,
|
||||
"on_message_received": EventTopic.MESSAGE_RECEIVED,
|
||||
"on_message_sending": EventTopic.MESSAGE_SENDING,
|
||||
"on_message_sent": EventTopic.MESSAGE_SENT,
|
||||
"on_before_agent_run": EventTopic.BEFORE_AGENT_RUN,
|
||||
"on_after_agent_run": EventTopic.AFTER_AGENT_RUN,
|
||||
"on_agent_bootstrap": EventTopic.AGENT_BOOTSTRAP,
|
||||
"on_session_start": EventTopic.SESSION_START,
|
||||
"on_session_end": EventTopic.SESSION_END,
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user