From 0049045eadfec49ce4259473777456a2cbddd88a Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Fri, 7 Aug 2026 09:39:29 +0800 Subject: [PATCH] =?UTF-8?q?modify:=20mqtt=E9=A2=9D=E5=A4=96=E4=B8=BB?= =?UTF-8?q?=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 18 ++ config.toml | 5 + src/chongming/app/core/config_.py | 10 ++ src/chongming/app/core/lifespan_setup.py | 37 +++- src/chongming/app/core/mqtt_bridge.py | 79 ++++++++- test_mqtt_custom_topics.py | 205 +++++++++++++++++++++++ 6 files changed, 343 insertions(+), 11 deletions(-) create mode 100644 test_mqtt_custom_topics.py diff --git a/README.md b/README.md index d5b35e6..a38c3cd 100644 --- a/README.md +++ b/README.md @@ -203,6 +203,24 @@ async def upload(file: UploadFile = File(...)): | 请求 | `chongming//` | MQTT → Worker | | 响应 | `chongming///response` | Worker → MQTT（`None` 时跳过） | +### 自定义订阅主题 + +除内置的 `chongming//` 接口外，可将外部 MQTT 主题桥接到任意 Worker。**推荐在插件自己的配置** `config/workers//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` 缺省为 `<插件类型>.`；`response_topic` 缺省为 `<实际匹配主题>/response`。 +- 网关发现插件时自动收集所有 `[[workers.mqtt_topics]]`，MQTT Bridge 一并订阅并路由。 +- 请求负载同样支持 `_method` / `_timeout` / `_cacheable` 控制字段。 +- 多条自定义主题按配置顺序匹配，先匹配者优先；内置 `chongming/...` 接口最后兜底。 +- 全局 `[.mqtt].custom_topics` 仍可作为通用兜底（不依赖具体插件时使用，可选）。 + ### 快速测试 ```bash diff --git a/config.toml b/config.toml index ca4b063..ff76a9f 100644 --- a/config.toml +++ b/config.toml @@ -46,6 +46,11 @@ enabled = true host = "0.0.0.0" port = 1883 +# MQTT Bridge 额外订阅主题推荐声明在各插件自己的配置里： +# config/workers//config.toml → [[workers.mqtt_topics]] +# （如 aibox_sophon_adapter 插件的 /board_ping 心跳主题） +# 全局 [.mqtt].custom_topics 仍可作为通用兜底（可选，不推荐）。 + [development.mqtt.broker] # amqtt Broker 可选配置，默认允许匿名连接 # 详细参数参考: https://github.com/Yakifo/amqtt diff --git a/src/chongming/app/core/config_.py b/src/chongming/app/core/config_.py index 1b35692..dc343f5 100644 --- a/src/chongming/app/core/config_.py +++ b/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}... 占位符；默认 "." + 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): diff --git a/src/chongming/app/core/lifespan_setup.py b/src/chongming/app/core/lifespan_setup.py index fec840c..5d66ad0 100644 --- a/src/chongming/app/core/lifespan_setup.py +++ b/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 则跳过。 + 订阅主题 = 全局 [.mqtt].custom_topics（可选兜底） + + 各插件在其 config/workers//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})") diff --git a/src/chongming/app/core/mqtt_bridge.py b/src/chongming/app/core/mqtt_bridge.py index 12b78fd..aa2a9d1 100644 --- a/src/chongming/app/core/mqtt_bridge.py +++ b/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)]) - logger.info(f"已订阅 {MQTT_SUBSCRIBE_TOPIC}") + topics = [(MQTT_SUBSCRIBE_TOPIC, QOS_1)] + 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")) - except json.JSONDecodeError: return - pl.setdefault("_method", "GET") + await self._route(wt, subj, pl, rt) + + 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 默认 "."，与内置 <插件>.<操作> 约定一致 + 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, diff --git a/test_mqtt_custom_topics.py b/test_mqtt_custom_topics.py new file mode 100644 index 0000000..72cc3a5 --- /dev/null +++ b/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 = .<最后一段>；响应 = <实际主题>/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//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())