MQTT - 508364/NetLeaf GitHub Wiki

NetLeaf MQTT 模块

概述

netleaf_mqtt 是 NetLeaf 的 MQTT v5.0 协议客户端模块，完整实现了 MQTT 3.1.1 和 MQTT v5.0 协议规范。支持 TLS/SSL 加密通信（需启用 BUILD_TLS 构建选项）。

架构说明

MQTT TLS 功能直接集成在 netleaf_mqtt.dll 中，不单独生成 DLL：

netleaf_mqtt.dll
├── netleaf_mqtt.c      (MQTT 核心功能)
└── netleaf_mqtt_tls.c  (MQTT TLS 功能，可选)

当 BUILD_TLS=ON 时，编译后会包含 TLS 支持；否则仅包含基本 MQTT 功能。

支持的协议版本

  • MQTT 3.1.1
  • MQTT v5.0（完整协议支持）

API 概览

客户端管理

// 创建 MQTT 客户端
nl_mqtt_client_t* nl_mqtt_create(void);

// 销毁 MQTT 客户端
void nl_mqtt_destroy(nl_mqtt_client_t* client);

// 获取连接状态
nl_mqtt_status_t nl_mqtt_get_status(const nl_mqtt_client_t* client);

连接管理

// 连接 MQTT broker
int nl_mqtt_connect(nl_mqtt_client_t* client,
                    const char* host, uint16_t port,
                    const nl_mqtt_connect_opts_t* opts,
                    nl_mqtt_connect_callback_t on_connect);

// 断开连接
int nl_mqtt_disconnect(nl_mqtt_client_t* client);

发布/订阅

// 发布消息
int nl_mqtt_publish(nl_mqtt_client_t* client,
                    const char* topic,
                    const void* payload, size_t payload_len,
                    const nl_mqtt_pub_opts_t* opts);

// 订阅主题
int nl_mqtt_subscribe(nl_mqtt_client_t* client,
                      const char* topic,
                      const nl_mqtt_sub_opts_t* opts,
                      nl_mqtt_message_callback_t on_message,
                      void* user_data);

// 取消订阅
int nl_mqtt_unsubscribe(nl_mqtt_client_t* client, const char* topic);

连接选项

typedef struct nl_mqtt_connect_opts {
    const char* client_id;           // 客户端 ID
    uint16_t    keep_alive;          // 心跳间隔（秒）
    const char* username;            // 用户名
    const char* password;            // 密码
    int         clean_session;       // 干净会话
    int         will_enabled;        // 启用遗嘱
    const char* will_topic;          // 遗嘱主题
    const char* will_msg;            // 遗嘱消息
    size_t      will_msg_len;        // 遗嘱消息长度
    int         will_qos;            // 遗嘱 QoS
    int         will_retain;         // 遗嘱保留
    int         is_utf8;             // UTF-8 编码

    /* MQTT 5.0 属性 */
    uint32_t    session_expiry_interval;  // 会话过期间隔
    uint32_t    will_delay_interval;      // 遗嘱延迟间隔
    uint32_t    max_packet_size;          // 最大包大小
    uint16_t    topic_alias_max;          // 主题别名最大值
    int         request_problem_info;     // 请求问题信息
    int         request_response_info;    // 请求响应信息
    nl_mqtt_property_t* properties;       // 自定义属性
    nl_mqtt_property_t* will_properties;  // 遗嘱属性
    nl_mqtt_property_t* user_properties;  // 用户属性
} nl_mqtt_connect_opts_t;

发布选项

typedef struct nl_mqtt_pub_opts {
    int   qos;                    // 服务质量等级 (0/1/2)
    int   retain;                 // 保留消息
    int   dup;                    // 重复标记

    /* MQTT 5.0 属性 */
    uint32_t    message_expiry_interval;   // 消息过期间隔
    int         payload_format_indicator;  // 载荷格式指示
    const char* content_type;              // 内容类型
    const char* response_topic;            // 响应主题
    const char* correlation_data;          // 关联数据
    size_t      correlation_data_len;      // 关联数据长度
    uint16_t    topic_alias;               // 主题别名
    nl_mqtt_property_t* properties;        // 自定义属性
    nl_mqtt_property_t* user_properties;   // 用户属性
} nl_mqtt_pub_opts_t;

订阅选项

typedef struct nl_mqtt_sub_opts {
    int   qos;                           // 服务质量等级
    int   no_local;                      // 不发送自己发布的消息
    int   retain_as_published;           // 保持遗嘱保留标志
    int   retain_handling;               // 保留消息处理策略
    int   subscription_identifiers_available; // 订阅标识符可用
    int   shared_subscription;           // 共享订阅
    const char* shared_topic;            // 共享主题

    /* MQTT 5.0 属性 */
    uint8_t    subscription_identifier;  // 订阅标识符
    nl_mqtt_property_t* properties;      // 自定义属性
    nl_mqtt_property_t* user_properties; // 用户属性
} nl_mqtt_sub_opts_t;

回调函数

// 连接回调
typedef void (*nl_mqtt_connect_callback_t)(int return_code, void* user_data);

// 消息回调
typedef void (*nl_mqtt_message_callback_t)(const char* topic, const void* payload,
                                           size_t payload_len, int qos, int retain,
                                           void* user_data);

// 断开回调
typedef void (*nl_mqtt_disconnect_callback_t)(void* user_data);

事件循环

// 处理入站数据
int nl_mqtt_handle_incoming(nl_mqtt_client_t* client);

// 处理出站数据
int nl_mqtt_handle_outgoing(nl_mqtt_client_t* client);

// 维护连接（综合处理）
int nl_mqtt_maintain(nl_mqtt_client_t* client, long timeout_ms);

QoS 2 辅助

int nl_mqtt_send_pubrec(nl_mqtt_client_t* client, uint16_t packet_id);
int nl_mqtt_send_pubrel(nl_mqtt_client_t* client, uint16_t packet_id);
int nl_mqtt_send_pubcomp(nl_mqtt_client_t* client, uint16_t packet_id);

Keep-alive

int nl_mqtt_ping(nl_mqtt_client_t* client);
int nl_mqtt_handle_pingresp(nl_mqtt_client_t* client);

连接状态

typedef enum {
    NL_MQTT_DISCONNECTED = 0,
    NL_MQTT_CONNECTING   = 1,
    NL_MQTT_CONNECTED    = 2,
    NL_MQTT_RECONNECTING = 3
} nl_mqtt_status_t;

连接返回码

typedef enum {
    NL_MQTT_RC_SUCCESS                          = 0,
    NL_MQTT_RC_GRANTED_QOS0                     = 0,
    NL_MQTT_RC_GRANTED_QOS1                     = 1,
    NL_MQTT_RC_GRANTED_QOS2                     = 2,
    NL_MQTT_RC_DISCONNECT_WITH_WILL_MSG         = 4,
    NL_MQTT_RC_NO_MATCHING_SUBSCRIBERS          = 16,
    NL_MQTT_RC_NO_SUBSCRIPTION_EXISTED          = 17,
    NL_MQTT_RC_CONTINUE_AUTHENTICATION          = 24,
    NL_MQTT_RC_REAUTHENTICATE                   = 25,
    NL_MQTT_RC_UNSPECIFIED_ERROR                = 128,
    NL_MQTT_RC_MALFORMED_PACKET                 = 129,
    NL_MQTT_RC_PROTOCOL_ERROR                   = 130,
    NL_MQTT_RC_IMPLEMENTATION_SPECIFIC_ERROR    = 131,
    NL_MQTT_RC_UNSUPPORTED_PROTOCOL_VERSION   = 132,
    NL_MQTT_RC_CLIENT_IDENTIFIER_NOT_VALID    = 133,
    NL_MQTT_RC_BAD_USERNAME_OR_PASSWORD       = 135,
    NL_MQTT_RC_NOT_AUTHORIZED                 = 136,
    NL_MQTT_RC_SERVER_UNAVAILABLE             = 137,
    NL_MQTT_RC_SERVER_BUSY                    = 138,
    NL_MQTT_RC_BANNED                         = 139,
    NL_MQTT_RC_BAD_AUTH_METHOD                = 140,
    NL_MQTT_RC_TOPIC_NAME_INVALID             = 144,
    NL_MQTT_RC_PACKET_TOO_LARGE               = 149,
    NL_MQTT_RC_QUOTA_EXCEEDED                 = 151,
    NL_MQTT_RC_PAYLOAD_FORMAT_INVALID         = 153,
    NL_MQTT_RC_RETAIN_NOT_SUPPORTED           = 154,
    NL_MQTT_RC_TOPIC_ALIAS_INVALID            = 155,
    NL_MQTT_RC_TOPIC_ALIAS_REQUIRED           = 156,
    NL_MQTT_RC_INVALID_STATE                  = 157,
    NL_MQTT_RC_NETWORK_CONNECTION_CLOSED      = 159,
    NL_MQTT_RC_INVALID_PROTOCOL_VERSION       = 160,
    NL_MQTT_RC_BAD_CLIENT_ID                = 161,
    NL_MQTT_RC_WILL_MESSAGE_INVALID           = 163,
    NL_MQTT_RC_WILL_MESSAGE_TOO_LARGE       = 164,
    NL_MQTT_RC_IDENTIFIER_REMOVED           = 165,
    NL_MQTT_RC_RATE_EXCEEDED                = 166,
    NL_MQTT_RC_CONNECT_AUTH               = 235
} nl_mqtt_rc_t;

TLS 支持

基本 TLS 连接

#include "netleaf_mqtt.h"
#include <stdio.h>

void on_connect(int rc, void* userdata) {
    printf("MQTT connected with result: %d\n", rc);
}

void on_message(const char* topic, const void* payload, size_t len, void* userdata) {
    printf("Received [%s]: %.20s...\n", topic, payload);
}

int main() {
    nl_mqtt_client_t* client = nl_mqtt_create();

    nl_mqtt_connect_opts_t opts = {
        .client_id = "netleaf_client_001",
        .keep_alive = 60,
        .clean_session = 1
    };

    // 使用 TLS 连接（端口 8883）
    nl_mqtt_connect(client, "broker.hivemq.com", 8883, &opts, on_connect);
    nl_mqtt_subscribe(client, "netleaf/#", &(nl_mqtt_sub_opts_t){.qos = 1}, on_message, NULL);

    // 事件循环
    while (1) {
        nl_mqtt_maintain(client, 1000);
    }

    nl_mqtt_destroy(client);
    return 0;
}

高级 TLS 配置

#include "netleaf_mqtt.h"
#include "netleaf_mqtt_tls.h"
#include <stdio.h>

int main() {
    nl_mqtt_client_t* client = nl_mqtt_create();

    // 设置 TLS 选项
    nl_mqtt_set_tls(client, 1);
    nl_mqtt_set_tls_cert(client, "ca.crt", "client.crt", "client.key");
    nl_mqtt_set_tls_insecure(client, 0);  // 启用证书验证

    nl_mqtt_connect_opts_t opts = {
        .client_id = "tls_client_001",
        .keep_alive = 60
    };

    nl_mqtt_connect(client, "broker.example.com", 8883, &opts, NULL);
    
    while (1) {
        nl_mqtt_maintain(client, 1000);
    }

    nl_mqtt_destroy(client);
    return 0;
}

使用示例

基本发布/订阅

#include "netleaf_mqtt.h"
#include <stdio.h>

void on_connect(int rc, void* userdata) {
    if (rc == 0) {
        printf("Connected successfully!\n");
        nl_mqtt_subscribe(userdata, "test/topic", 
                         &(nl_mqtt_sub_opts_t){.qos = 1},
                         on_message, userdata);
    }
}

void on_message(const char* topic, const void* payload, size_t len, 
                int qos, int retain, void* userdata) {
    printf("Received [%s]: %.20s...\n", topic, payload);
}

int main() {
    nl_mqtt_client_t* client = nl_mqtt_create();

    nl_mqtt_connect_opts_t opts = {
        .client_id = "netleaf_client_001",
        .keep_alive = 60,
        .clean_session = 1
    };

    nl_mqtt_connect(client, "broker.hivemq.com", 1883, &opts, on_connect);

    // 发布消息
    const char* msg = "Hello MQTT!";
    nl_mqtt_pub_opts_t pub_opts = {.qos = 1, .retain = 0};
    nl_mqtt_publish(client, "test/topic", msg, strlen(msg), &pub_opts);

    // 事件循环
    while (1) {
        nl_mqtt_maintain(client, 1000);
    }

    nl_mqtt_destroy(client);
    return 0;
}

遗嘱消息

#include "netleaf_mqtt.h"

int main() {
    nl_mqtt_client_t* client = nl_mqtt_create();

    nl_mqtt_connect_opts_t opts = {
        .client_id = "client_with_will",
        .keep_alive = 60,
        .clean_session = 1,
        .will_enabled = 1,
        .will_topic = "client/status",
        .will_msg = "offline",
        .will_msg_len = 7,
        .will_qos = 1,
        .will_retain = 1
    };

    nl_mqtt_connect(client, "broker.hivemq.com", 1883, &opts, NULL);

    while (1) {
        nl_mqtt_maintain(client, 1000);
    }

    nl_mqtt_destroy(client);
    return 0;
}

MQTT v5.0 属性

#include "netleaf_mqtt.h"

int main() {
    nl_mqtt_client_t* client = nl_mqtt_create();

    // 设置 MQTT 5.0 连接属性
    nl_mqtt_connect_opts_t opts = {
        .client_id = "mqtt5_client",
        .keep_alive = 60,
        .session_expiry_interval = 3600,  // 会话保持 1 小时
        .max_packet_size = 1024 * 1024,   // 最大包大小 1MB
        .topic_alias_max = 100,           // 主题别名最大值
        .request_problem_info = 1,        // 请求问题信息
        .request_response_info = 1        // 请求响应信息
    };

    // 添加自定义属性
    nl_mqtt_property_t* props = NULL;
    nl_mqtt_property_add_str(&props, NL_MQTT_PROP_CONTENT_TYPE, "application/json", -1);
    nl_mqtt_property_add_int(&props, NL_MQTT_PROP_PAYLOAD_FORMAT_INDICATOR, 1);
    opts.properties = props;

    nl_mqtt_connect(client, "broker.hivemq.com", 1883, &opts, NULL);

    // 发布带属性的消息
    nl_mqtt_pub_opts_t pub_opts = {
        .qos = 1,
        .message_expiry_interval = 60,      // 消息 60 秒后过期
        .payload_format_indicator = 1,      // UTF-8 编码
        .content_type = "application/json"
    };

    const char* msg = "{\"temperature\": 25.5}";
    nl_mqtt_publish(client, "sensors/temp", msg, strlen(msg), &pub_opts);

    while (1) {
        nl_mqtt_maintain(client, 1000);
    }

    nl_mqtt_property_list_destroy(props);
    nl_mqtt_destroy(client);
    return 0;
}

构建配置

CMake 选项

# 仅构建 MQTT（不含 TLS）
cmake -B build -DBUILD_MQTT=ON

# 构建 MQTT + TLS
cmake -B build -DBUILD_MQTT=ON -DBUILD_TLS=ON

构建脚本

# Windows
build_all-MSVC.bat
build_all-Clang.bat

# Linux/macOS
./build_all-linux.sh
./build_all-macos.sh

协议特性

QoS 等级

QoS 说明
0 最多一次（fire and forget）
1 至少一次（带确认）
2 恰好一次（四次握手）

主题通配符

  • + - 匹配单个路径段
  • # - 匹配剩余所有路径段

示例：

  • sensor/temp/+ - 匹配 sensor/temp/live、sensor/temp/history
  • sensor/# - 匹配 sensor/temp、sensor/temp/live、sensor/humidity

遗嘱消息

  • 连接时设置
  • 断开时自动发布
  • 支持自定义 QoS 和保留标志

开源项目来源

本模块参考了以下开源项目：

项目 用途 许可证 仓库链接
MQTT Specification MQTT v5.0 协议规范 EPL-2.0 / EDL 1.0 mqtt.org
Paho MQTT C MQTT C 客户端参考实现 EPL-2.0 / EDL 1.0 eclipse/paho.mqtt.c

注意事项

  1. 客户端 ID：必须唯一，建议包含设备标识
  2. 心跳间隔：Keep-alive 应小于 broker 超时设置
  3. 主题长度：最大 65535 字节
  4. 消息长度：最大 268435455 字节（256MB）
  5. 线程安全：单个客户端不保证线程安全，多线程使用请各自创建独立客户端

相关文档

许可声明 / License Notice 本仓库基于 MIT 协议开源。如果您使用了本项目，请保留原作者署名并附带 LICENSE 文件

This repository is open-sourced under the MIT License. If you use this project, please retain the original author's attribution and include the LICENSE file.

⚠️ **GitHub.com Fallback** ⚠️