# 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`）。