374 lines
11 KiB
Python
374 lines
11 KiB
Python
"""
|
|
Neural Sync - Event Bus System
|
|
|
|
实现事件驱动架构的核心组件，支持异步事件发布/订阅模式。
|
|
"""
|
|
|
|
from enum import Enum
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from typing import Callable, Dict, List, Any, Optional
|
|
import asyncio
|
|
import logging
|
|
from contextlib import asynccontextmanager
|
|
|
|
from .utils.timezone import now_cn_aware
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class EventType(Enum):
|
|
"""系统事件类型定义"""
|
|
|
|
# 信号事件
|
|
SIGNAL_INGESTED = "signal.ingested"
|
|
SIGNAL_ANALYZED = "signal.analyzed"
|
|
SIGNAL_FEEDBACK = "signal.feedback"
|
|
SIGNAL_PRIORITY_CHANGED = "signal.priority_changed"
|
|
|
|
# 任务事件
|
|
TASK_ASSIGNED = "task.assigned"
|
|
TASK_STARTED = "task.started"
|
|
TASK_COMPLETED = "task.completed"
|
|
TASK_FAILED = "task.failed"
|
|
|
|
# 代码事件
|
|
CODE_COMMIT = "code.commit"
|
|
CODE_REVIEW_REQUESTED = "code.review_requested"
|
|
CODE_DRAFT_GENERATED = "code.draft_generated"
|
|
|
|
# 审计事件
|
|
AUDIT_STARTED = "audit.started"
|
|
AUDIT_COMPLETED = "audit.completed"
|
|
AUDIT_FAILED = "audit.failed"
|
|
|
|
# 合并事件
|
|
AUTO_MERGE_REQUESTED = "auto_merge.requested"
|
|
AUTO_MERGE_SUCCESS = "auto_merge.success"
|
|
AUTO_MERGE_FAILED = "auto_merge.failed"
|
|
AUTO_MERGE_COOLDOWN = "auto_merge.cooldown"
|
|
|
|
# 集成事件
|
|
PLANE_SYNC_SUCCESS = "integration.plane.success"
|
|
PLANE_SYNC_FAILED = "integration.plane.failed"
|
|
LARK_NOTIFY_SUCCESS = "integration.lark.success"
|
|
LARK_NOTIFY_FAILED = "integration.lark.failed"
|
|
|
|
# 项目域事件
|
|
DOMAIN_CREATED = "domain.created"
|
|
DOMAIN_UPDATED = "domain.updated"
|
|
DOMAIN_DELETED = "domain.deleted"
|
|
DOMAIN_SWITCHED = "domain.switched"
|
|
DOMAIN_SYNCING = "domain.syncing"
|
|
DOMAIN_SYNCED = "domain.synced"
|
|
|
|
# 项目事件
|
|
PROJECT_CREATED = "project.created"
|
|
PROJECT_UPDATED = "project.updated"
|
|
PROJECT_DELETED = "project.deleted"
|
|
SIGNAL_LINKED = "signal.linked"
|
|
SIGNAL_UNLINKED = "signal.unlinked"
|
|
|
|
|
|
@dataclass
|
|
class Event:
|
|
"""事件数据结构"""
|
|
|
|
type: EventType
|
|
data: Dict[str, Any]
|
|
timestamp: float = field(default_factory=lambda: now_cn_aware().timestamp())
|
|
source: Optional[str] = None
|
|
correlation_id: Optional[str] = None
|
|
metadata: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""转换为字典格式"""
|
|
return {
|
|
"type": self.type.value,
|
|
"data": self.data,
|
|
"timestamp": self.timestamp,
|
|
"source": self.source,
|
|
"correlation_id": self.correlation_id,
|
|
"metadata": self.metadata,
|
|
}
|
|
|
|
|
|
class EventBus:
|
|
"""
|
|
事件总线 - 核心事件分发系统
|
|
|
|
特性:
|
|
- 异步事件发布
|
|
- 多订阅者支持
|
|
- 事件过滤
|
|
- 错误处理
|
|
- 性能监控
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._subscribers: Dict[EventType, List[Callable]] = {}
|
|
self._wildcard_subscribers: List[Callable] = []
|
|
self._event_history: List[Event] = []
|
|
self._max_history = 1000
|
|
self._metrics = {
|
|
"published": 0,
|
|
"delivered": 0,
|
|
"failed": 0,
|
|
}
|
|
self._lock = asyncio.Lock()
|
|
self._running = False
|
|
|
|
@property
|
|
def metrics(self) -> Dict[str, int]:
|
|
"""获取事件指标"""
|
|
return self._metrics.copy()
|
|
|
|
@property
|
|
def history(self) -> List[Dict[str, Any]]:
|
|
"""获取事件历史"""
|
|
return [e.to_dict() for e in self._event_history]
|
|
|
|
async def start(self):
|
|
"""启动事件总线"""
|
|
self._running = True
|
|
logger.info("EventBus started")
|
|
|
|
async def stop(self):
|
|
"""停止事件总线"""
|
|
self._running = False
|
|
logger.info("EventBus stopped")
|
|
|
|
def subscribe(
|
|
self,
|
|
event_type: EventType,
|
|
handler: Callable,
|
|
filter_func: Optional[Callable[[Event], bool]] = None,
|
|
) -> Callable:
|
|
"""
|
|
订阅事件
|
|
|
|
Args:
|
|
event_type: 事件类型
|
|
handler: 事件处理函数 (async def handler(event: Event) -> None)
|
|
filter_func: 事件过滤函数 (def filter(event: Event) -> bool)
|
|
|
|
Returns:
|
|
取消订阅的函数
|
|
"""
|
|
|
|
async def wrapped_handler(event: Event):
|
|
if filter_func and not filter_func(event):
|
|
return
|
|
try:
|
|
await handler(event)
|
|
self._metrics["delivered"] += 1
|
|
except Exception as e:
|
|
logger.error(
|
|
f"Error in event handler for {event_type.value}: {e}",
|
|
exc_info=True,
|
|
)
|
|
self._metrics["failed"] += 1
|
|
|
|
# 包装为原始 handler 用于取消订阅
|
|
wrapped_handler.original = handler # type: ignore
|
|
|
|
async def _subscribe():
|
|
async with self._lock:
|
|
if event_type not in self._subscribers:
|
|
self._subscribers[event_type] = []
|
|
self._subscribers[event_type].append(wrapped_handler)
|
|
|
|
asyncio.create_task(_subscribe())
|
|
|
|
def unsubscribe():
|
|
asyncio.create_task(self._unsubscribe(event_type, wrapped_handler))
|
|
|
|
return unsubscribe
|
|
|
|
def subscribe_wildcard(
|
|
self,
|
|
handler: Callable,
|
|
filter_func: Optional[Callable[[Event], bool]] = None,
|
|
) -> Callable:
|
|
"""订阅所有事件（通配符订阅）"""
|
|
|
|
async def wrapped_handler(event: Event):
|
|
if filter_func and not filter_func(event):
|
|
return
|
|
try:
|
|
await handler(event)
|
|
self._metrics["delivered"] += 1
|
|
except Exception as e:
|
|
logger.error(
|
|
f"Error in wildcard event handler: {e}",
|
|
exc_info=True,
|
|
)
|
|
self._metrics["failed"] += 1
|
|
|
|
wrapped_handler.original = handler # type: ignore
|
|
|
|
async def _subscribe():
|
|
async with self._lock:
|
|
self._wildcard_subscribers.append(wrapped_handler)
|
|
|
|
asyncio.create_task(_subscribe())
|
|
|
|
def unsubscribe():
|
|
asyncio.create_task(self._unsubscribe_wildcard(wrapped_handler))
|
|
|
|
return unsubscribe
|
|
|
|
async def _unsubscribe(self, event_type: EventType, handler: Callable):
|
|
"""取消订阅"""
|
|
async with self._lock:
|
|
if event_type in self._subscribers:
|
|
self._subscribers[event_type] = [
|
|
h for h in self._subscribers[event_type] if h != handler
|
|
]
|
|
|
|
async def _unsubscribe_wildcard(self, handler: Callable):
|
|
"""取消通配符订阅"""
|
|
async with self._lock:
|
|
self._wildcard_subscribers = [
|
|
h for h in self._wildcard_subscribers if h != handler
|
|
]
|
|
|
|
async def publish(self, event: Event) -> bool:
|
|
"""
|
|
发布事件
|
|
|
|
Args:
|
|
event: 事件对象
|
|
|
|
Returns:
|
|
是否成功发布
|
|
"""
|
|
if not self._running:
|
|
logger.warning(f"EventBus not running, discarding event: {event.type.value}")
|
|
return False
|
|
|
|
self._metrics["published"] += 1
|
|
|
|
# 记录历史
|
|
self._event_history.append(event)
|
|
if len(self._event_history) > self._max_history:
|
|
self._event_history.pop(0)
|
|
|
|
# 通知订阅者
|
|
tasks = []
|
|
|
|
# 特定事件订阅者
|
|
if event.type in self._subscribers:
|
|
for handler in self._subscribers[event.type]:
|
|
tasks.append(handler(event))
|
|
|
|
# 通配符订阅者
|
|
for handler in self._wildcard_subscribers:
|
|
tasks.append(handler(event))
|
|
|
|
if tasks:
|
|
results = await asyncio.gather(*tasks, return_exceptions=True)
|
|
failed = sum(1 for r in results if isinstance(r, Exception))
|
|
if failed > 0:
|
|
self._metrics["failed"] += failed
|
|
|
|
logger.debug(
|
|
f"Published event {event.type.value} to {len(tasks)} subscribers"
|
|
)
|
|
return True
|
|
|
|
async def publish_and_wait(
|
|
self, event: Event, timeout: float = 5.0
|
|
) -> List[Any]:
|
|
"""
|
|
发布事件并等待所有处理完成
|
|
|
|
Args:
|
|
event: 事件对象
|
|
timeout: 超时时间（秒）
|
|
|
|
Returns:
|
|
所有处理器的返回值列表
|
|
"""
|
|
if not self._running:
|
|
raise RuntimeError("EventBus not running")
|
|
|
|
self._metrics["published"] += 1
|
|
self._event_history.append(event)
|
|
|
|
tasks = []
|
|
|
|
if event.type in self._subscribers:
|
|
tasks.extend(self._subscribers[event.type])
|
|
|
|
for handler in self._wildcard_subscribers:
|
|
tasks.append(handler(event))
|
|
|
|
if not tasks:
|
|
return []
|
|
|
|
try:
|
|
results = await asyncio.wait_for(
|
|
asyncio.gather(*tasks, return_exceptions=True),
|
|
timeout=timeout,
|
|
)
|
|
return results
|
|
except asyncio.TimeoutError:
|
|
logger.warning(
|
|
f"Event {event.type.value} processing timed out after {timeout}s"
|
|
)
|
|
raise
|
|
|
|
def clear_history(self):
|
|
"""清空事件历史"""
|
|
self._event_history.clear()
|
|
|
|
def get_subscriber_count(self, event_type: Optional[EventType] = None) -> int:
|
|
"""获取订阅者数量"""
|
|
if event_type:
|
|
return len(self._subscribers.get(event_type, []))
|
|
return sum(len(subs) for subs in self._subscribers.values()) + len(
|
|
self._wildcard_subscribers
|
|
)
|
|
|
|
|
|
# 全局事件总线实例
|
|
_global_event_bus: Optional[EventBus] = None
|
|
|
|
|
|
def get_event_bus() -> EventBus:
|
|
"""获取全局事件总线实例"""
|
|
global _global_event_bus
|
|
if _global_event_bus is None:
|
|
_global_event_bus = EventBus()
|
|
return _global_event_bus
|
|
|
|
|
|
@asynccontextmanager
|
|
async def event_bus_context():
|
|
"""事件总线上下文管理器"""
|
|
bus = get_event_bus()
|
|
await bus.start()
|
|
try:
|
|
yield bus
|
|
finally:
|
|
await bus.stop()
|
|
|
|
|
|
# 便捷函数
|
|
async def publish_event(event_type: EventType, data: Dict[str, Any], **kwargs) -> bool:
|
|
"""发布事件的便捷函数"""
|
|
event = Event(type=event_type, data=data, **kwargs)
|
|
return await get_event_bus().publish(event)
|
|
|
|
|
|
def subscribe_event(
|
|
event_type: EventType, handler: Callable
|
|
) -> Callable:
|
|
"""订阅事件的便捷函数"""
|
|
return get_event_bus().subscribe(event_type, handler)
|
|
|
|
|
|
def subscribe_all_events(handler: Callable) -> Callable:
|
|
"""订阅所有事件的便捷函数"""
|
|
return get_event_bus().subscribe_wildcard(handler)
|