""" 进程内事件注册与分发，管理处理器绑定、并发执行及关停收敛。 """ import asyncio import concurrent.futures import random import threading import traceback import uuid from contextvars import ContextVar from dataclasses import dataclass from queue import Empty, PriorityQueue from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Set, Tuple, Type, TypeVar, Union, cast from app.foundation.singleton import Singleton from app.runtime.correlation import get_correlation_id from app.runtime.event.binding import ( EventBindingResolver, EventHandlerBinding, HandlerInstanceResolver, ) from app.runtime.event.contracts import normalize_event_type, validate_event_payload from app.runtime.event.dispatch import EventDispatcher from app.runtime.event.errors import EventErrorNotifier, EventErrorPolicy from app.runtime.event.registry import EventRegistry from app.runtime.log import logger from app.runtime.loop import main_loop_registry from app.runtime.observability import record_metric from app.runtime.rate import ExponentialBackoffRateLimiter from app.runtime.thread import ThreadHelper from app.schemas.event import BaseEventData from app.schemas.types import ChainEventType, EventType if TYPE_CHECKING: from app.runtime.event.snapshot import EventPayloadSnapshot # 事件的默认优先级 DEFAULT_EVENT_PRIORITY = 10 # 最小事件消费者线程数 MIN_EVENT_CONSUMER_THREADS = 1 # 事件队列空闲时的初始超时时间（秒） INITIAL_EVENT_QUEUE_IDLE_TIMEOUT_SECONDS = 1 # 事件队列空闲时的最大超时时间（秒） MAX_EVENT_QUEUE_IDLE_TIMEOUT_SECONDS = 5 _EVENT_STOP_SENTINEL = object() _CURRENT_EVENT_HANDLER_OWNER: ContextVar[object | None] = ContextVar( "current_event_handler_owner", default=None, ) _EventHandler = TypeVar("_EventHandler", bound=Callable[..., Any]) _EventKind = Union[EventType, ChainEventType] _EventRegistration = Union[ _EventKind, List[_EventKind], Type[EventType], Type[ChainEventType], ] @dataclass(slots=True) class _EventAsyncHandle: """ 记录异步广播的取消代理和真实完成信号。 :param handle: 提供给广播调用方的取消代理 :param completion: 真实异步执行完成后才结束的信号，用于关停等待 """ handle: concurrent.futures.Future[Any] completion: concurrent.futures.Future[Any] class Event: """ 事件类，封装事件的基本信息。 """ def __init__(self, event_type: Union[EventType, ChainEventType, str], event_data: Optional[Union[Dict[str, Any], BaseEventData]] = None, priority: Optional[int] = DEFAULT_EVENT_PRIORITY, correlation_id: Optional[str] = None): """ 归一化事件类型并创建携带关联标识的事件对象。 :param event_type: 事件的类型，支持 EventType 或 ChainEventType :param event_data: 可选，事件携带的数据，默认为空字典 :param priority: 可选，事件的优先级，默认为 10 :param correlation_id: 生产事件时固化的请求关联 ID """ event_type = normalize_event_type(event_type) payload_problems = validate_event_payload(event_type, event_data) if payload_problems: logger.warning( "事件 %s payload 与登记契约不一致：%s；当前保留旧 payload 继续投递", getattr(event_type, "value", event_type), "; ".join(payload_problems), ) # 事件ID self.event_id = str(uuid.uuid4()) # 事件类型 self.event_type = event_type # 事件数据 self.event_data = event_data or {} # 事件优先级 self.priority = priority if priority is not None else DEFAULT_EVENT_PRIORITY self.correlation_id = correlation_id or get_correlation_id() def __repr__(self) -> str: """ 重写 __repr__ 方法，用于返回事件的详细信息，包括事件类型、事件ID和优先级。 """ event_kind = Event.get_event_kind(self.event_type) event_name = getattr(self.event_type, "value", self.event_type) return f"<{event_kind}: {event_name}, ID: {self.event_id}, Priority: {self.priority}>" def snapshot(self) -> "EventPayloadSnapshot": """ 返回插件可安全读取的类型化 payload 快照，原始 event_data 保持不变。 """ from app.runtime.event.snapshot import snapshot_event_data return snapshot_event_data(self.event_type, self.event_data) def __lt__(self, other: "Event") -> bool: """ 定义事件对象的比较规则，基于优先级比较 优先级小的事件会被认为“更小”，优先级高的事件将被认为“更大” """ return self.priority < other.priority @staticmethod def get_event_kind(event_type: Union[EventType, ChainEventType, str]) -> str: """ 根据事件类型判断事件是广播事件还是链式事件。 :param event_type: 事件类型，支持 EventType 或 ChainEventType :return: 返回 Broadcast Event 或 Chain Event """ return "Broadcast Event" if isinstance(event_type, EventType) else "Chain Event" class EventManager(metaclass=Singleton): """ EventManager 负责管理和调度广播事件和链式事件，包括订阅、发送和处理事件。 """ def __init__(self) -> None: """ 初始化订阅表、处理队列、解析器和消费者状态。 """ # 动态线程池，用于消费事件 self.__executor = ThreadHelper() # 用于保存启动的事件消费者线程 self.__consumer_threads: List[threading.Thread] = [] # 优先级队列 self.__event_queue: PriorityQueue[Any] = PriorityQueue() # 广播事件的订阅者 self.__broadcast_subscribers: Dict[EventType, Dict[str, Callable[..., Any]]] = {} # 链式事件的订阅者 self.__chain_subscribers: Dict[ChainEventType, Dict[str, tuple[int, Callable[..., Any]]]] = {} # 禁用的事件处理器集合 self.__disabled_handlers: Set[str] = set() # 禁用的事件处理器类集合 self.__disabled_classes: Set[str] = set() # 线程锁 self.__lock = threading.Lock() # 退出事件 self.__event = threading.Event() # 广播处理器由事件总线统一持有，确保插件卸载前可以建立结算屏障。 self.__lifecycle_lock = threading.RLock() self.__lifecycle_state = "new" self.__sync_handles: Dict[object, concurrent.futures.Future[Any]] = {} self.__async_handles: Dict[object, _EventAsyncHandle] = {} # 由上层管理器注册的处理器实例解析器 self.__handler_instance_resolvers: Dict[str, HandlerInstanceResolver] = {} # 由启动组合层注入的错误通知回调 self.__error_notifier: Optional[EventErrorNotifier] = None self.__registry = EventRegistry( lock=self.__lock, broadcast_subscribers=lambda: self.__broadcast_subscribers, chain_subscribers=lambda: self.__chain_subscribers, disabled_handlers=lambda: self.__disabled_handlers, disabled_classes=lambda: self.__disabled_classes, ) self.__binding_resolver = EventBindingResolver( lock=self.__lock, resolvers=lambda: self.__handler_instance_resolvers, ) self.__error_policy = EventErrorPolicy( notifier=lambda: self.__error_notifier, emit_system_error=lambda payload: self.send_event( EventType.SystemError, payload, ), ) self.__dispatcher = EventDispatcher( registry=self.__registry, binding_resolver=self.__binding_resolver, event_factory=Event, error_handler=lambda **kwargs: self.__handle_event_error(**kwargs), async_handle_sink=self.__register_async_handle, sync_handle_sink=self.__register_sync_handle, ) def register_handler_instance_resolver( self, name: str, resolver: HandlerInstanceResolver, ) -> None: """ 注册上层运行时的事件处理器实例解析器。 同名解析器会被替换，避免测试重建单例或热重载后保留旧实例引用。 """ self.__binding_resolver.register(name, resolver) def unregister_handler_instance_resolver(self, name: str) -> None: """ 撤销命名实例解析器；重复撤销不影响其它 owner。 """ self.__binding_resolver.unregister(name) def unresolved_handler_bindings(self) -> tuple[str, ...]: """ 返回未命中显式 resolver 的类处理器诊断清单。 """ return self.__binding_resolver.unresolved_handlers() def set_error_notifier(self, notifier: Optional[EventErrorNotifier]) -> None: """ 设置事件处理异常的外部通知回调。 """ with self.__lock: self.__error_notifier = notifier def reset_error_notifier(self) -> None: """ 撤销当前 lifespan 的事件错误通知回调。 """ self.set_error_notifier(None) def start(self) -> None: """ 开始广播事件处理线程。 """ with self.__lifecycle_lock: if self.__lifecycle_state == "running": return if self.__lifecycle_state == "stopping": logger.warning("事件处理仍在停止，忽略重复启动") return if self.__lifecycle_state == "sealed": logger.warning("事件处理已封口，忽略重复启动") return self.__lifecycle_state = "running" self.__event.set() self.__consumer_threads = [] for _ in range(MIN_EVENT_CONSUMER_THREADS): thread = threading.Thread(target=self.__broadcast_consumer_loop, daemon=True) thread.start() self.__consumer_threads.append(thread) def stop(self) -> None: """ 兼容同步关闭入口，等待同步处理器并请求取消异步处理器。 调用线程无法安全等待主事件循环完成异步清理；需要完整异步收口时应使用 stop_async()。 """ logger.info("正在停止事件处理...") consumer_threads = self.__begin_stop() try: self.__join_consumer_threads(consumer_threads) self.__discard_stop_sentinels() current_owner = _CURRENT_EVENT_HANDLER_OWNER.get() self.__cancel_async_handles(exclude_owner=current_owner) self.__wait_sync_handles(exclude_owner=current_owner) logger.info("事件处理停止完成") except Exception as e: logger.error(f"停止事件处理线程出错：{str(e)} - {traceback.format_exc()}") finally: with self.__lifecycle_lock: self.__lifecycle_state = "stopped" self.__consumer_threads = [] async def stop_async(self) -> None: """ 停止广播消费者，等待同步处理器并取消收口异步处理器。 """ logger.info("正在停止事件处理...") consumer_threads = self.__begin_stop() try: if consumer_threads: await asyncio.to_thread( self.__join_consumer_threads, consumer_threads, ) self.__discard_stop_sentinels() current_owner = _CURRENT_EVENT_HANDLER_OWNER.get() with self.__lifecycle_lock: async_handles = tuple( handle for owner, handle in self.__async_handles.items() if owner is not current_owner ) sync_handles = tuple( handle for owner, handle in self.__sync_handles.items() if owner is not current_owner ) for handle in async_handles: handle.handle.cancel() if async_handles or sync_handles: await asyncio.gather( *( asyncio.shield(asyncio.wrap_future(handle.completion)) for handle in async_handles ), *( asyncio.shield(asyncio.wrap_future(handle)) for handle in sync_handles ), return_exceptions=True, ) logger.info("事件处理停止完成") except Exception as e: logger.error(f"停止事件处理线程出错：{str(e)} - {traceback.format_exc()}") raise with self.__lifecycle_lock: self.__lifecycle_state = "stopped" self.__consumer_threads = [] def __begin_stop(self) -> tuple[threading.Thread, ...]: """ 关闭提交入口并唤醒消费者线程。 """ with self.__lifecycle_lock: self.__lifecycle_state = "stopping" self.__event.clear() consumer_threads = tuple(self.__consumer_threads) if consumer_threads: self.__event_queue.put((float("-inf"), _EVENT_STOP_SENTINEL)) return consumer_threads @staticmethod def __join_consumer_threads(consumer_threads: tuple[threading.Thread, ...]) -> None: """ 在线程池或同步兼容入口中等待事件消费者退出。 """ for consumer_thread in consumer_threads: consumer_thread.join() def __discard_stop_sentinels(self) -> None: """ 清理仅用于唤醒消费者的标记，保留尚未消费的业务事件。 """ pending = [] while True: try: item = self.__event_queue.get_nowait() except Empty: break if item[1] is not _EVENT_STOP_SENTINEL: pending.append(item) self.__event_queue.task_done() for item in pending: self.__event_queue.put(item) def __cancel_async_handles( self, *, exclude_owner: object | None = None, ) -> None: """ 请求取消异步处理器，并避免 handler 关闭事件总线时取消自身。 """ with self.__lifecycle_lock: handles = tuple( handle for owner, handle in self.__async_handles.items() if owner is not exclude_owner ) for handle in handles: handle.handle.cancel() def __wait_sync_handles( self, *, exclude_owner: object | None = None, ) -> None: """ 等待同步处理器完成，并避免 handler 关闭事件总线时等待自身。 """ with self.__lifecycle_lock: handles = tuple( handle for owner, handle in self.__sync_handles.items() if owner is not exclude_owner ) if handles: concurrent.futures.wait(handles) def __register_sync_handle( self, callback: Callable[..., Any], args: tuple[Any, ...], ) -> bool: """ 在同一生命周期临界区提交并登记同步广播处理器。 """ with self.__lifecycle_lock: if self.__lifecycle_state != "running": logger.warning( "事件处理处于 %s 状态，拒绝同步广播处理器", self.__lifecycle_state, ) return False try: owner = object() completion: concurrent.futures.Future[Any] = concurrent.futures.Future() self.__sync_handles[owner] = completion def _tracked_sync() -> Any: """ 在同步 handler 调用栈中发布当前事件 owner。 """ context_token = _CURRENT_EVENT_HANDLER_OWNER.set(owner) try: return callback(*args) finally: _CURRENT_EVENT_HANDLER_OWNER.reset(context_token) handle = self.__executor.submit(_tracked_sync) except RuntimeError: self.__sync_handles.pop(owner, None) logger.warning("同步事件处理器无法投递，线程池已停止") return False def _complete_sync_handle( completed: concurrent.futures.Future[Any], ) -> None: """ 真实线程结束后先清理 owner，再向结算句柄发布结果，避免关闭等待者先被唤醒。 :param completed: 已经完成或取消的原始线程任务句柄 """ self.__remove_sync_handle(owner) if completed.cancelled(): completion.cancel() else: error = completed.exception() if error is not None: completion.set_exception(error) else: completion.set_result(completed.result()) handle.add_done_callback(_complete_sync_handle) return True def __remove_sync_handle(self, owner: object) -> None: """ 同步处理器完成后移除其 owner 句柄。 """ with self.__lifecycle_lock: self.__sync_handles.pop(owner, None) def __register_async_handle( self, coroutine: Any, ) -> bool: """ 在同一生命周期临界区提交并登记异步广播处理器。 """ with self.__lifecycle_lock: if self.__lifecycle_state != "running": coroutine.close() logger.warning( "事件处理处于 %s 状态，拒绝异步广播处理器", self.__lifecycle_state, ) return False completion: concurrent.futures.Future[Any] = concurrent.futures.Future() started = threading.Event() owner = object() async def _tracked() -> None: """ 在所属上下文中执行处理器，真实结束后先清理 owner，再发布完成信号。 """ started.set() context_token = _CURRENT_EVENT_HANDLER_OWNER.set(owner) try: result = await coroutine except asyncio.CancelledError: self.__remove_async_handle(owner) if not completion.done(): completion.cancel() except Exception as err: self.__remove_async_handle(owner) if not completion.done(): completion.set_exception(err) else: self.__remove_async_handle(owner) if not completion.done(): completion.set_result(result) finally: _CURRENT_EVENT_HANDLER_OWNER.reset(context_token) tracked = _tracked() try: handle = asyncio.run_coroutine_threadsafe( tracked, main_loop_registry.require(), ) except RuntimeError: tracked.close() coroutine.close() logger.warning("异步事件处理器无法投递，事件循环已停止") return False self.__async_handles[owner] = _EventAsyncHandle( handle=handle, completion=completion, ) def _complete_unstarted_submission( submitted: concurrent.futures.Future[Any], ) -> None: """ 关闭尚未启动就被取消的协程，先移除 owner，再发布取消完成信号。 :param submitted: 跨线程提交到所属事件循环的代理句柄 """ if submitted.cancelled() and not started.is_set(): coroutine.close() self.__remove_async_handle(owner) completion.cancel() handle.add_done_callback(_complete_unstarted_submission) return True def __remove_async_handle(self, owner: object) -> None: """ 异步处理器完成后移除其 owner 句柄。 """ with self.__lifecycle_lock: self.__async_handles.pop(owner, None) async def drain_async( self, timeout: Optional[float] = None, *, seal: bool = False, ) -> bool: """ 等待已接纳的广播事件及其派生事件自然结算。 seal=True 会在确认稳定空闲的同一临界区关闭后续广播提交，供插件卸载前 建立不可穿透的投递屏障；超时或停止交错时保持原提交状态并返回 False。 广播处理器调用栈内无法等待自身完成，因此会立即返回 False，且不执行封口。 """ if _CURRENT_EVENT_HANDLER_OWNER.get() is not None: logger.warning("事件处理器内部不能建立事件投递屏障") return False loop = asyncio.get_running_loop() deadline = None if timeout is None else loop.time() + max(timeout, 0) while True: with self.__lifecycle_lock: if self.__lifecycle_state not in {"running", "sealed"}: return False with self.__event_queue.all_tasks_done: queue_idle = self.__event_queue.unfinished_tasks == 0 handlers_idle = not self.__sync_handles and not self.__async_handles if queue_idle and handlers_idle: if seal: self.__lifecycle_state = "sealed" return True if deadline is not None: remaining = deadline - loop.time() if remaining <= 0: return False await asyncio.sleep(min(0.01, remaining)) else: await asyncio.sleep(0.01) def check(self, etype: Union[EventType, ChainEventType]) -> bool: """ 检查是否有启用的事件处理器可以响应某个事件类型。 :param etype: 事件类型 (EventType 或 ChainEventType) :return: 返回是否存在可用的处理器 """ return self.__registry.check(etype) def send_event(self, etype: Union[EventType, ChainEventType], data: Optional[Union[Dict[str, Any], BaseEventData]] = None, priority: Optional[int] = DEFAULT_EVENT_PRIORITY) -> Optional[Event]: """ 发送事件，根据事件类型决定是广播事件还是链式事件。 :param etype: 事件类型 (EventType 或 ChainEventType) :param data: 可选，事件数据 :param priority: 广播事件的优先级，默认为 10 :return: 如果是链式事件，返回处理后的事件数据；否则返回 None """ event = Event(etype, data, priority) if isinstance(etype, EventType): self.__trigger_broadcast_event(event) return None elif isinstance(etype, ChainEventType): return self.__trigger_chain_event(event) else: logger.error(f"Unknown event type: {etype}") return None def send_event_strict( self, etype: EventType, data: Optional[Union[dict[str, object], BaseEventData]] = None, priority: Optional[int] = DEFAULT_EVENT_PRIORITY, ) -> Event: """ 同步等待全部广播处理器完成，任一失败时阻止 durable 消息结算。 """ event = Event(etype, data, priority) with self.__lifecycle_lock: if self.__lifecycle_state != "running": raise RuntimeError(f"事件处理处于 {self.__lifecycle_state} 状态") self.__dispatcher.dispatch_broadcast_strict( event, self.__wait_strict_async_handler, ) return event async def async_send_event_strict( self, etype: EventType, data: Optional[Union[dict[str, object], BaseEventData]] = None, priority: Optional[int] = DEFAULT_EVENT_PRIORITY, ) -> Event: """ 异步发送广播事件并等待全部处理器完成，保证配置写入返回时已完成重载。 """ event = Event(etype, data, priority) with self.__lifecycle_lock: if self.__lifecycle_state != "running": raise RuntimeError(f"事件处理处于 {self.__lifecycle_state} 状态") await self.__dispatcher.dispatch_broadcast_async_strict(event) return event @staticmethod def __wait_strict_async_handler(coroutine: Any) -> Any: """ 在主事件循环等待异步处理器，禁止循环线程同步等待自身。 """ loop = main_loop_registry.require() try: running_loop = asyncio.get_running_loop() except RuntimeError: running_loop = None if running_loop is loop: coroutine.close() raise RuntimeError("主事件循环线程不能同步等待 durable 事件处理器") return asyncio.run_coroutine_threadsafe(coroutine, loop).result() async def async_send_event(self, etype: Union[EventType, ChainEventType], data: Optional[Union[Dict[str, Any], BaseEventData]] = None, priority: Optional[int] = DEFAULT_EVENT_PRIORITY) -> Optional[Event]: """ 异步发送事件，根据事件类型决定是广播事件还是链式事件。 :param etype: 事件类型 (EventType 或 ChainEventType) :param data: 可选，事件数据 :param priority: 广播事件的优先级，默认为 10 :return: 如果是链式事件，返回处理后的事件数据；否则返回 None """ event = Event(etype, data, priority) if isinstance(etype, EventType): self.__trigger_broadcast_event(event) return None elif isinstance(etype, ChainEventType): return await self.__trigger_chain_event_async(event) else: logger.error(f"Unknown event type: {etype}") return None def add_event_listener(self, event_type: Union[EventType, ChainEventType], handler: Callable[..., Any], priority: Optional[int] = DEFAULT_EVENT_PRIORITY) -> None: """ 注册事件处理器，将处理器添加到对应的事件订阅列表中。 :param event_type: 事件类型 (EventType 或 ChainEventType) :param handler: 处理器 :param priority: 可选，链式事件的优先级，默认为 10；广播事件不需要优先级 """ self.__registry.add(event_type, handler, priority or DEFAULT_EVENT_PRIORITY) def remove_event_listener(self, event_type: Union[EventType, ChainEventType], handler: Callable[..., Any]) -> None: """ 移除事件处理器，将处理器从对应事件的订阅列表中删除。 :param event_type: 事件类型 (EventType 或 ChainEventType) :param handler: 要移除的处理器 """ self.__registry.remove(event_type, handler) def disable_event_handler(self, target: Union[Callable[..., Any], type]) -> None: """ 禁用指定的事件处理器或事件处理器类。 :param target: 处理器函数或类 """ self.__registry.disable(target) def enable_event_handler(self, target: Union[Callable[..., Any], type]) -> None: """ 启用指定的事件处理器或事件处理器类。 :param target: 处理器函数或类 """ self.__registry.enable(target) def visualize_handlers(self) -> List[Dict[str, Any]]: """ 可视化所有事件处理器，包括是否被禁用的状态。 :return: 处理器列表，包含事件类型、处理器标识符、优先级（如果有）和状态 """ return self.__registry.visualize() @classmethod def __get_handler_identifier(cls, target: Union[Callable[..., Any], type]) -> Optional[str]: """ 获取处理器或处理器类的唯一标识符，包括模块名和类名/方法名。 :param target: 处理器函数或类 :return: 唯一标识符 """ return EventRegistry.handler_identifier(target) @classmethod def __get_class_from_callable(cls, handler: Callable[..., Any]) -> Optional[str]: """ 获取可调用对象所属类的唯一标识符。 :param handler: 可调用对象（函数、方法等） :return: 类的唯一标识符 """ return EventRegistry.handler_class_identifier(handler) def __is_handler_enabled(self, handler: Callable[..., Any]) -> bool: """ 检查处理器是否已启用（没有被禁用）。 :param handler: 处理器函数 :return: 如果处理器启用则返回 True，否则返回 False """ return self.__registry.is_handler_enabled(handler) def __trigger_chain_event(self, event: Event) -> Optional[Event]: """ 触发链式事件，按顺序调用订阅的处理器，并记录处理耗时。 """ logger.debug(f"Triggering synchronous chain event: {event}") dispatch = self.__dispatch_chain_event(event) return event if dispatch else None async def __trigger_chain_event_async(self, event: Event) -> Optional[Event]: """ 异步触发链式事件，按顺序调用订阅的处理器，并记录处理耗时。 """ logger.debug(f"Triggering asynchronous chain event: {event}") dispatch = await self.__dispatch_chain_event_async(event) return event if dispatch else None def __trigger_broadcast_event(self, event: Event) -> None: """ 触发广播事件，将事件插入到优先级队列中。 :param event: 要处理的事件对象 """ logger.debug(f"Triggering broadcast event: {event}") with self.__lifecycle_lock: if self.__lifecycle_state in {"sealed", "stopping", "stopped"}: logger.warning( "事件处理处于 %s 状态，拒绝广播事件 %s", self.__lifecycle_state, event.event_type, ) return None self.__event_queue.put((event.priority, event)) record_metric( "event.queue.depth", 1, delivery="broadcast", ) def __dispatch_chain_event(self, event: Event) -> bool: """ 同步方式调度链式事件，按优先级顺序逐个调用事件处理器，并记录每个处理器的处理时间。 :param event: 要调度的事件对象 """ return self.__dispatcher.dispatch_chain(event) async def __dispatch_chain_event_async(self, event: Event) -> bool: """ 异步方式调度链式事件，按优先级顺序逐个调用事件处理器，并记录每个处理器的处理时间。 :param event: 要调度的事件对象 """ return await self.__dispatcher.async_dispatch_chain(event) def __dispatch_broadcast_event(self, event: Event) -> None: """ 异步方式调度广播事件，通过线程池逐个调用事件处理器。 :param event: 要调度的事件对象 """ self.__dispatcher.dispatch_broadcast(event) @classmethod def __should_dispatch_to_target_plugin( cls, handler: Callable[..., Any], handler_identifier: str, target_plugin_id: str, ) -> bool: """ 限定插件输入事件只投递给目标插件，避免自由文本被其他插件观察到。 """ return EventDispatcher.should_dispatch_to_target_plugin( handler, handler_identifier, target_plugin_id, ) def __safe_invoke_handler(self, handler: Callable[..., Any], event: Event) -> None: """ 调用处理器，处理链式或广播事件。 :param handler: 处理器 :param event: 事件对象 """ self.__dispatcher.safe_invoke_sync(handler, event) async def __safe_invoke_handler_async(self, handler: Callable[..., Any], event: Event) -> None: """ 异步调用处理器，处理链式事件。 :param handler: 处理器 :param event: 事件对象 """ await self.__dispatcher.safe_invoke_async(handler, event) def __invoke_handler_by_type_sync(self, handler: Callable[..., Any], event: Event) -> None: """ 同步方式根据处理器类型调用相应的方法。 :param handler: 处理器 :param event: 要处理的事件对象 """ self.__dispatcher.invoke_sync(handler, event) async def __invoke_handler_by_type_async(self, handler: Callable[..., Any], event: Event) -> None: """ 异步方式根据处理器类型调用相应的方法。 :param handler: 处理器 :param event: 要处理的事件对象 """ await self.__dispatcher.invoke_async(handler, event) @staticmethod def __parse_handler_names(handler: Callable[..., Any]) -> Tuple[str, str]: """ 解析处理器的类名和方法名。 :param handler: 处理器 :return: 类名与方法名的二元组，用于查找已绑定处理器 """ return EventBindingResolver.parse_handler_names(handler) @staticmethod def __get_handler_owner_class(handler: Callable[..., Any]) -> Optional[Type[Any]]: """ 从处理器对象本身解析声明它的类，不按命名约定动态导入模块。 """ return EventBindingResolver.owner_class(handler) def __resolve_handler( self, handler: Callable[..., Any], ) -> Optional[Tuple[Callable[..., Any], EventHandlerBinding, str, str]]: """ 将装饰阶段保存的函数解析为当前运行实例上的可调用方法。 """ return self.__binding_resolver.resolve(handler) def __broadcast_consumer_loop(self) -> None: """ 持续从队列中提取事件的后台广播消费者线程。 """ jitter_factor = 0.1 rate_limiter = ExponentialBackoffRateLimiter(base_wait=INITIAL_EVENT_QUEUE_IDLE_TIMEOUT_SECONDS, max_wait=MAX_EVENT_QUEUE_IDLE_TIMEOUT_SECONDS, backoff_factor=2.0, source="BroadcastConsumer", enable_logging=False) while self.__event.is_set(): try: priority, event = self.__event_queue.get(timeout=rate_limiter.current_wait) try: if event is _EVENT_STOP_SENTINEL: break record_metric( "event.queue.depth", -1, delivery="broadcast", ) rate_limiter.reset() self.__dispatch_broadcast_event(event) finally: self.__event_queue.task_done() except Empty: rate_limiter.current_wait = rate_limiter.current_wait * random.uniform(1, 1 + jitter_factor) rate_limiter.trigger_limit() @staticmethod def __log_event_lifecycle(event: Event, stage: str) -> None: """ 记录事件的生命周期日志。 """ logger.debug(f"{stage} - {event}") def __handle_event_error(self, event: Event, module_name: str, class_name: str, method_name: str, e: Exception) -> None: """ 全局错误处理器，用于处理事件处理中的异常。 """ self.__error_policy.handle( event=event, module_name=module_name, class_name=class_name, method_name=method_name, error=e, ) def register( self, etype: _EventRegistration, priority: Optional[int] = DEFAULT_EVENT_PRIORITY, ) -> Callable[[_EventHandler], _EventHandler]: """ 事件注册装饰器，用于将函数注册为事件的处理器。 :param etype: 要登记的事件范围，支持以下形式 - 单个事件类型成员 (如 EventType.MetadataScrape, ChainEventType.PluginAction) - 事件类型类 (EventType, ChainEventType) - 或事件类型成员的列表 :param priority: 可选，链式事件的优先级，默认为 DEFAULT_EVENT_PRIORITY """ def decorator(f: _EventHandler) -> _EventHandler: """ 登记处理器的事件类型和优先级，返回原函数以保持直接调用行为。 """ if isinstance(etype, list): event_list = etype else: event_list = [cast(_EventKind, etype)] for event in event_list: if isinstance(event, (EventType, ChainEventType)): self.add_event_listener(event, f, priority) elif event is EventType or event is ChainEventType: for et in event.__members__.values(): self.add_event_listener(et, f, priority) else: raise ValueError(f"无效的事件类型: {event}") return f return decorator # 模块热重载时类对象会重新创建，但插件和 SDK 可能仍持有旧全局实例。把旧实例登记到 # 新 EventManager 类的单例键，确保所有公开入口继续共享同一个事件总线。 _existing_eventmanager = globals().get("eventmanager") if _existing_eventmanager is not None: Singleton._instances[(EventManager, (), frozenset())] = _existing_eventmanager eventmanager = EventManager()