chongming/chongming-mqtt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
chakcy
2026-08-07 14:25:52 +08:00
commit 5ca88d4de8
7 changed files with 903 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+72
View File
@@ -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
README.md
+170
View File
@@ -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`）。
pyproject.toml
+16
View File
@@ -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"]
src/chongming_mqtt/__init__.py
+19
View File
@@ -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",
]
src/chongming_mqtt/internal_mqtt.py
+253
View File
@@ -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
src/chongming_mqtt/mqtt_forward.py
+120
View File
@@ -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
src/chongming_mqtt/mqtt_subscribe.py
+253
View File
@@ -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