# lb_mqtt

**Repository Path**: litebmc/lb_mqtt

## Basic Information

- **Project Name**: lb_mqtt
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: LGPL-2.1
- **Default Branch**: feat/IJUFXC-dbus-mqtt-bridge
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-06-15
- **Last Updated**: 2026-06-29

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# lb_mqtt — LiteBMC MQTT 桥接组件

[![License](https://img.shields.io/badge/license-LGPL--2.1-blue.svg)](LICENSE)

**lb_mqtt** 是 [LiteBMC](https://litebmc.com) 框架的 MQTT 桥接组件，负责将 DBus 对象属性/方法/信号桥接到 MQTT 消息总线，实现 BMC 与外部系统的双向通信。

## 组件组成

| 组件 | 类型 | 说明 |
|------|------|------|
| `lb_mqtt` | 应用程序 | Broker 管理器 — 管理本地 mosquitto daemon 生命周期 |
| `mqtt_bridge.so` | 插件 | MQTT Bridge — DBus ↔ MQTT 双向桥接 |
| `lb_mqtt` (lib) | 库 | MQTT v5 客户端封装（基于 libmosquitto） |
| `lb_system_info` (lib) | 库 | 多源优先级链采集设备信息（device_id、device_model） |

## 架构

```
                 lb_core (global hooks)
                       │
       ┌───────────────┼───────────────┐
       │               │               │
  broker_manager  mqtt_bridge.so   其他插件...
  (priority 0)    (priority 6)
       │               │
       └───────┬───────┘
               │ localhost:8883
       ┌───────┴───────┐
       │   mosquitto   │  ← 本地 Broker
       └───────┬───────┘
               │
   ┌───────────┴───────────┐
   │                       │
外部客户端            上级 Broker（级联）
```

- **broker_manager**（优先级 0）：最先启动，确保 Bridge 初始化前 mosquitto 已就绪
- **mqtt_bridge.so**（优先级 6）：在 Broker 就绪后加载，注册全局 Hook 监听 DBus 事件

## 快速开始

### 前置要求

- **lbkit (lbk)**：LiteBMC 构建工具包
- **Conan 2.x**：依赖管理
- **CMake 3.5+**

### 构建

```bash
git clone https://gitee.com/litebmc/lb_mqtt.git
cd lb_mqtt

# 构建
lbk build

# 运行测试
lbk test

# 启用覆盖率
lbk build -o gcov=True
```

### 构建选项

| 选项 | 默认值 | 说明 |
|------|--------|------|
| `BUILD_TEST` | OFF | 启用测试编译 |
| `BUILD_ENABLE_DBUS_SYSTEM` | ON | 启用 DBus 系统总线 |
| `BUILD_ENABLE_DBUS_SESSION` | ON | 启用 DBus 会话总线 |

### 安装

```bash
# 安装到系统
lbk install
```

安装路径：
- 应用程序：`/opt/litebmc/apps/lb_mqtt/lb_mqtt`
- 插件：`/opt/litebmc/plugins/mqtt_bridge.so`
- systemd 服务：`/usr/lib/systemd/system/litebmc_lb_mqtt@.service`
- DBus 配置：`/usr/share/dbus-1/system.d/com.litebmc.lb_mqtt.conf`

---

## MQTT 客户端库

基于 libmosquitto 的 MQTT v5 客户端封装，提供简洁的 C API。

### 类型

```c
typedef struct _LbMqttClient LbMqttClient;

typedef struct {
    gchar *host;              // Broker 地址（必需）
    guint16 port;             // 端口（默认 1883）
    guint16 keepalive;        // 保活间隔秒数（默认 60）
    gchar *client_id;         // 客户端 ID
    gchar *username;          // 用户名
    gchar *password;          // 密码

    // TLS
    gboolean tls_enabled;
    gchar *ca_file;
    gchar *cert_file;
    gchar *key_file;
    gchar *key_password;
    gchar *tls_version;
    gchar *cipher_list;
    gboolean tls_insecure;

    // MQTT 5.0
    gboolean clean_start;
    guint32 session_expiry_sec;

    // 重连
    struct {
        guint32 initial_delay_ms;    // 默认 1000
        guint32 max_delay_ms;        // 默认 30000
        guint8 backoff_multiplier;
    } reconnect;
} LbMqttConfig;
```

### API

```c
// 初始化配置默认值
void lb_mqtt_config_init(LbMqttConfig *cfg);

// 创建客户端并连接（深拷贝配置字符串）
LbMqttClient *lb_mqtt_new(const LbMqttConfig *cfg,
                           LbMqttMessageHandler on_msg,
                           gpointer user_data);

// 发布消息
gint lb_mqtt_publish(LbMqttClient *c, const gchar *topic,
                      const void *payload, gsize len,
                      guint8 qos, gboolean retain);

// 订阅/取消订阅
gint lb_mqtt_subscribe(LbMqttClient *c, const gchar *topic, guint8 qos);
gint lb_mqtt_unsubscribe(LbMqttClient *c, const gchar *topic);

// 查询连接状态
gboolean lb_mqtt_is_connected(LbMqttClient *c);

// 释放客户端
void lb_mqtt_free(LbMqttClient **c);
```

### 用法示例

```c
#include "lb_mqtt.h"
#include "lb_attributes.h"

static void on_message(const gchar *topic, const void *payload,
                       gsize len, gpointer user_data)
{
    g_info("收到消息: %s (%.*s)", topic, (int)len, (const char *)payload);
}

void example(void)
{
    LbMqttConfig cfg;
    lb_mqtt_config_init(&cfg);
    cfg.host = "127.0.0.1";
    cfg.port = 8883;
    cfg.client_id = "my-client";
    cfg.tls_enabled = TRUE;
    cfg.ca_file = "/etc/mosquitto/certs/ca.crt";

    cleanup_LbMqttClient LbMqttClient *c = lb_mqtt_new(&cfg, on_message, NULL);
    if (!c) return;

    lb_mqtt_subscribe(c, "viewer/+/+/properties/changed", 0);
    lb_mqtt_publish(c, "operator/my-topic", "hello", 5, 0, FALSE);
    // 作用域结束时自动释放
}
```

---

## 系统信息库

多源优先级链采集设备信息，含 SHM 共识机制。

### 信息源链

每个 `LbInfoItem` 配置多个 `LbInfoSource`，按优先级依次尝试，首个返回值非 NULL 的源胜出。

内置源：INI 配置文件 → 网卡 MAC 地址 → CPU 序列号 → SHM 共识

### API

```c
// 信息源函数签名
typedef gchar *(*LbInfoSource)(void);

typedef struct {
    const gchar *key;          // 查询键名
    const gchar *description;  // 说明
    gboolean required;         // 无值时是否致命错误
    LbInfoSource *sources;     // 优先级源链（NULL 结尾）
} LbInfoItem;

// 初始化
gboolean lb_system_info_init(const gchar *config_path,
                              const LbInfoItem *items, gsize n_items);

// 查询
const gchar *lb_system_info_get(const gchar *key);

// 便捷访问
const gchar *lb_device_model(void);  // → lb_system_info_get("device_model")
const gchar *lb_device_id(void);     // → lb_system_info_get("device_id")
```

### 用法示例

```c
#include "lb_system_info.h"

// 自定义信息源
static gchar *_source_ini_device_id(void) { /* 从 INI 读取 */ }
static gchar *_derive_device_id_from_mac(void) { /* 从 MAC 推导 */ }

void init_system_info(void)
{
    LbInfoSource device_id_sources[] = {
        _source_ini_device_id,
        _derive_device_id_from_mac,
        NULL
    };

    LbInfoItem items[] = {
        {.key = "device_id", .description = "设备ID",
         .required = TRUE, .sources = device_id_sources},
        {.key = "device_model", .description = "设备型号",
         .required = TRUE, .sources = ...},
    };

    lb_system_info_init("/etc/litebmc/config.ini", items, G_N_ELEMENTS(items));

    const gchar *dev_id = lb_device_id();
    const gchar *model = lb_device_model();
    g_info("设备: %s (%s)", dev_id, model);
}
```

### SHM 共识机制

多进程竞争生成设备 ID，通过 3 次连续一致读确认胜者，避免竞态条件。

---

## Broker 管理器

管理本地 mosquitto MQTT broker 实例。

### 职责

| 功能 | 说明 |
|------|------|
| 进程管理 | fork+exec mosquitto，SIGTERM 停止 |
| 配置生成 | 从系统信息生成 mosquitto.conf + ACL + bridge 配置 |
| 健康检查 | 周期性探测 Broker 连通性 |
| 证书管理 | TLS 证书过期预警 |
| DBus 接口 | 通过 `com.litebmc.lbMqttBroker` 接口暴露状态和控制 |

### 启动时序

```c
// constructor(200) 确保在 lb_core 模块初始化前注册
static void __attribute__((constructor(200))) _broker_manager_init(void) {
    lb_module_register(_broker_manager_start, "BrokerManager", 0);
}
```

状态机：`STOPPED → STARTING → RUNNING / ERROR`

---

## MQTT Bridge 插件

以独立共享库 `mqtt_bridge.so` 加载，将 DBus 属性变更/信号发射桥接到 MQTT。

### 插件入口

```c
void plugin_start(void) {
    lb_module_register(_mqtt_bridge_start, "MqttBridge", 6);
}
```

### 插件目录结构

```
src/plugin/
├── mqtt_bridge_plugin.c       # plugin_start() + 模块注册
├── mqtt_bridge_config.c       # YAML 配置加载
├── mqtt_bridge_routing.c      # Hook 回调 + topic 路由
├── mqtt_bridge_throttle.c     # 限流 + 批处理
├── mqtt_bridge_serialize.c    # GVariant↔JSON 序列化
├── mqtt_bridge_security.c     # 入站校验 + 错误处理
└── mqtt_bridge_internal.h     # 内部头文件
```

### 消息队列解耦

全局 Hook 在 DBus worker 线程同步调用。Bridge 通过内部队列解耦，避免阻塞 DBus 事件处理：

```
DBus worker 线程                    Bridge 内部队列              Bridge worker 线程
─────────────────                  ──────────────              ─────────────────
Hook callback (过滤 + GVariant ref)
  → g_async_queue_push() ────→  [event] ────→  g_async_queue_pop()
  返回（< 1μs）                                      │
                                                     ├─ throttle 判决
                                                     ├─ batch 聚合
                                                     ├─ 序列化
                                                     └─ lb_mqtt_publish()
```

### 属性缓存

Bridge 维护每个对象的全量属性 JSON 缓存（TTL 300s），changed 事件增量更新，get 请求直接返回缓存。

---

## Topic 设计

| 操作 | Topic | 方向 |
|------|-------|------|
| 属性写入 | `{role}/{sid}/{path}/properties/set` | MQTT → DBus |
| 属性读取 | `{role}/{sid}/{path}/properties/get` | MQTT → DBus |
| 属性读取响应 | `{role}/{sid}/{path}/properties/get/rsp` | DBus → MQTT |
| 属性变更通知 | `viewer/{sid}/{path}/properties/changed` | DBus → MQTT |
| 方法调用 | `{role}/{sid}/{path}/methods` | MQTT → DBus |
| 方法响应 | `{role}/{sid}/{path}/methods/rsp` | DBus → MQTT |
| 信号通知 | `viewer/{sid}/{path}/signals` | DBus → MQTT |
| 接口检索 | `viewer/{sid}/{path}/inspect` | MQTT → DBus |
| 接口检索响应 | `viewer/{sid}/{path}/inspect/rsp` | DBus → MQTT |

其中 `{sid}` 为 Service ID — 由 bus_name 经 SHA256 XOR-fold 生成的 8 字符短标识，首字符 `a-z`，其余 `0-9a-zA-Z`。

### 级联 Topic

```
BMC 内部:     operator/com/litebmc/sensor/t1/properties/set
BMC 对外:     R5300/bmc-00e04c680012/operator/com/litebmc/.../set
机架对外:     rack01/R5300/bmc-00e04c680012/operator/com/litebmc/.../set
DC 对外:      dc-east/rack01/R5300/bmc-00e04c680012/operator/.../set
```

---

## 配置文件

### mqtt_bridge.yaml

```yaml
# /opt/litebmc/config/mqtt_bridge.yaml
broker:
  host: "127.0.0.1"
  port: 8883
  client_id: "lb_mqtt_<bus_name>"
  username: "bridge_user"
  password: "***"
  tls:
    enabled: true
    ca_file: "/etc/mosquitto/certs/ca.crt"
    cert_file: "/etc/mosquitto/certs/bridge_client.crt"
    key_file: "/etc/mosquitto/certs/bridge_client.key"
    tls_version: "tlsv1.3"
    insecure: false

roles: [operator, admin]

throttle_defaults:
  interval_ms: 500
  change_threshold: 0.0

method_limits:
  defaults:
    max_calls_per_sec: 10

bridges:
  - interface: "com.litebmc.Sensor"
    properties: { all: true }
    signals: { all: true }
    methods: { include: ["SetThreshold"] }
```

---

## 安全防御

外部 MQTT 消息不可信，逐层防御：

| 层级 | 校验 | 失败处理 |
|------|------|---------|
| Broker ACL | topic 权限 | Broker 拒绝 |
| Payload 长度 | ≤ 64KB | 丢弃 + 告警 |
| 全局速率 | ≤ 1000 msg/s | 丢弃 + 告警 |
| JSON 解析 | 深度 ≤ 16 | 返回 error |
| 属性名白名单 | IDF 定义中 | 返回 "unknown property" |
| 类型校验 | codegen from_json | 返回 "type mismatch" |
| 值域校验 | min/max/enum | 返回 "value out of range" |
| 时间戳 | |now - ts| ≤ 60s | 拒绝重放 |

---

## 依赖

| 依赖 | 版本 | 说明 |
|------|------|------|
| lb_core | ≥ 0.7.0 | 全局 Hook API、插件加载 |
| lb_base | ≥ 0.10.1 | MQTT 客户端库、系统信息库 |
| mosquitto | ≥ 2.0.22 | MQTT broker + 客户端库 |
| glib-2.0 | — | GLib 运行时 |
| json-glib-1.0 | — | YAML 配置解析 |

---

## 测试

```bash
lbk test
```

测试覆盖（115 个用例）：

| 分组 | 用例数 |
|------|--------|
| Service ID 生成 | 6 |
| Broker 管理器 | 8 |
| 插件生命周期 | 8 |
| 属性写入 | 15 |
| 属性读取 | 7 |
| 属性变更通知 | 7 |
| 方法调用 | 8 |
| 信号通知 | 4 |
| 接口检索 | 12 |
| 限流与批处理 | 8 |
| 安全防御 | 15 |
| ACL 权限 | 5 |
| MQTT 5.0 特性 | 4 |
| 消息队列解耦 | 3 |
| 端到端 | 5 |

---

## 许可证

[LGPL-2.1](LICENSE)

## 相关链接

- [LiteBMC 文档](https://litebmc.com/docs)
- [lbkit 工具包](https://litebmc.com/docs/#/docs/lbkit/readme)
- [interfaces 仓库](https://gitee.com/litebmc/interfaces)