diff --git a/README.md b/README.md index 261c2c8..8f355ae 100644 --- a/README.md +++ b/README.md @@ -30,7 +30,7 @@ | **example** | 最简示例，Plugin 基类 + Database + 生命周期 | 入门学习 | | **auth** | SQLModel ORM + 管理员初始化 + 多进程幂等 | 数据库操作 | | **doubao** | 外部 API 客户端（Ark/OpenAI）+ AICC 安全通道 | 外部集成 | -| **aibox_sophon_adapter** | 自定义 `APIRouter` + MQTT 转发 + 后台任务 | 高级模式 | +| **aibox_sophon_adapter** | 自定义 `APIRouter` + MQTT 客户端组件（`mqtt_clients.py`）+ 后台任务 | 高级模式 | --- @@ -56,21 +56,24 @@ touch src/plugins//router/__init__.py ### 第二步：注册插件 -在 `config/workers/config.toml` 中添加： +在 `config/workers//config.toml` 中通过 `[workers]` 节注册（网关启动时自动扫描发现）： ```toml -[[workers]] -type = "" -module = "plugins." -factory = "create_workers" -cleanup = "cleanup" -count = 3 +# config/workers//config.toml +[workers] +type = "" # 插件类型（worker type） +module = "plugins." # Python 模块路径 +factory = "create_workers" # 工厂函数名（Plugin 基类绑定的方法） +cleanup = "cleanup" # 清理函数名 +count = 3 # Worker 副本数 +# 运行时配置（展平后传给 create_workers 的 config 参数） [workers.config] -type = "" data_dir = "./data" ``` +> 旧式全局注册文件 `config/workers/config.toml`（`[[workers]]` 数组）仅作兜底兼容，新插件请使用各插件自己的 `config.toml`。 + ### 第三步：编写端点 TOML 配置 创建 `config/workers//config.toml`（分组配置）： @@ -167,9 +170,9 @@ create_workers = _plugin.create_workers cleanup = _plugin.cleanup ``` -基类自动处理：幂等、统一日志、数据库建库/释放、后台任务取消、清理钩子、模型自动绑定。详见 `utils/worker/README.md` 的「Plugin 基类」章节。 +基类自动处理：幂等（初始化锁）、统一日志、数据库建库/释放、后台任务取消、清理钩子、模型自动绑定、cleanup 时重置已发布的 global_var。详见 `utils/worker/README.md` 的「Plugin 基类」章节。 -> **兼容旧写法**：不依赖基类的传统 `create_workers` / `cleanup` 模块级函数写法仍受支持（见 `utils/worker/README.md`）。 +> **旧写法已废弃**：不依赖基类的传统 `create_workers` / `cleanup` 模块级函数写法仅框架层面保留兼容，现有插件（auth / doubao / aibox_sophon_adapter / example）均已迁移到 Plugin 基类，新插件请勿采用旧写法。 --- @@ -310,17 +313,39 @@ async def handler(request: RequestCls) -> ResponseCls: ... ``` -| method | 参数来源 | 请求绑定 | -| ------ | ----------------- | ---------------------------- | -| `GET` | URL query string | TOML `params` 定义 | -| `POST` | JSON body | 自动绑定到 Pydantic 请求模型 | -| `MQTT` | MQTT payload JSON | 同 POST | +| method | 参数来源 | 请求绑定 | +| ------ | ------------------ | ---------------------------- | +| `GET` | URL query string | TOML `params` 定义 | +| `POST` | JSON body | 自动绑定到 Pydantic 请求模型 | +| `MQTT` | MQTT payload JSON | 同 POST | + +**请求信封协议**：网关统一把请求组装为 `{"meta": {"method", "user_info"}, "data": {...}}` +发给 Worker；运行时按 `meta.method` 分派到对应 `@rpc_method`，`data` 经请求模型校验后 +传给处理器。MQTT 客户端发布消息时需按信封格式（见主 README「请求负载格式」）， +MQTT 专用端点要显式 `"meta": {"method": "MQTT"}`。 - 设置 `internal = true` 后，端点不注册 HTTP 路由，仅可通过 Zenoh RPC 或 MQTT 内部调用 - MQTT 方法返回 `None` 时，Bridge 自动跳过响应发布 --- +## 启动契约校验 + +服务启动时（`load_workers` 之后、注册路由之前），网关自动核对三方一致性： + +| 检查项 | 策略 | +| ---------------------------- | --------------------------- | +| TOML HTTP 路由缺 `@rpc_method` | **硬失败**，启动直接终止 | +| `[MQTT]` / `internal` 路由缺注册 | 警告日志 | +| 已注册方法无 TOML 路由 | 警告日志（反向漂移检测） | +| params/response_model 缺生成模型 | 警告日志（gen_models 过期） | +| 生成模型未被任何路由引用 | 警告日志（孤儿模型） | + +新增端点后如果启动报错「缺少对应的 @rpc_method 注册」，检查 TOML 的 `subject`/方法节 +与 `router/.py` 中的 `@rpc_method(subject=..., method=...)` 是否一致。 + +--- + ## 命名约定 | 概念 | 约定 | 示例 | diff --git a/aibox_sophon_adapter/README.md b/aibox_sophon_adapter/README.md index b5e36ce..dadbad2 100644 --- a/aibox_sophon_adapter/README.md +++ b/aibox_sophon_adapter/README.md @@ -87,10 +87,12 @@ sequenceDiagram ```text src/plugins/aibox_sophon_adapter/ -├── __init__.py # 插件入口：create_workers() 工厂 + cleanup() -│ # - 初始化 DB / StreamManager / 各 MQTT 客户端 -│ # - _build_subscribe_routes() 构建订阅路由 -│ # - 启动离线检查后台任务 +├── __init__.py # 插件入口：AiboxSophonAdapterPlugin（Plugin 基类） +│ # - on_init: DB / StreamManager / MQTT 客户端 / 全局变量 / 后台任务 +│ # - on_cleanup: 流控制器与 MQTT 客户端关闭（db/后台任务由基类处理） +├── mqtt_clients.py # MQTT 客户端能力组件（从 __init__ 抽取，统一"启动失败降级为 None"模式） +│ # - start_mqtt_forward / start_mqtt_subscribe / start_internal_mqtt +│ # - build_subscribe_routes() 构建订阅路由 ├── global_var.py # 模块级全局变量（db / stream_manager / mqtt_* / 各配置） ├── routes.py # 自定义 FastAPI 路由（video_upload / video / alarm_report） ├── edge_app_controller.py # 外部 MQTT 请求-响应处理器（command 注册表分发） @@ -356,7 +358,7 @@ tags = ["aibox_sophon_adapter", "sophon", "adapter"] ### 修改配置 编辑 `config/workers/aibox_sophon_adapter/config.toml` 后重启服务即可生效，无需重新生成模型 -（运行时配置均在 `create_workers` 中读取）。 +（运行时配置均在 `AiboxSophonAdapterPlugin.on_init` 中读取；MQTT 客户端启动逻辑见 `mqtt_clients.py`）。 --- diff --git a/aibox_sophon_adapter/__init__.py b/aibox_sophon_adapter/__init__.py index 830bb0d..b417e22 100644 --- a/aibox_sophon_adapter/__init__.py +++ b/aibox_sophon_adapter/__init__.py @@ -2,7 +2,7 @@ AI Box Sophon Adapter 插件 基于 MQTT 的 AI Box 设备心跳适配器，实现： - 1. 通过 MQTT 主题 /board_ping 接收设备心跳上报 + 1. 通过 MQTT 主题接收设备心跳上报 2. 解析设备系统参数（温度、硬盘、内存、通道、任务、TPU 等） 3. 持久化心跳数据到 SQLite 数据库 4. 提供 HTTP API 查询设备在线状态和详情 @@ -15,122 +15,25 @@ AI Box Sophon Adapter 插件 - 设备按配置间隔（默认 5 秒）上报心跳 - 服务端每 check_interval 秒检查一次 - 超过 heartbeat_timeout 秒未收到心跳的设备标记为离线 + +生命周期: 基于 Plugin 基类统一管理（数据库/后台任务/清理钩子由基类处理）， +MQTT 客户端启动块已抽取到 mqtt_clients 能力组件。 """ import asyncio import logging from datetime import datetime from pathlib import Path -from typing import Any -from chongming_worker import auto_bind_models -from chongming_worker.runtime import create_workers as _create_workers +from chongming_worker.plugin import Plugin -from .router import * -from .routes import router # 自定义 FastAPI 路由（网关自动发现） - -# 外部 MQTT 订阅请求处理函数（订阅 /edge_app_controller 等主题的 handler） +from .router import * # noqa: F401,F403 触发 @rpc_method 注册 +from .routes import router # noqa: F401 自定义 FastAPI 路由（网关通过包属性自动发现） from .edge_app_controller import edge_app_controller # noqa: F401 +from .mqtt_clients import start_mqtt_forward, start_mqtt_subscribe, start_internal_mqtt logger = logging.getLogger(__name__) -def _build_subscribe_routes(topics_cfg: list) -> list: - """ - 根据配置的订阅主题列表构建 MQTTSubscribeClient 路由。 - - topics_cfg: [{"topic": "...", "handler": "board_ping", "reply_topic": "..."}, ...] - - topic: 订阅主题（支持 + / # 通配符） - - handler: 处理函数名，按序从插件 router 模块、插件包自身查找 - - reply_topic: 可选，回复主题；处理后返回值将发布到该主题（支持 {1}{2} 占位符） - - 适配约定（按首个位置参数类型）: - - Pydantic 模型: 用 payload 构造模型实例后调用（如 board_ping -> BoardPingRequest(**payload)） - - dict: 直接传整个 payload 和 topic（如 edge_app_controller(payload, topic)） - - 其他: 将 payload 作为关键字参数展开调用 - - 返回: [{"topic": ..., "handler": async callable(payload, topic), "reply_topic": ...}] - """ - import importlib - from typing import get_origin - - from pydantic import BaseModel - - # 注意: 不能用 "from . import router"，因为 __init__.py 中 - # "from .routes import router" 已将 router 设为包级属性（FastAPI APIRouter 对象）， - # 会遮蔽子模块包 aibox_sophon_adapter.router，故用 importlib 显式按名导入。 - router_mod = importlib.import_module(f"{__name__}.router") - pkg_mod = importlib.import_module(__name__) - - routes = [] - for t in topics_cfg or []: - topic = t.get("topic", "") - handler_name = t.get("handler", "") - reply_topic = t.get("reply_topic") - if not topic or not handler_name: - logger.warning(f"[MQTTSubscribe] 跳过无效订阅项: {t}") - continue - - fn = getattr(router_mod, handler_name, None) - if fn is None: - fn = getattr(pkg_mod, handler_name, None) - if fn is None: - logger.warning( - f"[MQTTSubscribe] 未知 handler '{handler_name}'，跳过 {topic}" - ) - continue - - # 探测首个位置参数的类型注解 - hints = getattr(fn, "__annotations__", {}) - first_type: Any = None - for pname, ptype in hints.items(): - if pname in ("worker_id", "return"): - continue - first_type = ptype - break - - if ( - isinstance(first_type, type) - and issubclass(first_type, BaseModel) - ): - model_cls = first_type - - async def _cb_model(data, topic, _fn=fn, _model=model_cls): - return await _fn(_model(**data)) - routes.append({ - "topic": topic, - "handler": _cb_model, - "reply_topic": reply_topic, - }) - elif first_type is dict or get_origin(first_type) is dict: - # 处理函数期望整个 dict 作为第一个参数（如 edge_app_controller(payload, topic)） - async def _cb_payload(data, topic, _fn=fn): - return await _fn(data, topic) - routes.append({ - "topic": topic, - "handler": _cb_payload, - "reply_topic": reply_topic, - }) - else: - # 其他签名：将 payload 按关键字参数展开调用 - async def _cb_kwargs(data, topic, _fn=fn): - if isinstance(data, dict): - return await _fn(**data) - return await _fn() - routes.append({ - "topic": topic, - "handler": _cb_kwargs, - "reply_topic": reply_topic, - }) - - return routes - -# 自动绑定 Pydantic 模型（用于 gen_models 生成的模型） -auto_bind_models("aibox_sophon_adapter") - -# 离线检查任务引用 -_offline_check_task: asyncio.Task | None = None - - async def _check_offline_devices(interval: float): """ 后台任务：定期检查离线设备 @@ -150,7 +53,7 @@ async def _check_offline_devices(interval: float): from .models.orm import BoardHeartbeat from sqlalchemy import select - stmt = select(BoardHeartbeat).where(BoardHeartbeat.is_online == True) # type: ignore + stmt = select(BoardHeartbeat).where(BoardHeartbeat.is_online) result = await session.execute(stmt) records = result.scalars().all() @@ -175,206 +78,110 @@ async def _check_offline_devices(interval: float): logger.error(f"[OfflineChecker] 检查异常: {e}") -async def create_workers(session, config, count=1, worker_manager=None): - """ - 插件入口 — 初始化数据库和全局资源，创建 Worker 副本 +class AiboxSophonAdapterPlugin(Plugin): + """AI Box 适配插件 — 统一生命周期。""" - 由 FastAPI lifespan 在启动时调用。 - """ - from .global_var import set_global_var, heartbeat_timeout as _default_timeout - from chongming_database import Database - from chongming_stream_manager import StreamManager + type = "aibox_sophon_adapter" + db_name = "aibox_sophon_adapter.db" # 自动在 data_dir 下建 SQLite 库 + auto_bind = True # 首次初始化时自动 auto_bind_models - # 所有运行时配置均从 [workers.config] 及其子节读取 - data_dir = Path(config.get("data_dir", "./data")) - mediamtx_addr = config.get("mediamtx_addr", "rtsp://127.0.0.1:8554") - link_path = config.get("link_path", "/data/bm-app") + async def on_init(self): + """首次 create_workers 时调用一次：初始化数据库客户端、流控制器、后台任务。""" + config = self.config + from .global_var import set_global_var, heartbeat_timeout as _default_timeout + from chongming_stream_manager import StreamManager - # 心跳配置 - heartbeat = config.get("heartbeat", {}) - hb_interval = heartbeat.get("interval", 5) - hb_timeout = heartbeat.get("timeout", _default_timeout) - check_interval = heartbeat.get("check_interval", 10) + # 所有运行时配置均从 [workers.config] 及其子节读取 + data_dir = Path(config.get("data_dir", "./data")) + mediamtx_addr = config.get("mediamtx_addr", "rtsp://127.0.0.1:8554") + link_path = config.get("link_path", "/data/bm-app") - # 告警保留配置 - alarm_retention = config.get("alarm_retention", {}) - alarm_max = alarm_retention.get("max_records", 0) + # 心跳配置 + heartbeat = config.get("heartbeat", {}) + hb_interval = heartbeat.get("interval", 5) + hb_timeout = heartbeat.get("timeout", _default_timeout) + check_interval = heartbeat.get("check_interval", 10) - # 豆包 AI 分析开关 - doubao_cfg = config.get("doubao_analysis", {}) - doubao_enabled = doubao_cfg.get("enabled", True) + # 告警保留配置 + alarm_retention = config.get("alarm_retention", {}) + alarm_max = alarm_retention.get("max_records", 0) - # 告警转发配置 - alarm_fwd_cfg = config.get("alarm_forward", {}) - alarm_fwd_enabled = alarm_fwd_cfg.get("enabled", False) - alarm_fwd_url = alarm_fwd_cfg.get("url", "http://dg.datagoo.cn:60014/open-api/event-alarm/receive") - alarm_fwd_video_url = alarm_fwd_cfg.get("video_upload_url", "http://dg.datagoo.cn:60014/open-api/event-alarm/receiveVideoUpload") - alarm_fwd_timeout = alarm_fwd_cfg.get("timeout", 10.0) + # 豆包 AI 分析开关 + doubao_cfg = config.get("doubao_analysis", {}) + doubao_enabled = doubao_cfg.get("enabled", True) - # 初始化数据库 - db_path = data_dir / "aibox_sophon_adapter.db" - db_path.parent.mkdir(parents=True, exist_ok=True) + # 告警转发配置 + alarm_fwd_cfg = config.get("alarm_forward", {}) + alarm_fwd_enabled = alarm_fwd_cfg.get("enabled", False) + alarm_fwd_url = alarm_fwd_cfg.get("url", "http://dg.datagoo.cn:60014/open-api/event-alarm/receive") + alarm_fwd_video_url = alarm_fwd_cfg.get("video_upload_url", "http://dg.datagoo.cn:60014/open-api/event-alarm/receiveVideoUpload") + alarm_fwd_timeout = alarm_fwd_cfg.get("timeout", 10.0) - db = Database(db_path) - await db.initialize() + # 数据库已由基类初始化（data_dir/aibox_sophon_adapter.db） + db = self.db - # 初始化视频流控制器 - stream_manager = StreamManager(mediamtx_addr) + # 初始化视频流控制器 + stream_manager = StreamManager(mediamtx_addr) - # upload 根目录（用于清理告警关联视频） - upload_root_path = str(data_dir / "uploads") + # upload 根目录（用于清理告警关联视频） + upload_root_path = str(data_dir / "uploads") - # 设定模块级全局变量（供 _check_offline_devices 使用） - global heartbeat_timeout - heartbeat_timeout = hb_timeout - - # 初始化外部 MQTT 转发客户端 - mqtt_fwd = config.get("mqtt_forward", {}) - mqtt_fwd_enabled = mqtt_fwd.get("enabled", False) - mqtt_fwd_client = None - if mqtt_fwd_enabled: - from chongming_mqtt.mqtt_forward import MQTTForwardClient - mqtt_fwd_uri = mqtt_fwd.get("uri", "") - mqtt_fwd_topic = mqtt_fwd.get("topic", "/board_ping") - logger.warning(f"[MQTTForward] 配置已加载: enabled={mqtt_fwd_enabled}, uri={mqtt_fwd_uri}, topic={mqtt_fwd_topic}") - if mqtt_fwd_uri: - mqtt_fwd_client = MQTTForwardClient(uri=mqtt_fwd_uri, topic=mqtt_fwd_topic) - try: - await mqtt_fwd_client.start() - logger.warning(f"[MQTTForward] 外部转发已启用: {mqtt_fwd_uri} → {mqtt_fwd_topic}") - except Exception as e: - logger.warning(f"[MQTTForward] 启动失败，将跳过转发: {e}") - mqtt_fwd_client = None - else: - logger.warning(f"[MQTTForward] mqtt_forward 未启用，配置内容: {mqtt_fwd}") - - # 初始化外部 MQTT 订阅客户端（本插件直接订阅外部 Broker，不走网关 Bridge） - mqtt_sub = config.get("mqtt_subscribe", {}) - mqtt_sub_enabled = mqtt_sub.get("enabled", False) - mqtt_sub_client = None - if mqtt_sub_enabled: - from chongming_mqtt.mqtt_subscribe import MQTTSubscribeClient - - mqtt_sub_uri = mqtt_sub.get("uri", "") - mqtt_sub_topics = mqtt_sub.get("topics", []) or [] - logger.warning( - f"[MQTTSubscribe] 配置已加载: enabled={mqtt_sub_enabled}, " - f"uri={mqtt_sub_uri}, topics={mqtt_sub_topics}" + # MQTT 客户端能力组件（启动失败降级为 None，仅警告） + mqtt_fwd_client = await start_mqtt_forward(config.get("mqtt_forward", {})) + mqtt_sub_client = await start_mqtt_subscribe(config.get("mqtt_subscribe", {})) + int_mqtt_client, int_req_topic, int_resp_topic = await start_internal_mqtt( + config.get("internal_mqtt", {}) ) - if mqtt_sub_uri and mqtt_sub_topics: - routes = _build_subscribe_routes(mqtt_sub_topics) - mqtt_sub_client = MQTTSubscribeClient(uri=mqtt_sub_uri, routes=routes) - try: - await mqtt_sub_client.start() - logger.warning(f"[MQTTSubscribe] 外部订阅已启用: {mqtt_sub_uri}") - except Exception as e: - logger.warning(f"[MQTTSubscribe] 启动失败，将跳过订阅: {e}") - mqtt_sub_client = None - else: - logger.warning( - "[MQTTSubscribe] 未配置 uri 或 topics，跳过外部订阅" - ) - else: - logger.warning(f"[MQTTSubscribe] mqtt_subscribe 未启用，配置内容: {mqtt_sub}") - # 初始化内部 MQTT 请求-响应客户端 - # （handler 内对任意应用进程发起请求-响应，主题显式传入，默认连内部 127.0.0.1:1883） - int_mqtt_cfg = config.get("internal_mqtt", {}) - int_mqtt_enabled = int_mqtt_cfg.get("enabled", False) - int_mqtt_client = None - # 内部请求-响应主题（edge_app_controller 的 alg_media_fetch 命令使用）。 - # 与盒子侧约定一致（盒子只设根主题 box，子路径固定 aibox_sophon/...）； - # 勿用 chongming/{插件}/{操作} 保留命名空间——网关 MQTT Bridge 订阅了 - # chongming/+/+，会拦截 3 级主题并按 Worker RPC 路由，而 edge_app_controller - # 不是 @rpc_method（Worker 无对应 Queryable），导致 worker_manager 报 - # "Worker ... 无回复，尝试重试"。 - int_req_topic = int_mqtt_cfg.get( - "request_topic", "box/aibox_sophon/edge_app_controller" - ) - int_resp_topic = int_mqtt_cfg.get( - "response_topic", "box/aibox_sophon/edge_app_controller_reply" - ) - if int_mqtt_enabled: - from chongming_mqtt.internal_mqtt import InternalMQTTClient + set_global_var("db", db) + set_global_var("heartbeat_timeout", hb_timeout) + set_global_var("heartbeat_interval", hb_interval) + set_global_var("stream_manager", stream_manager) + set_global_var("link_path", link_path) + set_global_var("mqtt_forward", mqtt_fwd_client) + set_global_var("mqtt_subscribe", mqtt_sub_client) + set_global_var("internal_mqtt", int_mqtt_client) + set_global_var("internal_request_topic", int_req_topic) + set_global_var("internal_response_topic", int_resp_topic) + set_global_var("upload_root", upload_root_path) + set_global_var("alarm_max_records", alarm_max) + set_global_var("doubao_analysis_enabled", doubao_enabled) + set_global_var("alarm_forward_enabled", alarm_fwd_enabled) + set_global_var("alarm_forward_url", alarm_fwd_url) + set_global_var("alarm_forward_video_url", alarm_fwd_video_url) + set_global_var("alarm_forward_timeout", alarm_fwd_timeout) - int_mqtt_uri = int_mqtt_cfg.get("uri", "mqtt://127.0.0.1:1883") - logger.warning( - f"[InternalMQTT] 配置已加载: enabled={int_mqtt_enabled}, uri={int_mqtt_uri}, " - f"request_topic={int_req_topic}, response_topic={int_resp_topic}" + logger.info( + "[aibox_sophon_adapter] 已初始化: " + f"心跳间隔={hb_interval}s, 超时={hb_timeout}s, 检查间隔={check_interval}s" ) - int_mqtt_client = InternalMQTTClient(uri=int_mqtt_uri) - try: - await int_mqtt_client.start() - logger.warning(f"[InternalMQTT] 内部请求-响应客户端已启用: {int_mqtt_uri}") - except Exception as e: - logger.warning(f"[InternalMQTT] 启动失败，将跳过: {e}") - int_mqtt_client = None - else: - logger.warning(f"[InternalMQTT] internal_mqtt 未启用，配置内容: {int_mqtt_cfg}") - set_global_var("db", db) - set_global_var("heartbeat_timeout", hb_timeout) - set_global_var("heartbeat_interval", hb_interval) - set_global_var("stream_manager", stream_manager) - set_global_var("link_path", link_path) - set_global_var("mqtt_forward", mqtt_fwd_client) - set_global_var("mqtt_subscribe", mqtt_sub_client) - set_global_var("internal_mqtt", int_mqtt_client) - set_global_var("internal_request_topic", int_req_topic) - set_global_var("internal_response_topic", int_resp_topic) - set_global_var("upload_root", upload_root_path) - set_global_var("alarm_max_records", alarm_max) - set_global_var("doubao_analysis_enabled", doubao_enabled) - set_global_var("alarm_forward_enabled", alarm_fwd_enabled) - set_global_var("alarm_forward_url", alarm_fwd_url) - set_global_var("alarm_forward_video_url", alarm_fwd_video_url) - set_global_var("alarm_forward_timeout", alarm_fwd_timeout) + # 后台任务由基类跟踪，cleanup 时自动取消 + self.spawn_task(_check_offline_devices(check_interval)) - logger.info( - f"[aibox_sophon_adapter] 已初始化: " - f"心跳间隔={hb_interval}s, 超时={hb_timeout}s, 检查间隔={check_interval}s" - ) + async def on_cleanup(self): + """cleanup 时调用：关闭流控制器与外部 MQTT 客户端（db 与后台任务由基类处理）。""" + from .global_var import ( + stream_manager, mqtt_forward, mqtt_subscribe, internal_mqtt, + ) - # 启动离线设备检查后台任务 - global _offline_check_task - _offline_check_task = asyncio.create_task(_check_offline_devices(check_interval)) + if stream_manager is not None: + await stream_manager.stop_all() - # 创建 Worker 副本 - return _create_workers(session, config, count, worker_manager) + if mqtt_forward is not None: + await mqtt_forward.stop() + + if mqtt_subscribe is not None: + await mqtt_subscribe.stop() + + if internal_mqtt is not None: + await internal_mqtt.stop() + + logger.info("[aibox_sophon_adapter] 清理完成") -async def cleanup(worker_item: str): - """ - 清理资源 - - 在 Worker 关闭时调用，关闭数据库连接和后台任务。 - """ - from .global_var import ( - db, stream_manager, mqtt_forward, mqtt_subscribe, internal_mqtt, - ) - - global _offline_check_task - if _offline_check_task is not None and not _offline_check_task.done(): - _offline_check_task.cancel() - try: - await _offline_check_task - except asyncio.CancelledError: - pass - logger.info("[aibox_sophon_adapter] 离线检查任务已停止") - - if db is not None: - await db.dispose() - - if stream_manager is not None: - await stream_manager.stop_all() - - if mqtt_forward is not None: - await mqtt_forward.stop() - - if mqtt_subscribe is not None: - await mqtt_subscribe.stop() - - if internal_mqtt is not None: - await internal_mqtt.stop() - - logger.info(f"[aibox_sophon_adapter] {worker_item} 清理完毕") +# 暴露给 app lifespan 的模块级入口（bound method，签名与旧写法完全一致） +# 注意: 实例名必须为 plugin（router 已被 routes.router APIRouter 占用） +plugin = AiboxSophonAdapterPlugin() +create_workers = plugin.create_workers +cleanup = plugin.cleanup diff --git a/aibox_sophon_adapter/mqtt_clients.py b/aibox_sophon_adapter/mqtt_clients.py new file mode 100644 index 0000000..81622d2 --- /dev/null +++ b/aibox_sophon_adapter/mqtt_clients.py @@ -0,0 +1,219 @@ +""" +AI Box Sophon Adapter — MQTT 客户端能力组件 + +从插件 __init__.py 抽取的三个外部 MQTT 客户端启动块，统一模式: + - enabled=False → 仅警告"未启用"并返回 None + - enabled=True → 懒加载客户端类，启动失败降级为 None（仅警告，不阻断插件） + +内部 MQTT 主题约定（勿改）: + 与盒子侧约定一致（盒子只设根主题 box，子路径固定 aibox_sophon/...）； + 勿用 chongming/{插件}/{操作} 保留命名空间——网关 MQTT Bridge 订阅了 + chongming/+/+，会拦截 3 级主题并按 Worker RPC 路由，而 edge_app_controller + 不是 @rpc_method（Worker 无对应 Queryable），导致 worker_manager 报 + "Worker ... 无回复，尝试重试"。 +""" +import logging +from typing import Any + +logger = logging.getLogger(__name__) + +# 插件包名（作为 importlib 探测的基准，兼容 plugins.xxx 与 xxx 两种导入路径） +_PKG = __package__ or "plugins.aibox_sophon_adapter" + + +def _import_first(candidates: list) -> Any: + """按序尝试导入模块，全部失败返回 None。""" + import importlib + + for name in candidates: + try: + return importlib.import_module(name) + except ImportError: + continue + return None + + +def build_subscribe_routes(topics_cfg: list) -> list: + """ + 根据配置的订阅主题列表构建 MQTTSubscribeClient 路由。 + + topics_cfg: [{"topic": "...", "handler": "board_ping", "reply_topic": "..."}, ...] + - topic: 订阅主题（支持 + / # 通配符） + - handler: 处理函数名，按序从插件 router 模块、插件包自身查找 + - reply_topic: 可选，回复主题；处理后返回值将发布到该主题（支持 {1}{2} 占位符） + + 适配约定（按首个位置参数类型）: + - Pydantic 模型: 用 payload 构造模型实例后调用（如 board_ping -> BoardPingRequest(**payload)） + - dict: 直接传整个 payload 和 topic（如 edge_app_controller(payload, topic)） + - 其他: 将 payload 作为关键字参数展开调用 + + 返回: [{"topic": ..., "handler": async callable(payload, topic), "reply_topic": ...}] + """ + from typing import get_origin + + from pydantic import BaseModel + + # 注意: 不能用 "from .. import router" 取子模块，因为插件包属性 router + # 已被 routes.router（FastAPI APIRouter）遮蔽，故用 importlib 按名导入。 + pkg_base = _PKG + pkg_leaf = _PKG.rsplit(".", 1)[-1] + router_mod = _import_first([f"{pkg_base}.router", f"{pkg_leaf}.router"]) + pkg_mod = _import_first([pkg_base, pkg_leaf]) + + routes = [] + for t in topics_cfg or []: + topic = t.get("topic", "") + handler_name = t.get("handler", "") + reply_topic = t.get("reply_topic") + if not topic or not handler_name: + logger.warning("[MQTTSubscribe] 跳过无效订阅项: %s", t) + continue + + fn = None + if router_mod is not None: + fn = getattr(router_mod, handler_name, None) + if fn is None and pkg_mod is not None: + fn = getattr(pkg_mod, handler_name, None) + if fn is None: + logger.warning( + "[MQTTSubscribe] 未知 handler '%s'，跳过 %s", handler_name, topic + ) + continue + + # 探测首个位置参数的类型注解 + hints = getattr(fn, "__annotations__", {}) + first_type: Any = None + for pname, ptype in hints.items(): + if pname in ("worker_id", "return"): + continue + first_type = ptype + break + + if ( + isinstance(first_type, type) + and issubclass(first_type, BaseModel) + ): + model_cls = first_type + + async def _cb_model(data, topic, _fn=fn, _model=model_cls): + return await _fn(_model(**data)) + routes.append({ + "topic": topic, + "handler": _cb_model, + "reply_topic": reply_topic, + }) + elif first_type is dict or get_origin(first_type) is dict: + # 处理函数期望整个 dict 作为第一个参数（如 edge_app_controller(payload, topic)） + async def _cb_payload(data, topic, _fn=fn): + return await _fn(data, topic) + routes.append({ + "topic": topic, + "handler": _cb_payload, + "reply_topic": reply_topic, + }) + else: + # 其他签名：将 payload 按关键字参数展开调用 + async def _cb_kwargs(data, topic, _fn=fn): + if isinstance(data, dict): + return await _fn(**data) + return await _fn() + routes.append({ + "topic": topic, + "handler": _cb_kwargs, + "reply_topic": reply_topic, + }) + + return routes + + +async def start_mqtt_forward(cfg: dict) -> Any: + """启动外部 MQTT 转发客户端，返回客户端或 None。""" + enabled = cfg.get("enabled", False) + if not enabled: + logger.warning("[MQTTForward] mqtt_forward 未启用，配置内容: %s", cfg) + return None + + from chongming_mqtt.mqtt_forward import MQTTForwardClient + + uri = cfg.get("uri", "") + topic = cfg.get("topic", "/board_ping") + logger.warning( + "[MQTTForward] 配置已加载: enabled=%s, uri=%s, topic=%s", enabled, uri, topic + ) + if not uri: + return None + + client = MQTTForwardClient(uri=uri, topic=topic) + try: + await client.start() + logger.warning("[MQTTForward] 外部转发已启用: %s → %s", uri, topic) + return client + except Exception as e: + logger.warning("[MQTTForward] 启动失败，将跳过转发: %s", e) + return None + + +async def start_mqtt_subscribe(cfg: dict) -> Any: + """启动外部 MQTT 订阅客户端（本插件直接订阅外部 Broker，不走网关 Bridge）。""" + enabled = cfg.get("enabled", False) + if not enabled: + logger.warning("[MQTTSubscribe] mqtt_subscribe 未启用，配置内容: %s", cfg) + return None + + from chongming_mqtt.mqtt_subscribe import MQTTSubscribeClient + + uri = cfg.get("uri", "") + topics = cfg.get("topics", []) or [] + logger.warning( + "[MQTTSubscribe] 配置已加载: enabled=%s, uri=%s, topics=%s", + enabled, uri, topics, + ) + if not (uri and topics): + logger.warning("[MQTTSubscribe] 未配置 uri 或 topics，跳过外部订阅") + return None + + routes = build_subscribe_routes(topics) + client = MQTTSubscribeClient(uri=uri, routes=routes) + try: + await client.start() + logger.warning("[MQTTSubscribe] 外部订阅已启用: %s", uri) + return client + except Exception as e: + logger.warning("[MQTTSubscribe] 启动失败，将跳过订阅: %s", e) + return None + + +async def start_internal_mqtt(cfg: dict) -> tuple[Any, str, str]: + """ + 启动内部 MQTT 请求-响应客户端 + （handler 内对任意应用进程发起请求-响应，主题显式传入，默认连内部 127.0.0.1:1883）。 + + 返回 (client | None, request_topic, response_topic)： + 主题无论是否启用都按配置解析并返回（由调用方写入 global_var）。 + """ + req_topic = cfg.get( + "request_topic", "box/aibox_sophon/edge_app_controller" + ) + resp_topic = cfg.get( + "response_topic", "box/aibox_sophon/edge_app_controller_reply" + ) + enabled = cfg.get("enabled", False) + if not enabled: + logger.warning("[InternalMQTT] internal_mqtt 未启用，配置内容: %s", cfg) + return None, req_topic, resp_topic + + from chongming_mqtt.internal_mqtt import InternalMQTTClient + + uri = cfg.get("uri", "mqtt://127.0.0.1:1883") + logger.warning( + "[InternalMQTT] 配置已加载: enabled=%s, uri=%s, request_topic=%s, response_topic=%s", + enabled, uri, req_topic, resp_topic, + ) + client = InternalMQTTClient(uri=uri) + try: + await client.start() + logger.warning("[InternalMQTT] 内部请求-响应客户端已启用: %s", uri) + return client, req_topic, resp_topic + except Exception as e: + logger.warning("[InternalMQTT] 启动失败，将跳过: %s", e) + return None, req_topic, resp_topic diff --git a/auth/README.md b/auth/README.md index c6d7d0d..2000ffc 100644 --- a/auth/README.md +++ b/auth/README.md @@ -159,6 +159,10 @@ JWT Token 有效期 3 小时，载荷包含：`sub`（用户ID）、`username` ## 默认管理员账户 +插件基于 `Plugin` 基类统一生命周期：`on_init` 中执行管理员种子逻辑 +（先按用户名、再按邮箱兜底查询；多进程并发启动时靠唯一约束 + `IntegrityError` 回滚保证幂等）， +数据库 `auth.db` 由基类统一初始化并在 cleanup 时释放。 + 首次启动时自动创建： | 字段 | 值 | diff --git a/auth/__init__.py b/auth/__init__.py index 2d82da9..ad0df86 100644 --- a/auth/__init__.py +++ b/auth/__init__.py @@ -1,65 +1,74 @@ -from pathlib import Path +""" +Auth Worker 插件 — 鉴权管理（登录 / 注册 / 资料 / 改密） + +基于 Plugin 基类统一生命周期：数据库由基类初始化（auth.db）， +on_init 中执行 admin 幂等种子逻辑，on_cleanup 无需重复 dispose。 +""" +import logging from datetime import datetime -from chongming_worker import auto_bind_models -from chongming_worker.runtime import create_workers as _create_workers +from chongming_worker.plugin import Plugin from chongming_password import hash_password -from .router import * +from .router import * # noqa: F401,F403 触发 @rpc_method 注册 + +logger = logging.getLogger(__name__) -auto_bind_models("auth") +class AuthPlugin(Plugin): + """Auth 插件 — 统一生命周期。""" + type = "auth" + db_name = "auth.db" # 自动在 data_dir 下建 SQLite 库 + auto_bind = True # 首次初始化时自动 auto_bind_models("auth") -async def create_workers(session, config, count=1, worker_manager=None): - from chongming_database import Database - from .global_var import set_global_var - from .models.auth import Users - from sqlalchemy import select + async def on_init(self): + """首次 create_workers 时调用一次：初始化 admin 账号（幂等）。""" + from .models.auth import Users + from sqlalchemy import select + from sqlalchemy.exc import IntegrityError - db_path = Path(config.get("data_dir", "./data")) / "auth.db" - db = Database(db_path) - await db.initialize() + db = self.db - # 初始化管理员（多进程并发启动时必须幂等，避免 UNIQUE 约束冲突） - from sqlalchemy.exc import IntegrityError - - async with db.get_session() as sess: - select_stmt = select(Users) \ - .where(Users.username == "admin") # type: ignore - result = await sess.execute(select_stmt) - user: Users | None = result.scalar_one_or_none() - - # 按用户名未命中时，再按邮箱兜底，避免历史数据中已存在同邮箱用户 - if user is None: + # 初始化管理员（多进程并发启动时必须幂等，避免 UNIQUE 约束冲突） + async with db.get_session() as sess: select_stmt = select(Users) \ - .where(Users.email == "admin@example.com") # type: ignore + .where(Users.username == "admin") # type: ignore result = await sess.execute(select_stmt) - user = result.scalar_one_or_none() + user: Users | None = result.scalar_one_or_none() - if user is None: - user = Users( - username="admin", - hashed_password=hash_password("admin"), - email="admin@example.com", - created_at=datetime.now(), - updated_at=datetime.now(), - is_superuser=True - ) - sess.add(user) - try: - await sess.commit() - except IntegrityError: - # 其他并发进程已抢先创建 admin，当前进程忽略即可 - await sess.rollback() - - set_global_var("db", db) + # 按用户名未命中时，再按邮箱兜底，避免历史数据中已存在同邮箱用户 + if user is None: + select_stmt = select(Users) \ + .where(Users.email == "admin@example.com") # type: ignore + result = await sess.execute(select_stmt) + user = result.scalar_one_or_none() - return _create_workers(session, config, count, worker_manager) + if user is None: + user = Users( + username="admin", + hashed_password=hash_password("admin"), + email="admin@example.com", + created_at=datetime.now(), + updated_at=datetime.now(), + is_superuser=True + ) + sess.add(user) + try: + await sess.commit() + except IntegrityError: + # 其他并发进程已抢先创建 admin，当前进程忽略即可 + await sess.rollback() -async def cleanup(worker_item: str): - from .global_var import db + # 写入全局变量供 router 使用（数据库由基类在 cleanup 时统一 dispose） + self.set_var("db", db) - if db is not None: - await db.dispose() - print(f"{worker_item} 清理完毕") \ No newline at end of file + async def on_cleanup(self): + """cleanup 时调用（数据库由基类处理）。""" + logger.info("[auth] 清理完成") + + +# 暴露给 app lifespan 的模块级入口（bound method，签名与旧写法完全一致） +plugin = AuthPlugin() +create_workers = plugin.create_workers +cleanup = plugin.cleanup diff --git a/doubao/README.md b/doubao/README.md index 5ec4e59..492fba0 100644 --- a/doubao/README.md +++ b/doubao/README.md @@ -15,7 +15,7 @@ ## 配置 -在 `config/workers/doubao/config.toml` 和 `config/workers/config.toml` 中设置： +在 `config/workers/doubao/config.toml` 中设置： ```toml [workers.config] @@ -64,9 +64,11 @@ model = "doubao-seed-evolving" # 模型名称 ## 实现细节 -- 使用 `httpx` 异步 HTTP 客户端调用火山引擎 API -- 支持流式响应（streaming）中的思考模式 token -- API Key 优先从环境变量 `VOLC_ACCESSKEY` 读取 +- 生命周期基于 `Plugin` 基类：`on_init` 按 `is_aicc` 分支创建 Ark（`AsyncArk`）或 + AICC（`AsyncOpenAI` + `AsyncSecureHttpClient`）客户端并写入 `global_var`； + `on_cleanup` 释放客户端引用（无数据库） +- API Key 优先取配置 `api_key`，其次环境变量 `VOLC_ACCESSKEY`（缺失时插件初始化报错） +- 支持 AICC 安全通信模式（懒加载 SDK，避免非 AICC 环境导入失败） - 超时时间默认 30 秒，在 `config.toml` 的 `defaults.timeout` 中配置 --- diff --git a/doubao/__init__.py b/doubao/__init__.py index 9aa33c8..a8b682b 100644 --- a/doubao/__init__.py +++ b/doubao/__init__.py @@ -1,95 +1,106 @@ """ 豆包视觉理解 Worker 插件 + +基于 Plugin 基类统一生命周期：on_init 中按配置创建 Ark / AICC 客户端， +on_cleanup 释放客户端引用（无数据库）。 """ import os import logging from openai import AsyncOpenAI from volcenginesdkarkruntime import AsyncArk -from chongming_worker import create_workers as _create_workers -from chongming_worker import auto_bind_models + +from chongming_worker.plugin import Plugin from .global_var import ( set_ark_client, set_aicc_client, set_is_aicc, cleanup_all, ) -from .router import * +from .router import * # noqa: F401,F403 触发 @rpc_method 注册 -logger = logging.getLogger("plugins.doubao") +logger = logging.getLogger(__name__) -# 自动绑定 gen_models 生成的 Pydantic 模型 -auto_bind_models("doubao") -async def create_workers(session, config, count=1, worker_manager=None): - """ - 工厂函数：由 app 生命周期调用，创建 Worker 协程。 - """ - logger.info("豆包插件初始化开始 (count=%d, aicc=%s)", count, config.get("is_aicc", False)) +class DoubaoPlugin(Plugin): + """豆包插件 — 统一生命周期。""" - is_aicc = config.get("is_aicc", False) - set_is_aicc(is_aicc) + type = "doubao" + db_name = None # 无数据库 + auto_bind = True # 首次初始化时自动 auto_bind_models("doubao") - api_key = config.get("api_key") or os.environ.get("VOLC_ACCESSKEY") - if not api_key: - logger.error("豆包插件初始化失败: 缺少 API_KEY") - raise ValueError("缺少 API_KEY，请在配置中设置 api_key 或环境变量 VOLC_ACCESSKEY") + async def on_init(self): + """首次 create_workers 时调用一次：初始化豆包客户端。""" + config = self.config + logger.info("豆包插件初始化开始 (count=%s, aicc=%s)", + config.get("count", 1), config.get("is_aicc", False)) - model = config.get("model", "doubao-seed-evolving") - os.environ["DOUBAO_MODEL"] = model - logger.info("豆包模型: %s", model) + is_aicc = config.get("is_aicc", False) + set_is_aicc(is_aicc) - if is_aicc: - logger.info("豆包使用 AICC 安全通信模式") - # AICC 模式：懒加载，避免非 AICC 环境导入失败 - try: - from bytedance.volcengine_aicc_sdk.secure_http_client import ( - AICCConfig, AsyncSecureHttpClient, + api_key = config.get("api_key") or os.environ.get("VOLC_ACCESSKEY") + if not api_key: + logger.error("豆包插件初始化失败: 缺少 API_KEY") + raise ValueError("缺少 API_KEY，请在配置中设置 api_key 或环境变量 VOLC_ACCESSKEY") + + model = config.get("model", "doubao-seed-evolving") + os.environ["DOUBAO_MODEL"] = model + logger.info("豆包模型: %s", model) + + if is_aicc: + logger.info("豆包使用 AICC 安全通信模式") + # AICC 模式：懒加载，避免非 AICC 环境导入失败 + try: + from bytedance.volcengine_aicc_sdk.secure_http_client import ( + AICCConfig, AsyncSecureHttpClient, + ) + except ImportError as e: + logger.error("AICC SDK 导入失败: %s，请确认已安装 bytedance_volcengine_aicc_sdk", e) + raise + + base_url = config.get("base_url", "https://ark.cn-beijing.volces.com/api/v3") + # ra_url 为远程证明（安全令牌）端点地址，可能与推理 base_url 不同 + # 若不单独配置则默认使用 base_url 的域名 + /v1/security/token + ra_url = config.get("ra_url", base_url) + ak = config.get("ak", "") + sk = config.get("sk", "") + ra_service_name = config.get("ra_service_name", "AICC.ConfidentialInference") + ra_policy_id = config.get("ra_policy_id", "router_policy") + + os.environ["DOUBAO_BASE_URL"] = base_url + logger.info("豆包 AICC base_url: %s, ra_url: %s, ra_service_name: %s, ra_policy_id: %s", + base_url, ra_url, ra_service_name, ra_policy_id) + + aicc_config = AICCConfig( + ra_url=ra_url, + auth_token=api_key, + ak=ak, + sk=sk, + aicc_llm_service_name=ra_service_name, + policy_id=ra_policy_id, ) - except ImportError as e: - logger.error("AICC SDK 导入失败: %s，请确认已安装 bytedance_volcengine_aicc_sdk", e) - raise + client = AsyncOpenAI( + api_key=api_key, + base_url=base_url, + http_client=AsyncSecureHttpClient(aicc_config=aicc_config), + ) + set_aicc_client(client) + logger.info("豆包 AICC 客户端已创建") + else: + logger.info("豆包使用标准 Ark SDK 模式") + # Ark 模式：使用 AsyncArk + base_url = config.get("base_url", "https://ark.cn-beijing.volces.com/api/v3") + os.environ["DOUBAO_BASE_URL"] = base_url + client = AsyncArk(base_url=base_url, api_key=api_key) + set_ark_client(client) + logger.info("豆包 Ark 客户端已创建 (base_url=%s)", base_url) - base_url = config.get("base_url", "https://ark.cn-beijing.volces.com/api/v3") - # ra_url 为远程证明（安全令牌）端点地址，可能与推理 base_url 不同 - # 若不单独配置则默认使用 base_url 的域名 + /v1/security/token - ra_url = config.get("ra_url", base_url) - ak = config.get("ak", "") - sk = config.get("sk", "") - ra_service_name = config.get("ra_service_name", "AICC.ConfidentialInference") - ra_policy_id = config.get("ra_policy_id", "router_policy") + logger.info("豆包插件初始化完成") - os.environ["DOUBAO_BASE_URL"] = base_url - logger.info("豆包 AICC base_url: %s, ra_url: %s, ra_service_name: %s, ra_policy_id: %s", - base_url, ra_url, ra_service_name, ra_policy_id) + async def on_cleanup(self): + """cleanup 时调用：释放客户端引用。""" + cleanup_all() + logger.info("[doubao] 清理完成") - aicc_config = AICCConfig( - ra_url=ra_url, - auth_token=api_key, - ak=ak, - sk=sk, - aicc_llm_service_name=ra_service_name, - policy_id=ra_policy_id, - ) - client = AsyncOpenAI( - api_key=api_key, - base_url=base_url, - http_client=AsyncSecureHttpClient(aicc_config=aicc_config), - ) - set_aicc_client(client) - logger.info("豆包 AICC 客户端已创建") - else: - logger.info("豆包使用标准 Ark SDK 模式") - # Ark 模式：使用 AsyncArk - base_url = config.get("base_url", "https://ark.cn-beijing.volces.com/api/v3") - os.environ["DOUBAO_BASE_URL"] = base_url - client = AsyncArk(base_url=base_url, api_key=api_key) - set_ark_client(client) - logger.info("豆包 Ark 客户端已创建 (base_url=%s)", base_url) - # 委托给通用运行时创建 Worker 任务 - logger.info("豆包插件初始化完成，创建 %d 个 Worker 副本", count) - return _create_workers(session, config, count, worker_manager) - -async def cleanup(worker_item: str): - """清理资源""" - logger.info("豆包插件清理: %s", worker_item) - cleanup_all() - logger.info("豆包插件清理完毕: %s", worker_item) \ No newline at end of file +# 暴露给 app lifespan 的模块级入口（bound method，签名与旧写法完全一致） +plugin = DoubaoPlugin() +create_workers = plugin.create_workers +cleanup = plugin.cleanup diff --git a/example/README.md b/example/README.md index 18a1033..b2e1ca2 100644 --- a/example/README.md +++ b/example/README.md @@ -54,7 +54,8 @@ curl "http://localhost:8000/api/v1/calc/count" 1. **配置**：`config/workers/example/*.toml` — 定义每个端点的参数和响应结构 2. **模型**：`models/pydantic.py` — 由 `gen_models` 根据 TOML 自动生成 3. **路由**：`router/` 下每个文件实现一个 RPC 端点，使用 `@rpc_method` 装饰器 -4. **工厂**：`__init__.py` 中的 `create_workers` 初始化数据库并启动 Worker +4. **入口**：`__init__.py` 的 `ExamplePlugin`（Plugin 基类）统一生命周期——`on_init` 初始化 + 数据库并发布到 global_var，`create_workers` / `cleanup` 为其 bound method --- diff --git a/example/router/add_mqtt.py b/example/router/add_mqtt.py index 0ec598f..a8ef2d4 100644 --- a/example/router/add_mqtt.py +++ b/example/router/add_mqtt.py @@ -2,7 +2,7 @@ MQTT 端点处理器 — 通过 MQTT 主题 chongming/example/add 调用 与 GET/POST 的 add 共用同一个 subject (example.add)， -通过 _method="MQTT" 区分。 +通过信封 meta.method="MQTT" 区分。 """ import time