From 983afa6957dd2fddd8eed9298e8ef40918965ae0 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Tue, 18 Aug 2026 12:31:20 +0800 Subject: [PATCH] =?UTF-8?q?modify:=20MQTT=20Bridge=20=E5=85=BC=E5=AE=B9?= =?UTF-8?q?=E9=9D=9E=E4=BF=A1=E5=B0=81=E5=8E=9F=E5=A7=8B=E6=8A=A5=E6=96=87?= =?UTF-8?q?=EF=BC=88=E7=9B=92=E5=AD=90=E5=BF=83=E8=B7=B3=EF=BC=89=E2=80=94?= =?UTF-8?q?=E2=80=94=E4=B8=8D=E5=86=8D=E4=B8=A2=E5=BC=83=EF=BC=8C=E6=95=B4?= =?UTF-8?q?=E5=8C=85=E8=A7=86=E4=B8=BA=20data=20=E8=B7=AF=E7=94=B1?= =?UTF-8?q?=EF=BC=8C=E4=B8=8E=20worker=20=E8=BF=90=E8=A1=8C=E6=97=B6?= =?UTF-8?q?=E5=AE=B9=E9=94=99=E8=AF=AD=E4=B9=89=E4=B8=80=E8=87=B4=EF=BC=9B?= =?UTF-8?q?=E6=AF=8F=E4=B8=BB=E9=A2=98=E4=BB=85=E5=91=8A=E8=AD=A6=E4=B8=80?= =?UTF-8?q?=E6=AC=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/chongming/app/core/mqtt_bridge.py | 33 ++++++++++++++++++--------- test_mqtt_custom_topics.py | 23 +++++++++++++------ 2 files changed, 38 insertions(+), 18 deletions(-) diff --git a/src/chongming/app/core/mqtt_bridge.py b/src/chongming/app/core/mqtt_bridge.py index 66179a1..f14eb79 100644 --- a/src/chongming/app/core/mqtt_bridge.py +++ b/src/chongming/app/core/mqtt_bridge.py @@ -12,6 +12,9 @@ logger = get_logger("mqtt") MQTT_TOPIC_PREFIX = "chongming" MQTT_SUBSCRIBE_TOPIC = f"{MQTT_TOPIC_PREFIX}/+/+" +# 已按"非信封载荷"路由过告警的主题（每主题只告警一次，避免设备 5s 心跳刷屏） +_warned_legacy_topics: set = set() + def _compile_topic(topic: str) -> re.Pattern: """ @@ -154,18 +157,26 @@ class MQTTBridge: except json.JSONDecodeError: return - # 信封协议（严格）: {"meta": {...}, "data": {...}} - # 非信封格式载荷直接丢弃并告警，请设备侧按新协议发布 + # 信封协议: {"meta": {...}, "data": {...}} + # 兼容非信封载荷（如第三方设备原始报文）: 整包视为业务数据、meta 置空， + # 与 worker 运行时 runtime.py 的容错语义一致（method 默认 GET）—— + # 若在此丢弃，盒子心跳等设备报文将永远到不了 Worker。 if not isinstance(pl, dict) or ("meta" not in pl and "data" not in pl): - logger.warning( - "丢弃非信封格式载荷 [%s]: 请使用 {'meta': {...}, 'data': {...}} 格式", t - ) - return - meta = pl.get("meta") or {} - data = pl.get("data") or {} - if not isinstance(data, dict): - data = {} - pl = {"meta": meta, "data": data} + data = pl if isinstance(pl, dict) else {} + pl = {"meta": {}, "data": data} + if t not in _warned_legacy_topics: + _warned_legacy_topics.add(t) + logger.warning( + "非信封格式载荷（缺少 meta/data）按原始报文路由 [%s]: " + "整包视为 data，建议设备侧改用 {'meta': {...}, 'data': {...}} 格式", + t, + ) + else: + meta = pl.get("meta") or {} + data = pl.get("data") or {} + if not isinstance(data, dict): + data = {} + pl = {"meta": meta, "data": data} # 1. 优先匹配显式配置的自定义订阅主题（顺序即优先级） for ct, compiled in self._compiled: diff --git a/test_mqtt_custom_topics.py b/test_mqtt_custom_topics.py index 10ca44c..fc84d5d 100644 --- a/test_mqtt_custom_topics.py +++ b/test_mqtt_custom_topics.py @@ -139,16 +139,18 @@ async def test_envelope_meta_fields(): print("✅ 信封 meta 字段 (method/timeout/cacheable) 透传通过") -async def test_flat_payload_dropped(): - """严格信封协议：扁平载荷（无 meta/data）应被丢弃且不调用 worker""" +async def test_flat_payload_routed(): + """非信封（扁平）载荷兼容：整包视为 data、meta 置空照常路由（与 worker 运行时容错一致）""" wm = FakeWM() bridge, published = make_bridge( wm, [{"topic": "devices/+/telemetry", "worker": "device"}] ) await bridge._handle(make_msg("devices/d1/telemetry", {"_method": "POST", "v": 1})) - assert wm.calls == [] - assert published == [] - print("✅ 非信封（扁平）载荷被丢弃通过") + assert wm.calls, "扁平载荷不应被丢弃，应照常路由到 worker" + assert wm.calls[-1]["worker_type"] == "device" + assert wm.calls[-1]["payload"] == {"meta": {}, "data": {"_method": "POST", "v": 1}} + assert published[-1][0] == "devices/d1/telemetry/response" + print("✅ 非信封（扁平）载荷按原始报文路由通过") async def test_non_json_dropped(): @@ -182,7 +184,14 @@ async def test_board_ping_topic(): 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 自定义主题路由通过") + # 回归：盒子心跳为原始报文（无信封），不得被丢弃，应整包作为 data 路由 + wm.calls.clear(); published.clear() + await bridge._handle(make_msg("/board_ping", {"BoardId": "B001", "Status": 1})) + assert wm.calls and wm.calls[-1]["payload"] == { + "meta": {}, "data": {"BoardId": "B001", "Status": 1} + }, wm.calls + assert published[-1][0] == "/board_ping/response" + print("✅ /board_ping 自定义主题路由通过（含原始报文兼容）") def test_plugin_config_driven_collection(): @@ -209,7 +218,7 @@ async def main(): test_compile_topic() await test_custom_routing() await test_envelope_meta_fields() - await test_flat_payload_dropped() + await test_flat_payload_routed() await test_non_json_dropped() await test_board_ping_topic() test_plugin_config_driven_collection()