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()