iii 自定义触发器类型实战：从绑定既有 Trigger 到发布你自己的事件源

iii 自定义触发器类型实战从绑定既有 Trigger 到发布你自己的事件源【免费下载链接】iiiEffortlessly compose, extend, and observe every service in real-time for the first time ever.项目地址: https://gitcode.com/GitHub_Trending/mo/iii导读本文是 iii 框架中触发器Triggers机制的完整实战指南围绕「编写触发器」的两种角色展开作为消费者将 worker 函数绑定到http、cron、queue、state等既有触发器类型作为发布者在 worker 中声明全新的触发器类型让其他 worker 的函数响应你产生的事件。读完本文你将掌握worker.registerTrigger/worker.registerTriggerType/TriggerHandler回调 / 元数据与 JSON Schema 附加 / 运行时注销 /worker.trigger事件分发这一整套 API并能参考仓库中的 SDK 源码与集成测试验证每一步行为。理解「编写触发器」的两面消费者与发布者在 iii 中一个 worker 使用触发器有两种方式对应两个完全不同的方向消费consume绝大多数情况下你通过worker.registerTrigger(...)把 worker 的函数绑定到其他 worker 已经发布的触发器类型上例如http来自 iii-http把函数暴露成端点、cron来自 iii-cron按计划执行函数、队列消息来自 iii-queue每条消息触发一次函数、state变化来自 iii-state响应数据变化以及系统中任何其他事件源。发布publish较少见但同样重要——在你的 worker 中注册一个新的触发器类型让其他 worker 能把它们的函数绑定到你 worker 产生的事件上一个 HTTP 请求、一次 webhook 命中、一个文件变更、一次数据库更新。本文的正文主体聚焦后者编写自己的触发器类型如果你想使用既有触发器调用侧的完整机制worker.trigger/iii trigger直接调用、TriggerAction变体、用条件门控、同一函数多绑定请参见 Using iii / Triggers。消费者侧将函数绑定到既有触发器类型绑定一个 worker 函数到触发器类型使用worker.registerTrigger({ type, function_id, config })。核心约束是发布该触发器类型的 worker 必须处于连接状态否则注册失败。config的形状由每个触发器类型自己定义记录在各发布 worker 的 Worker Docs 中。以把math::add函数暴露为POST /math/add端点为例三种语言 SDK 写法如下worker.registerTrigger({ type: http, function_id: math::add, config: { api_path: /math/add, http_method: POST }, });worker.register_trigger({ type: http, function_id: math::add, config: {api_path: /math/add, http_method: POST}, })use iii_sdk::RegisterTriggerInput; use serde_json::json; worker.register_trigger(RegisterTriggerInput { trigger_type: http.into(), function_id: math::add.into(), config: json!({ api_path: /math/add, http_method: POST }), metadata: None, })?;从 SDK 源码看Node 端的registerTriggeriii.ts会为每次绑定生成一个 UUID 作为触发器实例id向引擎发送RegisterTrigger消息并返回一个带unregister()方法的Trigger句柄用于后续解绑。这里还有一个容易被忽略的命名空间细节注册时若未显式指定namespaceSDK 会用worker 自身的命名空间填充而不是引擎默认的default——因为触发器指向的函数注册在 worker 命名空间下若触发器落在default命名空间会触发却解析不到任何函数。需要特别注意的是绑定在引擎侧是宽容的根据 engine-functions 的 SKILL 文档即使type的提供者未连接或config键写错registerTrigger在引擎层也可能成功——但绑定永远不会被触发。因此在写绑定代码时应先通过engine::triggers::list确认提供者在线并严格照抄类型 schema 中的config键名。为绑定附加 metadata每个触发器绑定都可以在注册时附带一个可选的metadataJSON 对象由消费方设置。引擎原样存储它并在两个位置暴露发布方 worker 的TriggerHandler.registerTrigger(config)回调中它以config.metadata的形式出现发布方可以据此响应消费者提供的标签优先级提示、审计标签、发布方自己记账用的路由键。engine::triggers::list在每个TriggerInfo上返回它控制台和任何做发现的 worker 都能读到。worker.registerTrigger({ type: http, function_id: math::add, config: { api_path: /math/add, http_method: POST }, metadata: { team: platform, env: staging }, });worker.register_trigger({ type: http, function_id: math::add, config: {api_path: /math/add, http_method: POST}, metadata: {team: platform, env: staging}, })use iii_sdk::RegisterTriggerInput; use serde_json::json; worker.register_trigger(RegisterTriggerInput { trigger_type: http.into(), function_id: math::add.into(), config: json!({ api_path: /math/add, http_method: POST }), metadata: Some(json!({ team: platform, env: staging })), })?;不要把触发器 metadata 与触发器类型的 schema 混为一谈这是两种完全不同的东西Metadata由消费者在每次绑定时设置是引擎原样存储的自由标签袋服务于发布方的记账与发现。例如绑定http时附带metadata: { team: platform, env: staging, on_call: alice }发布 worker 可以记录团队信息engine::triggers::list也会在请求时把它暴露出来。Schemas由发布者在声明触发器类型时设置描述消费者将交互的 JSON 形状。例如 iii-http 发布的http类型声明了绑定时消费者传入的config形状{ api_path, http_method }以及绑定函数在每次请求时收到的调用载荷{ method, headers, query_params, body }。另外注意触发器类型本身没有 metadata 字段metadata 是按绑定附加的而不是按类型附加的。在 SDK 的TriggerConfig类型定义中也可以印证这一点Node SDK triggers.tsid触发器实例 ID、function_id、config、可选的metadata、以及解析后的namespace——后者要求发布方在之后调用trigger()时必须原样透传。发布者侧声明一个触发器类型当你的 worker 要成为发布者时目标是把其他 worker 注册的函数绑定到你 worker 观察到的事件上HTTP 请求、webhook、文件变更、数据库更新。触发器类型的两个组成部分一个触发器类型是两样东西捆绑在一起一个字符串id消费者绑定时引用的类型标识例如type: mini-http。一张按绑定维护的路由表这张表由你的 worker 在进程内维护。引擎的注册表会规范地记录每个绑定这就是engine::triggers::list返回的内容但引擎不会基于它做分发。引擎只是把网络上任何消费者 worker 的每一次 bind/unbind 作为回调转发给你的 worker由你的 worker 决定如何处理每个绑定。你在启动时用worker.registerTriggerType({ id, description }, handler)声明一次触发器类型。你实现的TriggerHandler接口暴露两个回调每当消费者绑定或解绑时引擎会在你的发布 worker 上调用它们registerTrigger(config)任何消费者 worker 把函数绑定到你的触发器类型时执行。config携带触发器实例的id、消费者的function_id以及与你类型接受的形状匹配的消费者config。把它存进你的路由表。unregisterTrigger(config)解绑时执行从路由表中移除该项。这两个回调的 SDK 定义可以在仓库中直接核对Node SDK 的TriggerHandler、Rust SDK 的TriggerHandlertraitasync fn register_trigger(self, config: TriggerConfig) - Result(), Error与unregister_trigger、Python SDK 的TriggerHandlerABC。三者签名完全对齐。触发器类型可以在运行期的任何时候拆除使用worker.unregisterTriggerType(...)Python 和 Rust 中为worker.unregister_trigger_type(...)。示例从零实现一个迷你iii-http下面这个例子勾勒了真实http触发器类型发布者iii-http的迷你版。发布 worker 需要声明一个 HTTP 形状的触发器类型mini-http维护一张bindings映射{ trigger id → function_id, methodpath }随消费者绑定/解绑而更新在收到 HTTP 请求时查表分发触发绑定函数的部分见下文「分发事件到绑定函数」。Node / TypeScriptimport { registerWorker } from iii-sdk; import type { TriggerConfig, TriggerHandler } from iii-sdk; const url process.env.III_URL; if (!url) throw new Error(III_URL must be set); const worker registerWorker(url); type MiniHttpConfig { api_path: string; // leading slash, e.g. /orders http_method?: GET | POST | PUT | DELETE; }; const bindings new Mapstring, TriggerConfigMiniHttpConfig(); const httpHandler: TriggerHandlerMiniHttpConfig { async registerTrigger(config) { bindings.set(config.id, config); }, async unregisterTrigger(config) { bindings.delete(config.id); }, }; worker.registerTriggerType( { id: mini-http, description: Routes HTTP requests to bound functions }, httpHandler, );Pythonimport os from iii import ( InitOptions, RegisterTriggerTypeInput, TriggerConfig, TriggerHandler, register_worker, ) worker register_worker( os.environ.get(III_URL), InitOptions(worker_namemini-http-worker), ) bindings: dict[str, TriggerConfig] {} class HttpHandler(TriggerHandler): async def register_trigger(self, config: TriggerConfig) - None: bindings[config.id] config async def unregister_trigger(self, config: TriggerConfig) - None: bindings.pop(config.id, None) worker.register_trigger_type( RegisterTriggerTypeInput( idmini-http, descriptionRoutes HTTP requests to bound functions, ), HttpHandler(), )Rustuse std::collections::HashMap; use std::sync::{Arc, Mutex}; use iii_sdk::{ InitOptions, RegisterTriggerType, TriggerConfig, TriggerHandler, register_worker, }; let url std::env::var(III_URL).expect(III_URL must be set); let worker register_worker(url, InitOptions::default()); #[derive(Default)] struct HttpHandler { bindings: ArcMutexHashMapString, TriggerConfig, } #[async_trait::async_trait] impl TriggerHandler for HttpHandler { async fn register_trigger(self, config: TriggerConfig) - Result(), iii_sdk::IIIError { self.bindings.lock().unwrap().insert(config.id.clone(), config); Ok(()) } async fn unregister_trigger(self, config: TriggerConfig) - Result(), iii_sdk::IIIError { self.bindings.lock().unwrap().remove(config.id); Ok(()) } } worker.register_trigger_type( RegisterTriggerType::new( mini-http, Routes HTTP requests to bound functions, HttpHandler::default(), ), );从 Node SDK 源码iii.ts可以看到registerTriggerType返回一个TriggerTypeRef它把「注册函数 绑定触发器」封装成两个类型安全的便捷方法registerTrigger(functionId, config, metadata?)与registerFunction(functionId, handler, config, metadata?)——后者一次完成函数注册与触发器绑定二者都会自动把触发器的namespace默认成 worker 自身的命名空间避免函数在 worker 命名空间、触发器在default命名空间导致永远无法解析的问题。为触发器类型附加 schema一个触发器类型可以携带两个可选的 JSON Schema 来描述它的载荷trigger_request_format消费者在把函数绑定到你的触发器类型时传给worker.registerTrigger(...)的按绑定config的 schema。call_request_format触发器触发时你的 worker 投递给绑定函数的载荷的 schema。两者都会馈送给 iii 控制台、Agent 可读的 skills以及engine::trigger-types::list的输出让消费者知道该传什么、会收到什么。需要强调的限制运行时校验目前尚未支持。附加的 schema 仅是信息性的——引擎不会拒绝不符合它们的config值或调用载荷。请把 schema 当作给消费者、Agent 和控制台的契约文档与函数 request/response schema 的注意事项相同参见 functions 文档。各 SDK 以各自惯用的方式接受这两个字段SDK传入方式Node / Browser在trigger_request_format/call_request_format上直接放原生 JSON Schema 对象Zod 4 schema 用z.toJSONSchema(...)转换Python一个 Pydantic 模型类自动转换或原生 dict放在RegisterTriggerTypeInput的对应字段上RustRegisterTriggerType的 builder 方法.trigger_request_format::T()和.call_request_format::T()其中T: schemars::JsonSchemaPython SDK 的TriggerTypeRef文档triggers.py给出了一个带 schema 的完整示例register_trigger_type传入trigger_request_formatWebhookTriggerConfig与call_request_formatWebhookCallRequest之后ref.register_trigger(my::handler, WebhookTriggerConfig(url/hook))会在序列化 Pydantic 模型时自动校验配置。注销一个触发器类型当触发器类型路由的工作不再需要时可以在运行期把它拆除。当 worker 断开连接时它发布的所有触发器类型会被自动移除引擎也会停止路由依赖它们的事件——所以这一步只在「worker 保持连接、但想主动丢弃某个类型」时才需要。在registerTriggerType之后的任何时刻都可以调用例如底层资源进入维护模式、特性开关关闭了该表面、或者你想在不重启的情况下把类型轮换到新 schema。延续mini-http示例这里 worker 因配置停用了 HTTP 监听器而丢弃mini-http// e.g. config reload disabled the HTTP listener; stop accepting new bindings // while the worker keeps serving other trigger types. worker.unregisterTriggerType({ id: mini-http, description: Routes HTTP requests to bound functions, });# e.g. config reload disabled the HTTP listener; stop accepting new bindings # while the worker keeps serving other trigger types. worker.unregister_trigger_type( {id: mini-http, description: Routes HTTP requests to bound functions} )// e.g. config reload disabled the HTTP listener; stop accepting new bindings // while the worker keeps serving other trigger types. worker.unregister_trigger_type(mini-http);三个 SDK 的签名存在细微差异iii.ts 中 Node 的unregisterTriggerType会同时发送UnregisterTriggerType消息并从本地triggerTypes表中删除NoderegisterTriggerType返回的TriggerTypeRef还带一个.unregister()快捷方式它委托给worker.unregisterTriggerType(...)。PythonTriggerTypeRef只暴露register_trigger和register_function拆除类型本身必须调用worker.unregister_trigger_type(...)。Rust只接受id字符串Node 和 Python 接受完整输入对象但只有id字段用于标识被拆除的类型。分发事件到绑定函数iii 没有专门的 fire API。当底层事件源投递事件时一个入站 HTTP 请求、一次 cron tick、一次 webhook 命中你的发布 worker 在registerTrigger回调构建的bindings表中查出对应条目然后通过worker.trigger(...)调用每个匹配的函数。延续上面的mini-http示例// Inside the workers HTTP listener, after matching methodpath to an // entry in the bindings map from the declare-trigger-type example: const binding bindings.get(matchedTriggerId); await worker.trigger({ function_id: binding.function_id, payload: { method, headers, body }, });# Inside the workers HTTP listener, after matching methodpath to an # entry in the bindings dict from the declare-trigger-type example: binding bindings[matched_trigger_id] worker.trigger({ function_id: binding.function_id, payload: {method: method, headers: headers, body: body}, })use iii_sdk::TriggerRequest; use serde_json::json; // Inside the workers HTTP listener, after matching methodpath to an // entry in the handlers bindings map from the declare-trigger-type example: let binding handler.bindings.lock().unwrap().get(matched_trigger_id).cloned(); if let Some(binding) binding { worker .trigger(TriggerRequest { function_id: binding.function_id.clone(), payload: json!({ method: method, headers: headers, body: body }), action: None, timeout_ms: None, }) .await?; }注意 Rust 的TriggerRequest还暴露了action与timeout_ms字段前者可指定TriggerAction变体Void、Enqueue等后者控制调用的超时上限。在每次分发的事件上引擎会评估消费者的config和可选的condition_function_id然后把匹配的调用路由到绑定函数并把结果返回给调用方。用集成测试验证生命周期行为仓库中的集成测试把上述生命周期行为固化为可验证的事实。trigger-type-lifecycle.test.ts 使用「双 worker」模式provider 注册触发器类型并维护bindings映射consumer 绑定函数随后验证触发分发provider 触发时所有绑定函数都被调用且载荷正确透传expect(handlerSpy.mock.calls[0][0]).toMatchObject({ n: 1 })回调契约consumer 绑定后provider 的registerTriggerSpy被调用一次且收到function_id正确重连恢复provider worker 重连后触发器重新绑定re-binds triggers when the provider worker reconnects解绑通知consumer 断开时provider 的unregisterTrigger被调用invokes unregisterTrigger on the provider when the consumer disconnects。这些用例同时印证了文档中的关键断言引擎负责把 bind/unbind 转发给发布方回调、绑定是网络级的、且发布方进程内的bindings表是真正驱动分发的数据结构。测试中还刻意让 provider 与 consumer 使用不同的 worker 名称因为引擎对每个(namespace, name)只保留一个存活 worker。小结触发器存在消费者与发布者两种角色消费用worker.registerTrigger({ type, function_id, config })发布用worker.registerTriggerType({ id, description }, handler)。触发器类型 一个字符串id 发布方进程内维护的按绑定路由表引擎只记录绑定、不做分发仅把 bind/unbind 以回调形式转发给发布方。TriggerHandler只需实现registerTrigger(config)与unregisterTrigger(config)两个回调三种 SDK 签名一致TriggerConfig携带id、function_id、config、metadata与解析后的namespace。元数据消费者按绑定附加与 schema发布者按类型声明职责分明schema 当前仅作为契约文档引擎不做运行时校验。断开连接时发布的类型会被自动清理运行期可用unregisterTriggerType主动拆除Node 的TriggerTypeRef提供.unregister()快捷方式。事件分发没有专用 API发布方查表后直接用worker.trigger(...)调用绑定函数引擎负责评估config与condition_function_id后路由。如需深入了解调用侧的完整机制TriggerAction变体、条件门控、多绑定可继续阅读 Using iii / Triggers引擎侧的发现接口engine::triggers::list、engine::registered-triggers::list、engine::trigger-types::list实现在 engine-functions worker 中。【免费下载链接】iiiEffortlessly compose, extend, and observe every service in real-time for the first time ever.项目地址: https://gitcode.com/GitHub_Trending/mo/iii创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考