event_router API

完整接口目录 · 插件运行时用法 · 工具包下载

模块说明

Local routing for durable Event deliveries; no subscriptions or automatic ACK.

EventHandler

EventHandler = Callable[['Event'], Awaitable[None]]

EventRegistration

class EventRegistration

SDK 说明：

Unregister exactly one registration; repeated calls return False.

EventRegistration.__init__

def __init__(self, router: 'EventRouter', registration: _Registration)

EventRegistration.unregister

def unregister(self) -> bool

EventRegistration.__call__

def __call__(self) -> bool

EventRouter

class EventRouter

SDK 说明：

Explicit, serial dispatch by Event.kind, in global registration order.

'*' matches any delivered kind locally; it does not grant Host subscriptions.
Handlers get the original Event. Failures/cancellation propagate immediately,
stopping remaining handlers without ACK or retry. Call from Plugin.on_event.

EventRouter.__init__

def __init__(self, *, max_handlers: int=256)

EventRouter.register

def register(self, kind: str, handler: EventHandler) -> EventRegistration

SDK 说明：

Add one async handler; duplicates are independent registrations.

EventRouter.on

def on(self, kind: str) -> Callable[[EventHandler], EventHandler]

SDK 说明：

Decorator that preserves the handler; remove it with off(kind, handler).

EventRouter.off

def off(self, kind: str, handler: EventHandler | None=None) -> int

SDK 说明：

Remove matching registrations and return their count.

Omitting handler removes all handlers for that exact kind. '*' selects
wildcard registrations, not registrations for every event kind.
An already running handler finishes; removed pending handlers are skipped.

EventRouter.dispatch

async def dispatch(self, event: 'Event') -> bool

SDK 说明：

True after at least one handler finishes; unmatched delivery is False.

Registrations added during dispatch start on the next delivery. Concurrent
or recursive dispatch raises instead of interleaving event side effects.
Worker normally serializes calls; this helper starts no tasks or listeners.

EventRouter.feed

async def feed(self, event: 'Event') -> bool

SDK 说明：

Alias for explicit dispatch; never consume another event source.