modify: mqtt额外主题

README.md

@@ -203,6 +203,24 @@ async def upload(file: UploadFile = File(...)):
 | 请求 | `chongming/<type>/<op>`          | MQTT → Worker                  |
 | 响应 | `chongming/<type>/<op>/response` | Worker → MQTT（`None` 时跳过） |
 
+### 自定义订阅主题
+
+除内置的 `chongming/<type>/<op>` 接口外，可将外部 MQTT 主题桥接到任意 Worker。**推荐在插件自己的配置** `config/workers/<type>/config.toml` 中通过 `[[workers.mqtt_topics]]` 声明（目标 worker 即该插件本身）：
+
+```toml
+# config/workers/aibox_sophon_adapter/config.toml
+[[workers.mqtt_topics]]
+topic = "/board_ping"                       # 支持 +（单级）/ #（多级，须在末尾）通配符
+subject = "aibox_sophon_adapter.board_ping" # 可选，{1} {2}... 对应通配符捕获值
+response_topic = "/board_ping/response"     # 可选，响应主题；默认 "<实际主题>/response"
+```
+
+- `subject` 缺省为 `<插件类型>.<topic 最后一段>`；`response_topic` 缺省为 `<实际匹配主题>/response`。
+- 网关发现插件时自动收集所有 `[[workers.mqtt_topics]]`，MQTT Bridge 一并订阅并路由。
+- 请求负载同样支持 `_method` / `_timeout` / `_cacheable` 控制字段。
+- 多条自定义主题按配置顺序匹配，先匹配者优先；内置 `chongming/...` 接口最后兜底。
+- 全局 `[<env>.mqtt].custom_topics` 仍可作为通用兜底（不依赖具体插件时使用，可选）。
+
 ### 快速测试
 
 ```bash

config.toml

@@ -46,6 +46,11 @@ enabled = true
 host = "0.0.0.0"
 port = 1883
 
+# MQTT Bridge 额外订阅主题推荐声明在各插件自己的配置里：
+#   config/workers/<type>/config.toml  →  [[workers.mqtt_topics]]
+# （如 aibox_sophon_adapter 插件的 /board_ping 心跳主题）
+# 全局 [<env>.mqtt].custom_topics 仍可作为通用兜底（可选，不推荐）。
+
 [development.mqtt.broker]
 # amqtt Broker 可选配置，默认允许匿名连接
 # 详细参数参考: https://github.com/Yakifo/amqtt

src/chongming/app/core/config_.py

@@ -102,6 +102,15 @@ class Cors(TypedDict, total=False):
     allow_headers: List[str]
 
 
+class CustomMQTTTopic(TypedDict, total=False):
+    """额外自定义 MQTT 订阅主题的路由配置"""
+
+    topic: Required[str]  # 订阅主题，支持 +（单级）/ #（多级）通配符
+    worker: Required[str]  # 目标 worker 类型
+    subject: str  # 目标 Zenoh subject，支持 {1} {2}... 占位符；默认 "<worker>.<topic 最后一段>"
+    response_topic: str  # 响应发布主题，支持占位符；默认 "<实际匹配主题>/response"
+
+
 class MQTTConfig(TypedDict, total=False):
     """MQTT Bridge 配置（按环境区分）"""
 
@@ -109,6 +118,7 @@ class MQTTConfig(TypedDict, total=False):
     host: str
     port: int
     broker: dict
+    custom_topics: List[CustomMQTTTopic]
 
 
 class ModuleSystem(TypedDict, total=False):

src/chongming/app/core/lifespan_setup.py

@@ -244,12 +244,24 @@ def _discover_plugins() -> list[dict]:
         if not workers_section:
             continue
 
+        plugin_type = workers_section.get("type", plugin_type)
+        # 插件声明的 MQTT Bridge 额外订阅主题（[[workers.mqtt_topics]]）
+        # 网关级路由配置，不传给插件运行时；worker 目标即本插件类型。
+        mqtt_topics = []
+        for t in workers_section.get("mqtt_topics", []) or []:
+            if not isinstance(t, dict) or not t.get("topic"):
+                continue
+            t = dict(t)
+            t.setdefault("worker", plugin_type)
+            mqtt_topics.append(t)
+
         # 合并运行时 config:
         #   1. plugin_conf 顶层 (worker, worker.registration 等 gen_models 用)
         #   2. [workers] 节的注册字段展平 (type, module, factory, cleanup, count)
         #      — 其中 type 被 chongming_worker.runtime.create_workers 用于匹配 RPC 方法
         #   3. [workers.config] 展平到顶层 (api_key, data_dir, heartbeat 等)
-        workers_flat = {k: v for k, v in workers_section.items() if k != "config"}
+        workers_flat = {k: v for k, v in workers_section.items()
+                        if k not in ("config", "mqtt_topics")}
         runtime_config = {
             **plugin_conf,
             **workers_flat,
@@ -257,11 +269,12 @@ def _discover_plugins() -> list[dict]:
         }
 
         plugins.append({
-            "type": workers_section.get("type", plugin_type),
+            "type": plugin_type,
             "module": workers_section.get("module", f"plugins.{plugin_type}"),
             "factory": workers_section.get("factory", "create_workers"),
             "cleanup": workers_section.get("cleanup", "cleanup"),
             "count": workers_section.get("count", 1),
+            "mqtt_topics": mqtt_topics,
             "config": runtime_config,
         })
 
@@ -301,6 +314,13 @@ async def load_workers(app: FastAPI, session: zenoh.Session, manager: WorkerMana
     workers_conf = _discover_plugins()
     logger.info(f"发现 {len(workers_conf)} 个 worker 插件")
 
+    # 汇总各插件声明的 MQTT Bridge 额外订阅主题，供 setup_mqtt 使用
+    app.state.plugin_mqtt_topics = [
+        topic
+        for wc in workers_conf
+        for topic in (wc.get("mqtt_topics") or [])
+    ]
+
     worker_tasks = []
     custom_routers = []
     for wc in workers_conf:
@@ -401,7 +421,9 @@ async def setup_mqtt(app: FastAPI, worker_manager: WorkerManager) -> MQTTBridge
     """
     启动 MQTT Bridge（客户端模式，Broker 由子进程管理）。
 
-    从 config.toml 的环境段读取配置，若 enabled 为 false 则跳过。
+    从 config.toml 的环境段读取 host/port/enabled；若 enabled 为 false 则跳过。
+    订阅主题 = 全局 [<env>.mqtt].custom_topics（可选兜底）
+             + 各插件在其 config/workers/<type>/config.toml 中声明的 [[workers.mqtt_topics]]。
     使用文件锁确保多个 gunicorn worker 中只有一个启动 Bridge。
     """
     env_config = config.get(env, {})
@@ -412,6 +434,8 @@ async def setup_mqtt(app: FastAPI, worker_manager: WorkerManager) -> MQTTBridge
 
     host = mqtt_config.get("host", "localhost")
     port = mqtt_config.get("port", 1883)
+    custom_topics = list(mqtt_config.get("custom_topics", []) or [])
+    custom_topics += list(getattr(app.state, "plugin_mqtt_topics", []) or [])
 
     # 1. 启动 Broker 子进程（文件锁已确保只有一个）
     setup_mqtt_broker(app, host, port)
@@ -421,7 +445,12 @@ async def setup_mqtt(app: FastAPI, worker_manager: WorkerManager) -> MQTTBridge
         logger.info("MQTT Bridge 已由其他 worker 启动，跳过")
         return None
 
-    bridge = MQTTBridge(worker_manager=worker_manager, host=host, port=port)
+    bridge = MQTTBridge(
+        worker_manager=worker_manager,
+        host=host,
+        port=port,
+        custom_topics=custom_topics,
+    )
     app.state.mqtt_bridge = bridge
     await bridge.start()
     logger.info(f"MQTT Bridge 已启动 ({host}:{port})")

src/chongming/app/core/mqtt_bridge.py

@@ -1,5 +1,5 @@
 """MQTT Bridge — 纯客户端模式，Broker 由子进程管理"""
-import asyncio, json
+import asyncio, json, re
 from typing import Any, Optional
 from amqtt.client import MQTTClient
 from amqtt.mqtt.constants import QOS_1
@@ -8,12 +8,42 @@ logger = get_logger("mqtt")
 MQTT_TOPIC_PREFIX = "chongming"
 MQTT_SUBSCRIBE_TOPIC = f"{MQTT_TOPIC_PREFIX}/+/+"
 
+
+def _compile_topic(topic: str) -> re.Pattern:
+    """
+    将 MQTT 主题模板（支持 + / # 通配符）编译为正则表达式。
+
+    - ``+`` 匹配单级（编译为捕获组 ``([^/]+)``）
+    - ``#`` 匹配多级（编译为捕获组 ``(.*)``，须位于末尾）
+    捕获组按出现顺序对应，供 {1} {2}... 占位符替换使用。
+    """
+    parts = topic.split("/")
+    regex = "^"
+    for i, part in enumerate(parts):
+        if i > 0:
+            regex += "/"
+        if part == "#":
+            regex += "(.*)"
+        elif part == "+":
+            regex += "([^/]+)"
+        else:
+            regex += re.escape(part)
+    regex += "$"
+    return re.compile(regex)
+
+
 class MQTTBridge:
-    def __init__(self, worker_manager, host="localhost", port=1883):
+    def __init__(self, worker_manager, host="localhost", port=1883, custom_topics=None):
         self._wm = worker_manager; self._host = host; self._port = port
         self._client: Optional[MQTTClient] = None
         self._lt: Optional[asyncio.Task] = None
         self._running = False
+        # custom_topics: [{"topic","worker","subject"?,"response_topic"?}, ...]
+        self._custom_topics = custom_topics or []
+        self._compiled = [
+            (ct, _compile_topic(ct.get("topic", "")))
+            for ct in self._custom_topics if ct.get("topic")
+        ]
 
     async def start(self):
         if self._running: return
@@ -31,8 +61,11 @@ class MQTTBridge:
     async def _sc(self):
         self._client = MQTTClient(config={"auto_reconnect": False})
         await self._client.connect(f"mqtt://{self._host}:{self._port}/")
-        await self._client.subscribe([(MQTT_SUBSCRIBE_TOPIC, QOS_1)])
+        topics = [(MQTT_SUBSCRIBE_TOPIC, QOS_1)]
-        logger.info(f"已订阅 {MQTT_SUBSCRIBE_TOPIC}")
+        topics += [(ct["topic"], QOS_1) for ct, _ in self._compiled]
+        await self._client.subscribe(topics)
+        logger.info(f"已订阅 {MQTT_SUBSCRIBE_TOPIC}"
+                   + (f" 及 {len(self._compiled)} 个自定义主题" if self._compiled else ""))
         self._lt = asyncio.create_task(self._loop())
 
     async def _sc_stop(self):
@@ -62,14 +95,46 @@ class MQTTBridge:
 
     async def _handle(self, m):
         t: str = m.topic; p = m.data  # type: ignore
+        try: pl = json.loads(p.decode("utf-8"))
+        except json.JSONDecodeError: return
+        pl.setdefault("_method", "GET")
+
+        # 1. 优先匹配显式配置的自定义订阅主题（顺序即优先级）
+        for ct, compiled in self._compiled:
+            matched = compiled.match(t)
+            if matched:
+                await self._route_custom(ct, matched, pl, t)
+                return
+
+        # 2. 内置接口 chongming/{worker}/{operation}
         parts = t.split("/")
         if len(parts) < 3: return
         wt, op = parts[1], parts[2]
         subj = f"{wt}.{op}"
         rt = f"{MQTT_TOPIC_PREFIX}/{wt}/{op}/response"
-        try: pl = json.loads(p.decode("utf-8"))
+        await self._route(wt, subj, pl, rt)
-        except json.JSONDecodeError: return
+
-        pl.setdefault("_method", "GET")
+    def _fill(self, template: str, matched: re.Match) -> str:
+        """用匹配到的通配符值填充模板中的 {1} {2}... 占位符"""
+        if "{" not in template: return template
+        for i, g in enumerate(matched.groups(), start=1):
+            template = template.replace("{" + str(i) + "}", g)
+        return template
+
+    async def _route_custom(self, ct: dict, matched: re.Match, pl: dict, topic: str):
+        worker = ct.get("worker", "")
+        # subject 默认 "<worker>.<topic 模板最后一段>"，与内置 <插件>.<操作> 约定一致
+        subject_tpl = ct.get("subject")
+        if not subject_tpl:
+            last = ct["topic"].split("/")[-1]
+            subject_tpl = f"{worker}.{last}" if last not in ("+", "#") else worker
+        subject = self._fill(subject_tpl, matched)
+        # response_topic 默认 "<实际匹配主题>/response"
+        rt_tpl = ct.get("response_topic")
+        rt = self._fill(rt_tpl, matched) if rt_tpl else f"{topic}/response"
+        await self._route(worker, subject, pl, rt)
+
+    async def _route(self, wt: str, subj: str, pl: dict, rt: str):
         logger.info(f"MQTT -> {subj}")
         try:
             r = await self._wm.call(worker_type=wt, subject=subj, payload=pl,

test_mqtt_custom_topics.py

@@ -0,0 +1,205 @@
+"""
+MQTT Bridge 自定义订阅主题单元测试
+
+验证:
+  1. _compile_topic 对 + / # 通配符的匹配与捕获
+  2. 自定义主题路由: 主题匹配 → worker/subject/response_topic 解析 → worker_manager.call 调用
+  3. 内置 chongming/... 主题兜底逻辑不变
+"""
+import asyncio
+import json
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+sys.path.insert(0, str(Path(__file__).parent / "src"))
+
+from chongming.app.core.mqtt_bridge import (
+    MQTTBridge,
+    MQTT_TOPIC_PREFIX,
+    _compile_topic,
+)
+
+
+def make_msg(topic, payload):
+    m = MagicMock()
+    m.topic = topic
+    m.data = json.dumps(payload).encode("utf-8")
+    return m
+
+
+class FakeWM:
+    def __init__(self):
+        self.calls = []
+        self.result = {"result": "ok"}
+
+    async def call(self, worker_type, subject, payload, timeout=30, cacheable=False):
+        self.calls.append(
+            {
+                "worker_type": worker_type,
+                "subject": subject,
+                "payload": payload,
+                "timeout": timeout,
+                "cacheable": cacheable,
+            }
+        )
+        return self.result
+
+
+def make_bridge(wm, custom_topics):
+    """构造一个不真正连接 Broker 的 Bridge，并捕获发布消息"""
+    bridge = MQTTBridge(worker_manager=wm, custom_topics=custom_topics)
+    bridge._client = MagicMock()
+    bridge._running = True
+    published = []
+
+    async def fake_publish(topic, data, qos=None):
+        published.append((topic, data.decode("utf-8")))
+
+    bridge._client.publish = fake_publish
+    return bridge, published
+
+
+def test_compile_topic():
+    # 单级通配符
+    p = _compile_topic("devices/+/telemetry")
+    m = p.match("devices/dev-01/telemetry")
+    assert m and m.groups() == ("dev-01",)
+    assert not p.match("devices/a/b/telemetry")
+
+    # 多级通配符
+    p = _compile_topic("data/#")
+    m = p.match("data/a/b/c")
+    assert m and m.groups() == ("a/b/c",)
+
+    # 无通配符
+    p = _compile_topic("fixed/topic")
+    assert p.match("fixed/topic")
+    assert not p.match("fixed/topic/x")
+
+    # 前缀必须完整匹配
+    p = _compile_topic("devices/+/telemetry")
+    assert not p.match("xdevices/a/telemetry")
+    print("✅ _compile_topic 通配符编译/匹配通过")
+
+
+async def test_custom_routing():
+    wm = FakeWM()
+    bridge, published = make_bridge(
+        wm,
+        [
+            {"topic": "devices/+/telemetry", "worker": "device"},
+            {
+                "topic": "foo/#",
+                "worker": "foo",
+                "subject": "foo.handle.{1}",
+                "response_topic": "out/{1}/resp",
+            },
+        ],
+    )
+
+    # 1. 默认 subject = <worker>.<最后一段>；响应 = <实际主题>/response
+    await bridge._handle(make_msg("devices/dev-01/telemetry", {"v": 1}))
+    assert wm.calls[-1]["worker_type"] == "device"
+    assert wm.calls[-1]["subject"] == "device.telemetry"
+    assert published[-1][0] == "devices/dev-01/telemetry/response"
+
+    # 2. 带占位符 + 自定义响应主题
+    await bridge._handle(make_msg("foo/a/b", {"v": 2}))
+    assert wm.calls[-1]["worker_type"] == "foo"
+    assert wm.calls[-1]["subject"] == "foo.handle.a/b"
+    assert published[-1][0] == "out/a/b/resp"
+
+    # 3. 内置 chongming 兜底
+    await bridge._handle(
+        make_msg(f"{MQTT_TOPIC_PREFIX}/example/add", {"a": 1, "b": 2})
+    )
+    assert wm.calls[-1]["worker_type"] == "example"
+    assert wm.calls[-1]["subject"] == "example.add"
+    assert published[-1][0] == f"{MQTT_TOPIC_PREFIX}/example/add/response"
+    print("✅ 自定义主题路由 + 内置兜底通过")
+
+
+async def test_custom_routing_control_fields():
+    wm = FakeWM()
+    bridge, published = make_bridge(
+        wm, [{"topic": "devices/+/telemetry", "worker": "device"}]
+    )
+    await bridge._handle(
+        make_msg(
+            "devices/d1/telemetry",
+            {"_timeout": 5, "_cacheable": True, "_method": "POST"},
+        )
+    )
+    assert wm.calls[-1]["timeout"] == 5.0
+    assert wm.calls[-1]["cacheable"] is True
+    assert "_timeout" not in wm.calls[-1]["payload"]
+    assert wm.calls[-1]["payload"].get("_method") == "POST"
+    print("✅ 控制字段 (_timeout/_cacheable/_method) 透传通过")
+
+
+async def test_non_json_dropped():
+    wm = FakeWM()
+    bridge, published = make_bridge(
+        wm, [{"topic": "devices/+/telemetry", "worker": "device"}]
+    )
+    bad = MagicMock()
+    bad.topic = "devices/d1/telemetry"
+    bad.data = b"not-json"
+    await bridge._handle(bad)
+    assert wm.calls == []
+    assert published == []
+    print("✅ 非 JSON 负载被丢弃通过")
+
+
+async def test_board_ping_topic():
+    """aibox_sophon_adapter 插件：设备发布到 /board_ping → 路由到 aibox_sophon_adapter.board_ping"""
+    wm = FakeWM()
+    bridge, published = make_bridge(
+        wm,
+        [
+            {
+                "topic": "/board_ping",
+                "worker": "aibox_sophon_adapter",
+                "subject": "aibox_sophon_adapter.board_ping",
+            }
+        ],
+    )
+    await bridge._handle(make_msg("/board_ping", {"BoardId": "B001", "Status": 1}))
+    assert wm.calls[-1]["worker_type"] == "aibox_sophon_adapter"
+    assert wm.calls[-1]["subject"] == "aibox_sophon_adapter.board_ping"
+    assert published[-1][0] == "/board_ping/response"
+    print("✅ /board_ping 自定义主题路由通过")
+
+
+def test_plugin_config_driven_collection():
+    """插件在其 config/workers/<type>/config.toml 中声明 [[workers.mqtt_topics]]，
+    网关 _discover_plugins 自动收集，并默认 worker 为该插件类型。"""
+    from chongming.app.core.lifespan_setup import _discover_plugins
+
+    plugins = _discover_plugins()
+    by_type = {p["type"]: p for p in plugins}
+    assert "aibox_sophon_adapter" in by_type, "未发现 aibox_sophon_adapter 插件"
+    topics = by_type["aibox_sophon_adapter"]["mqtt_topics"]
+    assert any(
+        t["topic"] == "/board_ping" and t["worker"] == "aibox_sophon_adapter"
+        and t["subject"] == "aibox_sophon_adapter.board_ping"
+        for t in topics
+    ), topics
+    # 不应泄漏到插件运行时配置
+    assert "mqtt_topics" not in by_type["aibox_sophon_adapter"]["config"]
+    print("✅ 插件级 [[workers.mqtt_topics]] 收集通过")
+
+
+async def main():
+    test_compile_topic()
+    await test_custom_routing()
+    await test_custom_routing_control_fields()
+    await test_non_json_dropped()
+    await test_board_ping_topic()
+    test_plugin_config_driven_collection()
+    print("\n🎉 全部自定义订阅主题测试通过")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())