Kris
b88c0ae29e
新增完整的 channels 限界上下文模块，包含契约层、领域核心层、应用服务、管道编排、插件体系、基础设施组合根等全层级代码，新增飞书与微信 iLink 渠道插件基础结构，补充各类 DTO、端口协议与领域服务实现。
225 lines
9.5 KiB
Python
225 lines
9.5 KiB
Python
"""进程内事件总线。
|
||
|
||
实现 ``PluginHost.publishEvent``，按事件类型分发领域事件到订阅者。仅
|
||
进程内分发，不引入消息中间件。订阅管理采用读无锁、写加锁策略，与
|
||
``core/registry`` 一致。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import threading
|
||
import uuid
|
||
from dataclasses import dataclass
|
||
|
||
from yuxi.channels.contract.dtos.event import OutboxStateChangedEvent
|
||
from yuxi.channels.contract.dtos.plugin import DomainEvent
|
||
from yuxi.channels.contract.plugin.extension_point import EventSubscription
|
||
from yuxi.channels.contract.plugin.manifest import FailurePolicy
|
||
from yuxi.channels.contract.ports.driven.logger_port import LoggerPort
|
||
from yuxi.channels.core.service.degradation_manager import DegradationManager
|
||
|
||
__all__ = ["EventBus"]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class _SubscriptionEntry:
|
||
"""内部订阅条目。
|
||
|
||
关联插件 ID 与事件订阅，用于注销时按插件 ID 过滤。``EventSubscription``
|
||
契约不含插件 ID 字段，故在总线内部维护关联关系。
|
||
|
||
字段：
|
||
plugin_id: 插件 ID。
|
||
subscription: 事件订阅值对象。
|
||
"""
|
||
|
||
plugin_id: str
|
||
subscription: EventSubscription
|
||
|
||
|
||
class EventBus:
|
||
"""进程内事件总线。
|
||
|
||
实现 ``PluginHost.publishEvent``，按事件类型查找订阅者并按优先级调用
|
||
handler，分发 21 类领域事件。不引入消息中间件，仅进程内分发。
|
||
|
||
线程安全：订阅管理采用读无锁、写加锁策略（与 ``core/registry`` 一致），
|
||
读操作直接访问内存数据结构，写操作通过 ``threading.Lock`` 保护。
|
||
|
||
关键约束：
|
||
- handler 异常不得拖垮宿主（FR-36）：捕获异常并记录日志，不中断
|
||
其他订阅者。
|
||
- 订阅按 ``priority`` 升序排列（越小越先执行，与扩展点优先级规则
|
||
一致）。
|
||
- handler 超时保护：单个 handler 调用施加 ``_HANDLER_TIMEOUT``
|
||
超时控制，防止恶意/故障 handler 阻塞分发。
|
||
- 失败策略联动：handler 抛错时按 ``subscription.failure_strategy``
|
||
处理，``CIRCUIT_BREAK`` / ``ISOLATE`` 触发 ``DegradationManager``
|
||
降级，``DEGRADE`` 仅记录告警。
|
||
|
||
依赖：
|
||
logger: 日志被驱动端口，记录 handler 失败等告警。
|
||
degradation_manager: 降级管理服务，用于在 handler 失败时按策略
|
||
触发插件降级/熔断/隔离。
|
||
|
||
关联 FR：FR-36。
|
||
"""
|
||
|
||
#: 单个事件 handler 调用超时（秒），防止恶意/故障 handler 阻塞分发
|
||
_HANDLER_TIMEOUT: float = 5.0
|
||
|
||
def __init__(
|
||
self,
|
||
logger: LoggerPort,
|
||
degradation_manager: DegradationManager,
|
||
) -> None:
|
||
"""初始化事件总线。
|
||
|
||
参数：
|
||
logger: 日志被驱动端口，用于记录 handler 失败等告警。
|
||
degradation_manager: 降级管理服务，用于在 handler 失败时按
|
||
``failure_strategy`` 触发插件降级/熔断/隔离。
|
||
"""
|
||
self._logger = logger
|
||
self._degradation = degradation_manager
|
||
self._lock = threading.Lock()
|
||
# 按事件类型索引订阅条目列表
|
||
self._subscriptions: dict[str, list[_SubscriptionEntry]] = {}
|
||
|
||
async def publish(self, event: DomainEvent) -> None:
|
||
"""发布事件，按优先级分发到所有订阅者。
|
||
|
||
按 ``event.event_type`` 查找订阅者，按 ``priority`` 升序调用 handler。
|
||
每个 handler 调用施加 ``_HANDLER_TIMEOUT`` 超时控制，超时或异常
|
||
按 ``subscription.failure_strategy`` 处理（FR-36）：
|
||
|
||
- DEGRADE：记录告警，跳过本次调用，继续其他订阅者。
|
||
- CIRCUIT_BREAK / ISOLATE：调用 ``DegradationManager.onPluginFailed``
|
||
触发插件降级，然后继续其他订阅者。
|
||
|
||
参数：
|
||
event: 领域事件 DTO。
|
||
"""
|
||
# 读无锁：取事件类型对应的订阅条目副本后遍历，避免并发写影响迭代
|
||
entries = list(self._subscriptions.get(event.event_type, []))
|
||
for entry in entries:
|
||
try:
|
||
await asyncio.wait_for(
|
||
entry.subscription.handler.handle(event),
|
||
timeout=self._HANDLER_TIMEOUT,
|
||
)
|
||
except TimeoutError:
|
||
strategy = entry.subscription.failure_strategy
|
||
if strategy in (FailurePolicy.CIRCUIT_BREAK, FailurePolicy.ISOLATE):
|
||
# 按策略触发插件降级/熔断/隔离，降级管理器自身异常不得
|
||
# 拖垮宿主，记录告警后继续分发其他订阅者
|
||
try:
|
||
await self._degradation.onPluginFailed(
|
||
entry.plugin_id,
|
||
f"handler timeout: {self._HANDLER_TIMEOUT}s",
|
||
)
|
||
except Exception as degrade_err:
|
||
await self._logger.warn(
|
||
f"降级管理器调用失败: {degrade_err}",
|
||
plugin_id=entry.plugin_id,
|
||
error=str(degrade_err),
|
||
)
|
||
# handler 超时不得拖垮宿主，记录告警后继续分发其他订阅者
|
||
await self._logger.warn(
|
||
f"事件订阅 handler 超时: event_type={event.event_type}, strategy={strategy}",
|
||
trace_id=event.trace_id,
|
||
plugin_id=entry.plugin_id,
|
||
timeout=self._HANDLER_TIMEOUT,
|
||
)
|
||
except Exception as e:
|
||
strategy = entry.subscription.failure_strategy
|
||
if strategy in (FailurePolicy.CIRCUIT_BREAK, FailurePolicy.ISOLATE):
|
||
# 按策略触发插件降级/熔断/隔离，降级管理器自身异常不得
|
||
# 拖垮宿主，记录告警后继续分发其他订阅者
|
||
try:
|
||
await self._degradation.onPluginFailed(
|
||
entry.plugin_id,
|
||
str(e),
|
||
)
|
||
except Exception as degrade_err:
|
||
await self._logger.warn(
|
||
f"降级管理器调用失败: {degrade_err}",
|
||
plugin_id=entry.plugin_id,
|
||
error=str(degrade_err),
|
||
)
|
||
# 插件 handler 失败不得拖垮宿主，记录日志后继续分发其他订阅者
|
||
await self._logger.warn(
|
||
f"事件订阅 handler 失败: event_type={event.event_type}, strategy={strategy.value}",
|
||
trace_id=event.trace_id,
|
||
plugin_id=entry.plugin_id,
|
||
error=str(e),
|
||
)
|
||
|
||
async def publishOutboxStateChanged(
|
||
self,
|
||
event: OutboxStateChangedEvent,
|
||
) -> None:
|
||
"""发布发件箱状态变更事件（FR-22 / FR-34）。
|
||
|
||
将 ``OutboxStateChangedEvent`` 转换为 ``DomainEvent``（event_type=
|
||
``"OutboxStateChanged"``），复用 ``publish`` 分发给已注册的订阅者
|
||
（如 ``OutboxStateAuditHandler``）。payload 对齐
|
||
``OutboxStateChanged`` 领域事件的序列化格式（``outbox_id`` /
|
||
``old_status`` / ``new_status``），供 ``OutboxStateAuditHandler``
|
||
提取；并额外携带 ``reason`` 与 ``occurred_at`` 供下游消费。
|
||
|
||
handler 异常由 ``publish`` 捕获并记录警告日志，不中断其他订阅者
|
||
（FR-36），故本方法不添加 try/except 包裹 ``publish`` 调用，禁止
|
||
静默吞没异常。
|
||
|
||
参数：
|
||
event: 发件箱状态变更事件 DTO，携带 entry_id / old_state /
|
||
new_state / reason / occurred_at。
|
||
"""
|
||
domain_event = DomainEvent(
|
||
event_id=str(uuid.uuid4()),
|
||
event_type="OutboxStateChanged",
|
||
payload={
|
||
"outbox_id": event.entry_id,
|
||
"old_status": event.old_state,
|
||
"new_status": event.new_state,
|
||
"reason": event.reason,
|
||
"occurred_at": event.occurred_at.isoformat(),
|
||
},
|
||
timestamp=event.occurred_at,
|
||
trace_id=None,
|
||
)
|
||
await self.publish(domain_event)
|
||
|
||
def register(self, plugin_id: str, subscription: EventSubscription) -> None:
|
||
"""注册事件订阅。
|
||
|
||
将订阅条目追加到对应事件类型的列表，并按 ``priority`` 升序排序
|
||
（越小越先执行）。
|
||
|
||
参数：
|
||
plugin_id: 插件 ID。
|
||
subscription: 事件订阅值对象。
|
||
"""
|
||
entry = _SubscriptionEntry(plugin_id=plugin_id, subscription=subscription)
|
||
with self._lock:
|
||
self._subscriptions.setdefault(subscription.event_type, []).append(entry)
|
||
self._subscriptions[subscription.event_type].sort(
|
||
key=lambda e: e.subscription.priority,
|
||
)
|
||
|
||
def unregister(self, plugin_id: str) -> None:
|
||
"""注销插件的所有事件订阅。
|
||
|
||
遍历所有事件类型的订阅列表，移除属于该插件 ID 的订阅条目。
|
||
|
||
参数：
|
||
plugin_id: 插件 ID。
|
||
"""
|
||
with self._lock:
|
||
for event_type in self._subscriptions:
|
||
self._subscriptions[event_type] = [
|
||
e for e in self._subscriptions[event_type] if e.plugin_id != plugin_id
|
||
]
|