From 5ca88d4de88a1f7ee79d881653801a0f32513c96 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Fri, 7 Aug 2026 14:25:52 +0800 Subject: [PATCH] first commit --- .gitignore | 72 ++++++++ README.md | 170 ++++++++++++++++++ pyproject.toml | 16 ++ src/chongming_mqtt/__init__.py | 19 ++ src/chongming_mqtt/internal_mqtt.py | 253 +++++++++++++++++++++++++++ src/chongming_mqtt/mqtt_forward.py | 120 +++++++++++++ src/chongming_mqtt/mqtt_subscribe.py | 253 +++++++++++++++++++++++++++ 7 files changed, 903 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 pyproject.toml create mode 100644 src/chongming_mqtt/__init__.py create mode 100644 src/chongming_mqtt/internal_mqtt.py create mode 100644 src/chongming_mqtt/mqtt_forward.py create mode 100644 src/chongming_mqtt/mqtt_subscribe.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c8f0442 --- /dev/null +++ b/.gitignore @@ -0,0 +1,72 @@ +/target + +# Byte-compiled / optimized / DLL files +__pycache__/ +.pytest_cache/ +*.py[cod] + +# C extensions +*.so + +# Distribution / packaging +.Python +.venv/ +env/ +bin/ +build/ +develop-eggs/ +dist/ +eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +include/ +man/ +venv/ +*.egg-info/ +.installed.cfg +*.egg + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt +pip-selfcheck.json + +# Unit test / coverage reports +htmlcov/ +.tox/ +.coverage +.cache +nosetests.xml +coverage.xml + +# Translations +*.mo + +# Mr Developer +.mr.developer.cfg +.project +.pydevproject + +# Rope +.ropeproject + +# Django stuff: +*.log +*.pot + +.DS_Store + +# Sphinx documentation +docs/_build/ + +# PyCharm +.idea/ + +# VSCode +.vscode/ + +# Pyenv +.python-version diff --git a/README.md b/README.md new file mode 100644 index 0000000..cfc2df9 --- /dev/null +++ b/README.md @@ -0,0 +1,170 @@ +# chongming-mqtt + +MQTT 工具包，封装基于 `amqtt` 的三类客户端，覆盖常见 MQTT 通信场景。 + +--- + +## 安装 + +```bash +uv pip install -e utils/mqtt +``` + +依赖：`amqtt>=0.11.3`，Python `>=3.12`。 + +--- + +## 客户端一览 + +| 客户端 | 方向 | 用途 | +| --------------------- | ----------- | ---------------------------------------------------------- | +| `MQTTForwardClient` | 单向发布 | 主动把数据上报/转发到外部 Broker | +| `MQTTSubscribeClient` | 订阅 + 回复 | 订阅外部主题，处理请求并按 `reply_topic` 回复（请求-响应） | +| `InternalMQTTClient` | 请求-响应 | 对任意应用进程发起 MQTT 请求-响应，显式指定请求/响应主题 | + +三者均支持： + +- 断线自动重连（无限重试，最大重连间隔 30s） +- URI percent-encoded 凭据解码（密码中的 `@` 写为 `%40`） +- `mqtt://user:pass@host:port` 连接串 + +--- + +## 1. MQTTForwardClient — 外部转发（单向发布） + +主动把数据推送到外部 Broker 的指定主题。 + +```python +from chongming_mqtt import MQTTForwardClient + +client = MQTTForwardClient(uri="mqtt://admin:Mixc%402021@host:48081", topic="/board_ping") +await client.start() +await client.publish({"BoardId": "B001", "Temp": 45.6}) # 发布到默认主题 +await client.publish({"x": 1}, topic="/other_topic") # 指定主题 +await client.stop() +``` + +| 方法 | 说明 | +| ------------------------------ | ------------------------------------- | +| `start()` | 建立长连接（自动重连） | +| `publish(payload, topic=None)` | 发布 JSON；`topic` 缺省用构造时的主题 | +| `stop()` | 断开连接 | + +--- + +## 2. MQTTSubscribeClient — 外部订阅（请求-响应） + +订阅外部主题，收到消息后调用绑定的处理函数；配置 `reply_topic` 后，把处理函数返回值发布到回复主题。 + +```python +from chongming_mqtt import MQTTSubscribeClient + +async def handler(payload: dict, topic: str): + # 处理请求，返回回复内容（返回 None 表示不回复） + return {"code": 0, "msg": "ok", "data": payload} + +client = MQTTSubscribeClient( + uri="mqtt://admin:Mixc%402021@host:48081", + routes=[ + { + "topic": "/edge_app_controller", # 订阅主题（支持 + / # 通配符） + "handler": handler, # async (payload, topic) -> dict | None + "reply_topic": "/edge_app_controller_reply", # 回复主题（支持 {1}{2} 占位符） + } + ], +) +await client.start() +# ... 运行中 ... +await client.stop() +``` + +- **不回复约定**：处理函数返回 `None` 时不发布回复（静默命令） +- **`reply_topic` 占位符**：`{1}`、`{2}`… 填充订阅主题通配符捕获值，如 `reply_topic="/dev/{1}/resp"` + +--- + +## 3. InternalMQTTClient — 内部请求-响应 + +对任意应用进程发起 MQTT 请求-响应：**显式指定"请求主题"与"响应主题"**，等待并返回响应报文。不依赖特定协议（如 `chongming/*`），适用于与其它应用/进程通信。 + +```python +from chongming_mqtt import InternalMQTTClient + +client = InternalMQTTClient(uri="mqtt://127.0.0.1:1883") # 默认连内部 broker +await client.start() + +resp = await client.request( + request_topic="/other_app/request", # 目标应用监听的请求主题 + response_topic="/other_app/reply/#", # 目标应用回复的主题（支持 +/# 或 {rid}） + payload={"BoardId": "B001"}, + timeout=30, +) +# resp: dict | None（None 表示超时/异常） +await client.stop() +``` + +### 响应关联策略 + +请求会自动注入 `request_id` 字段。收到响应时按以下优先级匹配： + +1. **对方回显 `request_id`** → 精确匹配（推荐） +2. `response_topic` 含 **`{rid}` 模板** → 订阅唯一主题匹配（对方把 request_id 拼进响应主题） +3. 兜底 **FIFO**（固定主题、不回显，串行场景） + +支持并发请求与响应乱序，互不串扰。 + +--- + +## 与 aibox_sophon_adapter 插件的集成 + +插件在 `[workers.config]` 下分别配置这三个客户端： + +```toml +[workers.config.mqtt_forward] # -> MQTTForwardClient +enabled = true +uri = "mqtt://admin:Mixc%402021@dg.datagoo.cn:48081" +topic = "/board_ping" + +[workers.config.mqtt_subscribe] # -> MQTTSubscribeClient +enabled = true +uri = "mqtt://admin:Mixc%402021@dg.datagoo.cn:48081" + [[workers.config.mqtt_subscribe.topics]] + topic = "/edge_app_controller" + handler = "edge_app_controller" + reply_topic = "/edge_app_controller_reply" + +[workers.config.internal_mqtt] # -> InternalMQTTClient（handler 内调用） +enabled = true +uri = "mqtt://127.0.0.1:1883" +``` + +handler 内使用 `InternalMQTTClient`： + +```python +# edge_app_controller.py +from chongming_mqtt import InternalMQTTClient + +async def _cmd_some_op(payload: dict) -> dict: + from aibox_sophon_adapter.global_var import internal_mqtt + if internal_mqtt is None: + return {"error": "内部 MQTT 未初始化"} + resp = await internal_mqtt.request( + request_topic="/other_app/request", + response_topic="/other_app/reply/#", + payload={"BoardId": payload.get("BoardId")}, + timeout=30, + ) + return resp or {"error": "内部请求超时"} +``` + +--- + +## 构建集成 + +若通过 PyInstaller/Nuitka 打包，需在 `config.toml` 的 `[pyinstaller_build_import] base` 中加入： + +```toml +"chongming_mqtt", +``` + +并确保 `chongming_mqtt` 已安装到目标 Python 环境（`uv pip install -e utils/mqtt`）。 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..8c6e47c --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,16 @@ +[build-system] +build-backend = "hatchling.build" +requires = ["hatchling"] + +[project] +name = "chongming-mqtt" +version = "0.1.0" +description = "chongming MQTT 工具包：外部转发、外部订阅、内部请求-响应" +readme = "README.md" +requires-python = ">=3.12" +dependencies = [ + "amqtt>=0.11.3", +] + +[tool.hatch.build.targets.wheel] +packages = ["src/chongming_mqtt"] diff --git a/src/chongming_mqtt/__init__.py b/src/chongming_mqtt/__init__.py new file mode 100644 index 0000000..aa0a524 --- /dev/null +++ b/src/chongming_mqtt/__init__.py @@ -0,0 +1,19 @@ +""" +chongming_mqtt - MQTT 工具包 + +提供三类 MQTT 客户端： + +- ``MQTTForwardClient`` 外部 MQTT 转发客户端（单向发布，主动上报数据） +- ``MQTTSubscribeClient`` 外部 MQTT 订阅客户端（订阅主题 + 请求-响应，收外部请求并回复） +- ``InternalMQTTClient`` 内部 MQTT 请求-响应客户端（handler 内对任意应用进程 + 发起请求，显式指定请求/响应主题） +""" +from chongming_mqtt.mqtt_forward import MQTTForwardClient +from chongming_mqtt.mqtt_subscribe import MQTTSubscribeClient +from chongming_mqtt.internal_mqtt import InternalMQTTClient + +__all__ = [ + "MQTTForwardClient", + "MQTTSubscribeClient", + "InternalMQTTClient", +] diff --git a/src/chongming_mqtt/internal_mqtt.py b/src/chongming_mqtt/internal_mqtt.py new file mode 100644 index 0000000..73df7b2 --- /dev/null +++ b/src/chongming_mqtt/internal_mqtt.py @@ -0,0 +1,253 @@ +""" +AI Box Sophon Adapter — 内部 MQTT 请求-响应客户端 + +建立到内部（或目标）MQTT Broker 的长连接，提供通用请求-响应能力： +调用方**显式传入"请求主题"与"响应主题"**，与任意应用进程通信 +（不依赖 chongming 协议，响应主题由对方应用发布）。 + +支持: + - 断线自动重连（无限重试） + - URI percent-encoded 凭据解码（如密码中的 @ 写为 %40） + - 请求时自动注入 request_id 用于关联回复（对方回显则按 request_id 匹配） + - 响应主题支持通配符（+ / #）或 {rid} 模板（对方把 request_id 拼进响应主题） + - 每个请求独立超时控制，并发安全 + +使用方式: + client = InternalMQTTClient(uri="mqtt://127.0.0.1:1883") + await client.start() + resp = await client.request( + request_topic="/app/request", + response_topic="/app/reply/#", # 或 "/app/reply/{rid}" + payload={"key": "value"}, + timeout=30, + ) + await client.stop() +""" +import asyncio +import json +import logging +import uuid +from urllib.parse import urlparse, unquote + +from amqtt.client import MQTTClient +from amqtt.mqtt.constants import QOS_1 + +logger = logging.getLogger(__name__) + +# 关联 ID 字段名（注入请求负载，供对方回显匹配） +RID_KEY = "request_id" + + +class InternalMQTTClient: + """ + 内部 MQTT 请求-响应客户端 + + 连接目标 MQTT Broker（默认内部 127.0.0.1:1883）， + 提供 request() 方法：发布到显式指定的请求主题，订阅显式指定的响应主题， + 等待并返回响应报文。 + """ + + def __init__(self, uri: str = "mqtt://127.0.0.1:1883"): + self._uri = uri + self._client: MQTTClient | None = None + self._loop_task: asyncio.Task | None = None + self._running = False + + # 关联状态 + self._pending: dict[str, asyncio.Future] = {} # rid -> future + self._rid_topics: dict[str, str] = {} # rid -> 唯一订阅主题（{rid} 模板） + self._subscribed: set[str] = set() # 已订阅主题（避免重复订阅） + + # 预解析 URI 并解码凭据（amqtt 的 urlparse 不会自动 unquote） + parsed = urlparse(uri) + self._clean_uri = f"{parsed.scheme}://{parsed.hostname}:{parsed.port or 1883}" + self._username: str | None = unquote(parsed.username) if parsed.username else None + self._password: str | None = unquote(parsed.password) if parsed.password else None + + async def start(self) -> None: + """建立到目标 MQTT Broker 的连接""" + if self._running: + return + + try: + self._client = MQTTClient( + config={ + "auto_reconnect": True, + "reconnect_retries": -1, # 无限重连 + "reconnect_max_interval": 30, # 最大重连间隔（秒） + "keep_alive": 60, + "connection_timeout": 10, # 连接超时（秒），避免永久挂起 + } + ) + # 手动初始化 session 并设置解码后的凭据（绕过 amqtt urlparse 不解码 %40 的问题） + self._client.session = self._client._init_session(self._clean_uri) + if self._username: + self._client.session.username = self._username + if self._password: + self._client.session.password = self._password + await self._client._do_connect() + + self._running = True + self._loop_task = asyncio.create_task(self._loop()) + logger.info(f"[InternalMQTT] 已连接: {self._clean_uri}") + except Exception as e: + logger.error(f"[InternalMQTT] 连接失败: {e}") + if self._client: + try: + await self._client.disconnect() + except Exception: + pass + self._client = None + raise + + async def stop(self) -> None: + """断开连接，并让所有挂起请求返回异常""" + if not self._running: + return + + self._running = False + if self._client: + try: + await self._client.disconnect() + except Exception as e: + logger.warning(f"[InternalMQTT] 断开连接时异常: {e}") + finally: + self._client = None + if self._loop_task and not self._loop_task.done(): + self._loop_task.cancel() + try: + await self._loop_task + except asyncio.CancelledError: + pass + + for fut in self._pending.values(): + if not fut.done(): + fut.set_exception(RuntimeError("内部 MQTT 客户端已停止")) + self._pending.clear() + self._rid_topics.clear() + self._subscribed.clear() + logger.info("[InternalMQTT] 已断开连接") + + async def request( + self, + request_topic: str, + response_topic: str, + payload: dict, + timeout: float = 30.0, + ) -> dict | None: + """ + 发起 MQTT 请求-响应（显式指定请求/响应主题） + + Args: + request_topic: 要发布请求的目标主题（对方应用订阅处理） + response_topic: 要监听的响应主题； + 支持 + / # 通配符；若含 {rid} 模板则替换为本次 request_id + （对方把 request_id 拼进响应主题时使用） + payload: 请求报文（会自动注入 request_id 字段） + timeout: 等待回复的超时（秒） + + Returns: + 响应报文（dict）；超时或异常返回 None + """ + if not self._running or self._client is None: + logger.warning("[InternalMQTT] 客户端未连接，跳过请求") + return None + + rid = uuid.uuid4().hex + req_payload = dict(payload) + req_payload[RID_KEY] = rid + + loop = asyncio.get_running_loop() + fut: asyncio.Future = loop.create_future() + self._pending[rid] = fut + + try: + # 1) 解析响应主题：{rid} 模板 → 唯一主题；否则原样（固定/通配，共享订阅） + resp_topic = response_topic + if "{rid}" in response_topic: + resp_topic = response_topic.replace("{rid}", rid) + self._rid_topics[rid] = resp_topic + await self._subscribe_once(resp_topic) + + # 2) 发布请求 + data = json.dumps(req_payload, ensure_ascii=False, default=str) + await self._client.publish(request_topic, data.encode("utf-8"), QOS_1) + logger.debug( + f"[InternalMQTT] 请求 {rid} -> {request_topic}，响应订阅: {resp_topic}" + ) + + # 3) 等待回复（超时） + return await asyncio.wait_for(fut, timeout) + except asyncio.TimeoutError: + logger.warning( + f"[InternalMQTT] 请求 {rid} 超时（{timeout}s）: {request_topic}" + ) + return None + except Exception as e: + logger.error(f"[InternalMQTT] 请求 {rid} 异常: {e}") + return None + finally: + self._pending.pop(rid, None) + self._rid_topics.pop(rid, None) + + async def _subscribe_once(self, topic: str) -> None: + """订阅主题（已订阅则跳过，供多个并发请求共享同一响应主题）""" + if self._client is None: + return + if topic in self._subscribed: + return + await self._client.subscribe([(topic, QOS_1)]) + self._subscribed.add(topic) + + async def _loop(self) -> None: + """响应消息监听循环""" + while self._running: + try: + m = await self._client.deliver_message() # type: ignore + if m is None: + break + await self._handle(str(m.topic), m.data) + except asyncio.CancelledError: + break + except Exception as e: + if self._running: + logger.error(f"[InternalMQTT] 接收异常: {e}") + break + + async def _handle(self, topic: str, payload: bytes) -> None: + """解析响应，按 request_id / 响应主题匹配挂起的请求""" + try: + data = json.loads(bytes(payload).decode("utf-8")) + except (json.JSONDecodeError, UnicodeDecodeError): + logger.debug(f"[InternalMQTT] 响应非 JSON，忽略: {topic}") + return + + # 1) 对方回显 request_id → 直接匹配 + rid = data.get(RID_KEY) + if rid and rid in self._pending: + fut = self._pending[rid] + if not fut.done(): + fut.set_result(data) + return + + # 2) {rid} 模板唯一主题 → 按主题匹配 + for r, t in self._rid_topics.items(): + if t == topic and r in self._pending: + fut = self._pending[r] + if not fut.done(): + fut.set_result(data) + return + + # 3) 兜底：对方不回显 request_id，按 FIFO 匹配最早挂起的请求 + if self._pending: + rid0 = next(iter(self._pending)) + fut = self._pending[rid0] + if not fut.done(): + logger.debug( + f"[InternalMQTT] 无 request_id，按 FIFO 匹配 {rid0}（topic={topic}）" + ) + fut.set_result(data) + + @property + def is_running(self) -> bool: + return self._running diff --git a/src/chongming_mqtt/mqtt_forward.py b/src/chongming_mqtt/mqtt_forward.py new file mode 100644 index 0000000..509e131 --- /dev/null +++ b/src/chongming_mqtt/mqtt_forward.py @@ -0,0 +1,120 @@ +""" +AI Box Sophon Adapter — 外部 MQTT 转发客户端 + +建立到外部 MQTT Broker 的长连接，将设备心跳数据转发到指定主题。 +""" +import asyncio +import json +import logging +from urllib.parse import urlparse, unquote + +from amqtt.client import MQTTClient +from amqtt.mqtt.constants import QOS_1 + +logger = logging.getLogger(__name__) + + +class MQTTForwardClient: + """ + 外部 MQTT 转发客户端 + + 在插件启动时建立到外部 MQTT Broker 的持久连接， + 提供 publish() 方法将数据转发到指定主题。 + + 支持 URI 中的 percent-encoded 凭据（如密码中的 @ 写为 %40）。 + + 使用方式: + client = MQTTForwardClient(uri="mqtt://user:pass@host:port") + await client.start() + await client.publish("/topic", {"key": "value"}) + await client.stop() + """ + + def __init__(self, uri: str, topic: str = "/board_ping"): + self._uri = uri + self._topic = topic + self._client: MQTTClient | None = None + self._running = False + self._lock = asyncio.Lock() + + # 预解析 URI 并解码凭据（amqtt 的 urlparse 不会自动 unquote） + parsed = urlparse(uri) + self._clean_uri = f"{parsed.scheme}://{parsed.hostname}:{parsed.port or 1883}" + self._username: str | None = unquote(parsed.username) if parsed.username else None + self._password: str | None = unquote(parsed.password) if parsed.password else None + + async def start(self) -> None: + """建立到外部 MQTT Broker 的连接""" + if self._running: + return + + try: + self._client = MQTTClient( + config={ + "auto_reconnect": True, + "reconnect_retries": -1, # 无限重连 + "reconnect_max_interval": 30, # 最大重连间隔（秒） + "keep_alive": 60, + "connection_timeout": 10, # 连接超时（秒），避免永久挂起 + } + ) + # 手动初始化 session 并设置解码后的凭据（绕过 amqtt urlparse 不解码 %40 的问题） + self._client.session = self._client._init_session(self._clean_uri) + if self._username: + self._client.session.username = self._username + if self._password: + self._client.session.password = self._password + await self._client._do_connect() + self._running = True + logger.info(f"[MQTTForward] 已连接到外部 Broker: {self._clean_uri}") + logger.info(f"[MQTTForward] 转发目标主题: {self._topic}") + except Exception as e: + logger.error(f"[MQTTForward] 连接外部 Broker 失败: {e}") + self._client = None + raise + + async def stop(self) -> None: + """断开与外部 MQTT Broker 的连接""" + if not self._running: + return + + self._running = False + if self._client: + try: + await self._client.disconnect() + except Exception as e: + logger.warning(f"[MQTTForward] 断开连接时异常: {e}") + finally: + self._client = None + logger.info("[MQTTForward] 已断开外部 Broker 连接") + + async def publish(self, payload: dict, topic: str | None = None) -> bool: + """ + 发布消息到外部 MQTT Broker + + Args: + payload: 要发送的 JSON 数据 + topic: 目标主题，默认使用初始化时配置的主题 + + Returns: + 是否发送成功 + """ + if not self._running or self._client is None: + logger.warning("[MQTTForward] 客户端未连接，跳过转发") + return False + + target = topic or self._topic + + async with self._lock: + try: + data = json.dumps(payload, ensure_ascii=False, default=str) + await self._client.publish(target, data.encode("utf-8"), QOS_1) + logger.debug(f"[MQTTForward] 已转发到 {target}: {len(data)} 字节") + return True + except Exception as e: + logger.error(f"[MQTTForward] 发布失败 [{target}]: {e}") + return False + + @property + def is_running(self) -> bool: + return self._running diff --git a/src/chongming_mqtt/mqtt_subscribe.py b/src/chongming_mqtt/mqtt_subscribe.py new file mode 100644 index 0000000..d3ce263 --- /dev/null +++ b/src/chongming_mqtt/mqtt_subscribe.py @@ -0,0 +1,253 @@ +""" +AI Box Sophon Adapter — 外部 MQTT 订阅客户端 + +建立到外部 MQTT Broker 的长连接，订阅指定主题， +收到消息后按主题匹配调用绑定的处理函数（插件内直接调用，不走网关 Bridge）。 + +支持: + - 断线自动重连（无限重试） + - URI percent-encoded 凭据解码（如密码中的 @ 写为 %40） + - 多个订阅主题，主题支持 +（单级）/ #（多级，须在末尾）通配符 + - 每个主题绑定一个处理函数，收到消息后直接调用 + - 请求-响应：配置 reply_topic 后，将处理函数返回值发布到回复主题 + （reply_topic 支持 {1} {2}... 占位符，填充订阅主题通配符捕获值） + +使用方式: + client = MQTTSubscribeClient( + uri="mqtt://user:pass@host:port", + routes=[{"topic": "/edge_app_controller", + "handler": async_fn, + "reply_topic": "/edge_app_controller_reply"}], + ) + await client.start() + await client.stop() +""" +import asyncio +import json +import logging +import re +from collections.abc import Callable +from typing import Any +from urllib.parse import urlparse, unquote + +from amqtt.client import MQTTClient +from amqtt.mqtt.constants import QOS_1 + +logger = logging.getLogger(__name__) + +# 处理函数类型: async def handler(payload: dict, topic: str) -> Any +Handler = Callable[[dict, str], Any] + + +def _compile_topic(topic: str) -> re.Pattern: + """ + 将 MQTT 主题模板（支持 + / # 通配符）编译为正则表达式。 + + - ``+`` 匹配单级（编译为捕获组 ``([^/]+)``） + - ``#`` 匹配多级（编译为捕获组 ``(.*)``，须位于末尾） + """ + 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 MQTTSubscribeClient: + """ + 外部 MQTT 订阅客户端 + + 在插件启动时建立到外部 MQTT Broker 的持久连接， + 订阅配置的主题，收到消息后按主题匹配调用绑定的处理函数。 + + 处理函数约定: async def handler(payload: dict, topic: str) -> Any + 若配置了 reply_topic，处理函数返回值（非 None）会发布到该回复主题。 + """ + + def __init__(self, uri: str, routes: list | None = None): + """ + Args: + uri: mqtt://user:pass@host:port + routes: list[dict]，每项含 + topic: 订阅主题（支持 + / # 通配符） + handler: 处理函数 async callable(payload: dict, topic: str) + reply_topic: 可选，回复主题；收到消息处理后，将返回值发布到该主题 + （支持 {1} {2}... 占位符） + """ + self._uri = uri + self._routes: list[tuple[str, re.Pattern, Handler, str | None]] = [] + for r in routes or []: + topic = r.get("topic", "") + handler = r.get("handler") + reply_topic = r.get("reply_topic") + if topic and handler is not None: + self._routes.append((topic, _compile_topic(topic), handler, reply_topic)) + self._client: MQTTClient | None = None + self._loop_task: asyncio.Task | None = None + self._running = False + + # 预解析 URI 并解码凭据（amqtt 的 urlparse 不会自动 unquote） + parsed = urlparse(uri) + self._clean_uri = f"{parsed.scheme}://{parsed.hostname}:{parsed.port or 1883}" + self._username: str | None = unquote(parsed.username) if parsed.username else None + self._password: str | None = unquote(parsed.password) if parsed.password else None + + async def start(self) -> None: + """建立到外部 MQTT Broker 的连接并订阅主题""" + if self._running: + return + + try: + if not self._routes: + raise ValueError("未配置任何订阅主题 (routes 为空)") + + self._client = MQTTClient( + config={ + "auto_reconnect": True, + "reconnect_retries": -1, # 无限重连 + "reconnect_max_interval": 30, # 最大重连间隔（秒） + "keep_alive": 60, + "connection_timeout": 10, # 连接超时（秒），避免永久挂起 + } + ) + # 手动初始化 session 并设置解码后的凭据（绕过 amqtt urlparse 不解码 %40 的问题） + self._client.session = self._client._init_session(self._clean_uri) + if self._username: + self._client.session.username = self._username + if self._password: + self._client.session.password = self._password + await self._client._do_connect() + + topics = [(t, QOS_1) for t, _, _, _ in self._routes] + await self._client.subscribe(topics) + + self._running = True + self._loop_task = asyncio.create_task(self._loop()) + logger.info(f"[MQTTSubscribe] 已连接到外部 Broker: {self._clean_uri}") + logger.info(f"[MQTTSubscribe] 已订阅 {len(topics)} 个主题: {[t for t, _ in topics]}") + except Exception as e: + logger.error(f"[MQTTSubscribe] 连接外部 Broker 失败: {e}") + if self._client: + try: + await self._client.disconnect() + except Exception: + pass + self._client = None + raise + + async def stop(self) -> None: + """断开与外部 MQTT Broker 的连接并停止消息循环""" + if not self._running: + return + + self._running = False + # 先断开连接，让 deliver_message() 解锁 + if self._client: + try: + await self._client.disconnect() + except Exception as e: + logger.warning(f"[MQTTSubscribe] 断开连接时异常: {e}") + finally: + self._client = None + # 再取消消息循环 + if self._loop_task and not self._loop_task.done(): + self._loop_task.cancel() + try: + await self._loop_task + except asyncio.CancelledError: + pass + logger.info("[MQTTSubscribe] 已断开外部 Broker 连接") + + async def _loop(self) -> None: + """消息接收循环""" + while self._running: + try: + m = await self._client.deliver_message() # type: ignore + if m is None: + break + await self._dispatch(str(m.topic), m.data) + except asyncio.CancelledError: + break + except Exception as e: + if self._running: + logger.error(f"[MQTTSubscribe] 消息接收异常: {e}") + break + + async def _dispatch(self, topic: str, payload: bytes | bytearray) -> None: + """按主题匹配，调用处理函数，并按需发布回复""" + try: + data = json.loads(bytes(payload).decode("utf-8")) + except (json.JSONDecodeError, UnicodeDecodeError) as e: + logger.warning(f"[MQTTSubscribe] 消息非 JSON，忽略 [{topic}]: {e}") + return + + for tpl, compiled, handler, reply_topic in self._routes: + matched = compiled.match(topic) + if matched: + result = None + try: + if asyncio.iscoroutinefunction(handler): + result = await handler(data, topic) + else: + result = handler(data, topic) + logger.debug(f"[MQTTSubscribe] [{tpl}] 处理完成: {topic}") + except Exception as e: + logger.error(f"[MQTTSubscribe] [{tpl}] 处理消息失败 [{topic}]: {e}") + result = None + + # 请求-响应：配置了回复主题且处理函数有返回结果时，发布回复 + if result is not None and reply_topic: + target = self._fill(reply_topic, matched) + ok = await self.publish(target, result) + if not ok: + logger.warning( + f"[MQTTSubscribe] 回复发布失败 [{topic}] -> {target}" + ) + return + + logger.debug(f"[MQTTSubscribe] 无匹配 handler，忽略: {topic}") + + async def publish(self, topic: str, payload: dict) -> bool: + """ + 发布消息到外部 MQTT Broker（用于回复请求方） + + Args: + topic: 目标主题 + payload: 要发送的 JSON 数据 + + Returns: + 是否发送成功 + """ + if not self._running or self._client is None: + logger.warning("[MQTTSubscribe] 客户端未连接，跳过发布") + return False + + try: + data = json.dumps(payload, ensure_ascii=False, default=str) + await self._client.publish(topic, data.encode("utf-8"), QOS_1) + logger.debug(f"[MQTTSubscribe] 已发布到 {topic}: {len(data)} 字节") + return True + except Exception as e: + logger.error(f"[MQTTSubscribe] 发布失败 [{topic}]: {e}") + return False + + @staticmethod + def _fill(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 + + @property + def is_running(self) -> bool: + return self._running