194 lines
6.9 KiB
Markdown
194 lines
6.9 KiB
Markdown
# 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]` 下分别配置这三个客户端，启动逻辑封装在插件的
|
||
`mqtt_clients.py` 能力组件中（`start_mqtt_forward` / `start_mqtt_subscribe` /
|
||
`start_internal_mqtt`，启动失败自动降级为 `None` 并告警），
|
||
由 `AiboxSophonAdapterPlugin.on_init` 调用：
|
||
|
||
```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": "内部请求超时"}
|
||
```
|
||
|
||
---
|
||
|
||
## 测试
|
||
|
||
单元测试位于 `utils/mqtt/test/`，通过注入 `FakeMQTT` 模拟底层连接，**不依赖真实 Broker**，可离线快速运行。
|
||
|
||
```bash
|
||
uv run pytest utils/mqtt/test -v
|
||
```
|
||
|
||
覆盖范围：
|
||
|
||
| 测试文件 | 覆盖内容 |
|
||
| ------------------- | -------------------------------------------------------------------------------------------------- |
|
||
| `test_forward.py` | 凭据解码、默认/指定主题发布、未连接与异常时的行为、stop 断开 |
|
||
| `test_subscribe.py` | 主题通配符（`+`/`#`）编译、无效路由跳过、消息分发、`reply_topic` 回复、`None` 不回复、`{1}` 占位符 |
|
||
| `test_internal.py` | 凭据解码、回显 `request_id` 关联、`{rid}` 模板、超时、FIFO 兜底、并发乱序、订阅去重 |
|
||
|
||
> 若需对真实 Broker 做端到端验证，可参考项目根目录的 `test_mqtt_e2e.py` 模式，自行启动本地 broker 后联测。
|
||
|
||
---
|
||
|
||
## 构建集成
|
||
|
||
若通过 PyInstaller/Nuitka 打包，需在 `config.toml` 的 `[pyinstaller_build_import] base` 中加入：
|
||
|
||
```toml
|
||
"chongming_mqtt",
|
||
```
|
||
|
||
并确保 `chongming_mqtt` 已安装到目标 Python 环境（`uv pip install -e utils/mqtt`）。
|