events — 插件事件总线

订阅主机事件并接收 `ide.event.emit` 入站帧。

self.events 提供事件订阅能力,用于订阅主机发布的事件。能否订阅某主题取决于该主题对应所需的权限(例如 runtime.* 主题需要 runtime.inspect)。订阅由插件会话持有,插件卸载时总线会自动全部释放。

事件从订阅到投递的完整链路:

订阅

subscribe(topic, handler, *, filter?, delivery?, debounce_ms?, callback?)

订阅 topic 事件。返回一个 Subscription 对象。

subscription = self.events.subscribe(
    "serial.data",
    self._on_serial_data,
    delivery="latest",
)

参数:

  • topic (str) — 订阅的主题
  • handler (callable) — 事件处理器,同步或异步均可
  • filter (dict, 可选) — 事件过滤器,仅投递匹配字段的事件
  • delivery (str, 可选) — 投递模式:every / latest / batch / debounce / throttle
  • debounce_ms (int, 可选) — 防抖/节流/批处理窗口(毫秒)
  • callback (callable, 可选) — 订阅结果的回调
投递模式说明
every每个事件都投递(默认)
latest窗口内只保留最新事件
batch窗口内事件批量合并投递
debounce停止触发 debounce_ms 后才投递
throttledebounce_ms 至少投递一次

当主机拒绝订阅(主题未知或权限被拒)时,订阅会在本地被丢弃,可通过 callback(error=...) 感知失败。

unsubscribe(subscription_id)

按订阅 ID 注销订阅。

Subscription 对象

subscribe 返回的 Subscription 提供:

成员说明
.id订阅唯一 ID
.topic订阅的主题
.disposedbool,是否已释放
.dispose()释放该订阅,停止投递
subscription.dispose()
print(subscription.disposed)  # True

处理器抛出的异常不会破坏 Bridge 事件循环,也不会影响其他订阅;异常会被捕获、记录并通过 report_error 上报。

使用示例

class MyPlugin(UiPlugin):
    def on_start(self):
        self.sub = self.events.subscribe(
            "serial.data",
            self._on_data,
            delivery="batch",
            debounce_ms=50,
        )

    def on_dispose(self):
        self.sub.dispose()

    async def _on_data(self, event):
        payload = event.get("data", {})
        print("收到数据:", payload)

On this page