MQTT 开发指导_Rev1.3
修订记录
版本 |
日期 |
作者 |
修订内容 |
|---|---|---|---|
Rev1.0 |
23-09-11 |
TL |
创建文档 |
Rev1.1 |
24-03-25 |
sxx |
更改文档名称 |
Rev1.2 |
24-05-01 |
YMX |
调整文档结构，优化部分文字描述内容 |
Rev1.3 |
26-01-16 |
lxh |
新增接口 |
1 引言
本文档介绍 LTE-EC71x MQTT 接口 API 情况，API 接口位于 components/kernel/lierda_api/liot_mqtt/liot_mqtt_client.h 文件声明中。
LTE-EC71x 系列模组最新SDK支持6 路 MQTT。
2 API 函数概览
函数 |
说明 |
|---|---|
|
初始化 MQTT 客户端资源并创建一个新的 MQTT 客户端句柄 |
|
初始化 MQTT 客户端资源并创建一个新的 MQTT 客户端句柄(该接口是新增接口与liot_mqtt_client_init()接口只能选一个，该接口是将所有的回调函数统一到一个回调函数中去使用，只注册一个回调函数即可) |
|
配置 MQTT 上下文，并与服务器建立连接 |
|
向指定topic发布消息 |
|
订阅/取消订阅topic |
|
断开连接 |
|
设置接收服务器发布消息的处理回调函数 |
|
查询mqtt连接状态 |
|
释放mqtt客户端资源 |
|
发送ping消息 |
|
获取onenet平台token，仅用于特定onenet云平台 |
|
获取MQTT状态 |
3 类型说明
3.1 liot_mqtt_error_code_e
MQTT API 执行结果错误码。
声明
typedef enum
{
LIOT_MQTTCLIENT_SUCCESS = 0,
LIOT_MQTTCLIENT_INVALID_PARAM = -1,
LIOT_MQTTCLIENT_WOUNDBLOCK = -2,
LIOT_MQTTCLIENT_OUT_OF_MEM = -3,
LIOT_MQTTCLIENT_ALLOC_FAIL = -4,
LIOT_MQTTCLIENT_TCP_CONNECT_FAIL = -5,
LIOT_MQTTCLIENT_NOT_CONNECT = -6,
LIOT_MQTTCLIENT_SEND_PKT_FAIL = -7,
LIOT_MQTTCLIENT_BAD_REQUEST = -8,
LIOT_MQTTCLIENT_TIMEOUT = -9,
} liot_mqtt_error_code_e;
参数
LIOT_MQTTCLIENT_SUCCESS ：函数执行成功。
LIOT_MQTTCLIENT_INVALID_PARAM ：传入参数错误。
LIOT_MQTTCLIENT_WOUNDBLOCK ：操作正在执行，等待结果返回。
LIOT_MQTTCLIENT_OUT_OF_MEM ：内存不足。
LIOT_MQTTCLIENT_ALLOC_FAIL ：内存分配出错。
LIOT_MQTTCLIENT_TCP_CONNECT_FAIL ：TCP 连接建立失败。
LIOT_MQTTCLIENT_NOT_CONNECT ：MQTT 会话没有建立。
LIOT_MQTTCLIENT_SEND_PKT_FAIL ：发送请求失败。
LIOT_MQTTCLIENT_BAD_REQUEST ：错误请求。
LIOT_MQTTCLIENT_TIMEOUT ：超时错误。
3.2 liot_mqtt_event_code_e
新增接口liot_mqtt_client_init_ex回调函数的event事件返回值
声明
enum MQTT_EVENT_STATUS
{
LIOT_MQTT_OPEN_EVENT, /*<!open event*/
LIOT_MQTT_CONNECT_EVENT, /*<!connect event*/
LIOT_MQTT_RECONNECT_EVENT, /*<!reconnect event*/
LIOT_MQTT_PUB_EVENT, /*<!publish event*/
LIOT_MQTT_SUB_EVENT, /*<!subscribe event*/
LIOT_MQTT_UNSUB_EVENT, /*<!unsubscribe event*/
LIOT_MQTT_CLOSE_EVENT, /*<!close event*/
LIOT_MQTT_DISCONNECT_EVENT, /*<!disconnect event*/
LIOT_MQTT_PUBLISH_EVENT, /*<!publish message event*/
}liot_mqtt_event_code_e;
参数
LIOT_MQTT_OPEN_EVENT : 打开事件
LIOT_MQTT_CONNECT_EVENT： 连接事件
LIOT_MQTT_RECONNECT_EVENT： 重连事件
LIOT_MQTT_PUB_EVENT： 发布事件
LIOT_MQTT_SUB_EVENT： 订阅事件
LIOT_MQTT_UNSUB_EVENT： 去订阅事件
LIOT_MQTT_CLOSE_EVENT： 关闭事件
LIOT_MQTT_DISCONNECT_EVENT: 断开事件
LIOT_MQTT_PUBLISH_EVENT： 收到数据事件
3.3 liot_mqtt_state_code_e
MQTT连接状态
声明
enum MQTT_STATE { MQTT_CONN_DEFAULT, /*<!default state*/ MQTT_CONN_NOT_OPEN, /*<!not open*/ MQTT_CONN_IS_OPENING, /*<!is opening*/ MQTT_CONN_OPENED, /*<!opened*/ MQTT_CONN_OPEN_FAIL, /*<!open fail*/ MQTT_CONN_IS_CONNECTING, /*<!is connecting*/ MQTT_CONN_CONNECTED, /*<!connected*/ MQTT_CONN_CONNECT_FAIL, /*<!connect fail*/ MQTT_CONN_IS_CLOSING, /*<!is closing*/ MQTT_CONN_CLOSED, /*<!closed*/ MQTT_CONN_CLOSED_FAIL, /*<!closed fail*/ MQTT_CONN_IS_DISCONNECTING, /*<!is disconnecting*/ MQTT_CONN_DISCONNECTED, /*<!disconnected*/ MQTT_CONN_DISCONNECTED_FAIL, /*<!disconnected fail*/ MQTT_CONN_RECONNECTING, /*<!reconnecting*/ MQTT_CONN_RECONNECTING_FAIL, /*<!reconnecting fail*/ }liot_mqtt_state_code_e;
参数
MQTT_CONN_DEFAULT： MQTT连接默认参数状态
MQTT_CONN_NOT_OPEN: 没有打开状态
MQTT_CONN_IS_OPENING： 正在打开状态
MQTT_CONN_OPENED: 已打开
MQTT_CONN_OPEN_FAIL： 打开失败
MQTT_CONN_IS_CONNECTING： MQTT连接中
MQTT_CONN_CONNECTED： 已连接
MQTT_CONN_CONNECT_FAIL： 连接失败
MQTT_CONN_IS_CLOSING： MQTT关闭中
MQTT_CONN_CLOSED： 已关闭
MQTT_CONN_CLOSED_FAIL： 关闭失败
MQTT_CONN_IS_DISCONNECTING：MQTT正在断开连接中
MQTT_CONN_DISCONNECTED： 已断开
MQTT_CONN_DISCONNECTED_FAIL： 断开失败
MQTT_CONN_RECONNECTING： MQTT重连中
MQTT_CONN_RECONNECTING_FAIL： 重连失败
3.4 liot_mqtt_client_t
MQTT 客户端句柄。
声明
typedef int liot_mqtt_client_t;
参数
liot_mqtt_client_t：MQTT 客户端句柄。由 liot_mqtt_client_init()插入链接获取，初始值必须为0。
3.5 liot_mqtt_connection_cb_t
该回调函数由内核调用以通知应用层 MQTT CONNECT 请求结果。
声明
typedef void (*liot_mqtt_connection_cb_t)(liot_mqtt_client_t *client, void *arg, int status);
参数
client：[In] MQTT 客户端句柄。由 liot_mqtt_client_init()获取。
arg：[In] 回调参数。由 liot_mqtt_connect()传入。
status：[In] 执行结果码，请参考 3.1。
3.6 liot_mqtt_client_option_t
MQTT 客户端的信息结构体定义如下
声明
typedef struct liot_mqtt_client_option_t
{
unsigned char version;
int pdp_cid;
char *client_id;
char *client_user;
char *client_pass;
unsigned char ssl_enable;
void *ssl_ctx;
struct liot_mqtt_ssl_config_t *ssl_cfg;
unsigned char clean_session;
unsigned short kalive_time;
unsigned char delivery_time;
unsigned char delivery_cnt;
unsigned char will_flag;
unsigned char will_qos;
unsigned char will_retain;
char will_topic[257];
char will_message[257];
unsigned char ping_timeout;
} liot_mqtt_client_option;
参数
类型 |
参数 |
描述 |
|---|---|---|
unsigned char |
version |
MQTT 版本； |
int |
pdp_cid |
已激活的数据通道 ID |
char * |
client_id |
客户端 ID |
char * |
client_user |
CONNECT 请求时携带的用户名；若无，则设置为 NULL |
char * |
client_pass |
CONNECT 请求时携带的密码；若无，则设置为 NULL |
unsigned char |
ssl_enable |
MQTT SSL控制 |
void * |
ssl_ctx |
SSL上下文（暂不支持） |
struct liot_mqtt_ssl_config_t * |
ssl_cfg |
MQTT SSL 的配置；详见第 ； |
unsigned char |
clean_session |
0 表示开启 MQTT 会话复用机制，服务器保留会话包括定于的topic； |
unsigned short |
kalive_time |
客户端与服务器之间的保活间隔；单位：秒(建议30~300) |
unsigned char |
delivery_time |
发送超时时间；单位：秒 |
unsigned char |
delivery_cnt |
发送超时，重发次数； |
unsigned char |
will_flag |
遗嘱消息控制，仅在 will_flag 为 true 时生效 |
unsigned char |
will_qos |
MQTT 遗嘱消息的 Qos 等级 |
unsigned char |
will_retain |
遗嘱消息被服务器发布出去后，will_retain 为 True 时， |
char |
will_topic[257] |
MQTT 遗嘱的主题；若无，则设置为 NULL，仅在 will_flag 为 true 时生效 |
char |
will_message[257] |
MQTT 遗嘱的消息；若无，则设置为 NULL |
unsigned char |
ping_timeout |
ping包超时时间；单位：秒 |
3.7 liot_mqtt_ssl_config_t
MQTT SSL配置的结构体定义如下
声明
struct liot_mqtt_ssl_config_t{
int ssl_ctx_id;
int verify_level;
char *cacert_path;
char *client_cert_path;
char *client_key_path;
char *client_key_pwd;
int ssl_version;
int sni_enable;
int ssl_negotiate_timeout;
int ignore_invalid_certsign;
int ignore_multi_certchain_verify;
uint32_t ignore_certitem;
char *cacert_buffer;
bool client_cert_type;
};
参数
类型 |
参数 |
描述 |
|---|---|---|
int |
ssl_ctx_id |
SSL 上下文 ID |
int |
verify_level |
SSL 校验等级 |
char * |
cacert_path |
CA 证书的路径 |
char * |
client_cert_path |
客户端证书的路径 |
char * |
client_key_path |
客户端私钥文件的路径 |
char * |
client_key_pwd |
客户端私钥文件加密的口令 |
int |
ssl_version |
SSL 版本等级 |
int |
sni_enable |
SNI 使能标志位 |
int |
ssl_negotiate_timeout |
SSL 协商超时时间（不支持） |
int |
ignore_invalid_certsign |
忽略无效证书（不支持） |
int |
ignore_multi_certchain_verify |
忽略多证书链验证（不支持） |
uint32_t |
ignore_certitem |
忽略证书（不支持） |
char * |
cacert_buffer |
客户端证书buf（不支持） |
bool |
client_cert_type |
ssl证书导入模式 |
3.8 liot_mqtt_request_cb_t
该回调函数由内核调用以通知应用层 MQTT 订阅、取消订阅和 PUBLISH 请求结果。
声明
typedef void (*liot_mqtt_request_cb_t)(mqtt_client_t *client, void *arg, int err);
参数
client：[In] MQTT 客户端句柄。由 liot_mqtt_client_init()获取。。
arg：回调参数。由 liot_mqtt_sub_unsub()和liot_mqtt_publish()传入。
err：[In] 执行结果码，请参考 3.1。
3.9 liot_mqtt_incoming_publish_cb_t
该回调函数由内核调用以通知应用层接收到发布消息
声明
typedef void (*liot_mqtt_incoming_publish_cb_t)(mqtt_client_t *client, void *arg, int pkt_id, const char *topic, const unsigned char *payload, unsigned short payload_len);
参数
client：[In] MQTT 客户端句柄。由 liot_mqtt_client_init()获取。。
arg：[In] 回调参数。由 liot_mqtt_set_inpub_callback()传入。
pkt_id：[In] 发布消息的 ID。
topic：[In] 发布消息关联的主题。
payload：[In] 发布消息的消息体。
payload_len：[In] 发布消息的消息体长度
3.10 liot_mqtt_disconnect_cb_t
该回调函数由内核调用以通知应用层 MQTT DISCONNECT 请求的结果。
声明
typedef void (*liot_mqtt_disconnect_cb_t)(mqtt_client_t *client, void *arg, int err);
参数
client：[In] MQTT 客户端句柄。由 liot_mqtt_client_init()获取。。
arg：[In] 回调参数。由 liot_mqtt_connect()传入。
err：[In] 执行结果码，请参考 3.1。
3.11 liot_mqtt_state_exception_cb_t
该回调函数由内核调用以通知应用层 MQTT 会话连接异常断开。例如，掉网或保活超时导致 MQTT 会话连接断开。
声明
typedef void(*liot_mqtt_state_exception_cb_t)(mqtt_client_t *client);
参数
client：[In] MQTT 客户端句柄。由 liot_mqtt_client_init()获取。
3.12 liot_mqtt_event_cb_t
该函数是新接口liot_mqtt_client_init_ex函数的回调函数，使用改回调函数后MQTT相关的其他回调函数则不用填，对应的MQTT连接、发布订阅、去订阅、重连等接口的回调函数都为空
声明
typedef void (*liot_mqtt_event_cb_t)(liot_mqtt_client_t *client, int event, void *arg, void *data);
参数
client：[out] MQTT 客户端句柄。由 liot_mqtt_client_init()获取。
cid: [out] MQTT网络连接cid
4 API 函数详解
4.1 liot_mqtt_client_init
该函数用于初始化 MQTT 客户端资源并创建一个新的 MQTT 客户端句柄，如若调用liot_mqtt_client_init_ex函数，本函数可不用调用，只能二选一。
声明
int liot_mqtt_client_init(liot_mqtt_client_t *client, int cid);
参数
client：[Out] MQTT 客户端句柄。初始值为0。
cid：[In] 数据通道号，需要传入活动的cid，需填写为1。
返回值
int ：0，成功，非0，失败；执行结果码，请参考 3.1。
4.2 liot_mqtt_connect
该函数用于向服务器发送 CONNECT 请求，在客户端与服务器之间建立 MQTT 会话连接。
声明
int liot_mqtt_connect(liot_mqtt_client_t *client,
const char *host,
liot_mqtt_connection_cb_t cb,
void *arg,
liot_mqtt_client_option *client_info,
liot_mqtt_state_exception_cb_t exp_cb);
参数
client：[In] MQTT 客户端句柄。由 liot_mqtt_client_init()获取。
host：[In] MQTT 服务器地址；
以 mqtt://或者 mqtts://开头
mqtt://220.180.239.212:8306
mqtts://220.180.239.212:8307
cb：[in] CONNECT 请求结果回调函数(如若调用liot_mqtt_client_init_ex注册回调函数，该参数可为空)。
arg: [in] CONNECT 请求结果回调函数的回调参数。
client_info: [in] MQTT 客户端的信息(如若调用liot_mqtt_client_init_ex注册回调函数，该参数可为空)。
exp_cb: [in] MQTT 会话连接异常断开的回调函数。
返回值
int ：执行结果码，请参考 3.1。
4.3 liot_mqtt_publish
该函数用于发布指定主题的消息。
声明
int liot_mqtt_publish(liot_mqtt_client_t *client,
const char *topic,
const void *payload,
unsigned short payload_length,
unsigned char qos,
unsigned char retain,
liot_mqtt_request_cb_t cb,
void *arg);
参数
client：[In] MQTT 客户端句柄。由 liot_mqtt_client_init()获取。
topic：[In] 发布消息的主题。
payload：[in] 发布的消息。
payload_length: [in] 发布的消息长度。
qos: [in] 发布消息的 QoS 等级。
retain: [in] 是否保留发布的消息。
1 消息发布后存储于服务器上；
0 消息发布后立即删除。
cb: [in] PUBLISH 请求结果的回调函数(如若调用liot_mqtt_client_init_ex注册回调函数，该参数可为空)。
arg: [in] PUBLISH 请求结果回调函数的回调参数。
返回值
int ：执行结果码，请参考 3.1。
4.4 liot_mqtt_sub_unsub
该函数用于订阅/取消订阅指定主题。
声明
int liot_mqtt_sub_unsub(liot_mqtt_client_t *client,
const char *topic,
unsigned char qos,
liot_mqtt_request_cb_t cb,
void *arg,
unsigned char sub);
参数
client：[In] MQTT 客户端句柄。由 liot_mqtt_client_init()获取。
topic：[In] 订阅或取消订阅的主题名。
qos：[in] QoS质量等级。
cb: [in] 订阅/取消订阅结果的回调函数(如若调用liot_mqtt_client_init_ex注册回调函数，该参数可为空)。
arg: [in] mqtt 订阅/取消订阅主题,自定义传参。
sub: [in] 操作类型。
1 订阅
0 取消订阅
返回值
int ：执行结果码，请参考 3.1。
4.5 liot_mqtt_disconnect
该函数用于向服务器发送 DISCONNECT 请求，断开 MQTT 客户端与服务器之间的 MQTT 会话连接。
声明
int liot_mqtt_disconnect(liot_mqtt_client_t *client, liot_mqtt_disconnect_cb_t cb, void *arg);
参数
client：[In] MQTT 客户端句柄。由 liot_mqtt_client_init()获取。
cb：[In] DISCONNECT 请求结果回调函数(如若调用liot_mqtt_client_init_ex注册回调函数，该参数可为空)。
arg：[in] DISCONNECT 请求结果回调函数的回调参数
返回值
int ：执行结果码，请参考 3.1。
4.6 liot_mqtt_set_inpub_callback
该函数用于设置接收服务器发布消息的处理回调函数(如若调用liot_mqtt_client_init_ex注册回调函数，该参数可为空)。
声明
int liot_mqtt_set_inpub_callback(liot_mqtt_client_t *client, liot_mqtt_incoming_publish_cb_t inpub_cb, void *arg);
参数
client：[In] MQTT 客户端句柄。由 liot_mqtt_client_init()获取。
inpub_cb：[In] 接收服务器发布消息的回调函数。
arg：[in] 接收服务器发布消息回调函数的参数。
返回值
int ：执行结果码，请参考 3.1。
4.7 liot_mqtt_client_is_connected
该函数用于查询客户端与服务器之间的会话连接是否已经建立。
声明
int liot_mqtt_client_is_connected(liot_mqtt_client_t *client);
参数
client：[In] MQTT 客户端句柄。由 liot_mqtt_client_init()获取。
返回值
int ：执行结果码，请参考 3.1。
4.8 liot_mqtt_client_deinit
该函数用于释放 MQTT 客户端资源。
声明
int liot_mqtt_client_deinit(liot_mqtt_client_t *client);
参数
client：[In] MQTT 客户端句柄。由 liot_mqtt_client_init()获取。
返回值
int ：执行结果码，请参考 3.1。
4.9 liot_mqtt_pingreq
该函数用于向 MQTT服务器发送ping请求。
声明
int liot_mqtt_pingreq(liot_mqtt_client_t *client);
参数
client：[In] MQTT 客户端句柄。由 liot_mqtt_client_init()获取。
返回值
int ：执行结果码，请参考 3.1。
4.10 liot_onenet_generate_auth_token
此函数是用于生成 OneNET 平台的 连接密码，使用 HMACSHA256 签名算法。
声明
char *liot_onenet_generate_auth_token(INT64 expire_time,
const char *product_id,
const char *device_name,
const char *version,
const char *device_access_key);
参数
client：[In] MQTT 客户端句柄。由 liot_mqtt_client_init()获取。
product_id：[In]
device_name：[In]
version：[In]
device_access_key：[In]
返回值
char：执行结果码。
非NULL 成功，执行完后需使用 free()释放空间
NULL 获取token失败
4.11 liot_mqtt_client_state
获取MQTT的状态
声明
int liot_mqtt_client_state(liot_mqtt_client_t *client);
参数
client：[In] MQTT 客户端句柄。由 liot_mqtt_client_init()获取。
返回值
int ：MQTT当前状态码，具体状态码见3.3
4.12 liot_mqtt_client_init_ex
该函数用于初始化 MQTT 客户端资源并创建一个新的 MQTT 客户端句柄。调用该接口函数创建MQTT句柄时，MQTT只需注册一个回调函数即可，无需注册多个回调函数
声明
int liot_mqtt_client_init_ex(liot_mqtt_client_t *client, int cid, liot_mqtt_event_cb_t event_cb, void *arg);
参数
client：[Out] MQTT 客户端句柄。初始值为0。
cid：[In] 数据通道号，需要传入活动的cid，根据网络拨号的cid去填写对应的值。
event_cb: [in]注册的MQTT回调函数，注册当前回调函数后无需注册其他回调函数
arg: [in] event_cb回调函数的参数
返回值
int ：0，成功，非0，失败；执行结果码，请参考 3.1
5 代码示例
示例代码参考 examples/demo/src/demo_mqtt_ali.c 文件。
/
* @File Name: demo_mqtt_ali.c
* @brief
* @Author : L email:ciot_iot_support@lierda.com
* @Version : 1.1
* @Creat Date : 2025-12-29
*
* @copyright Copyright (c) 2023 Lierda.com Inc. All rights reserved.
*
* @note : The function implemented by this file is to realize MQTT publishing,
* subscribing and other functions using MQTT API and Alibaba Cloud IOT.
* It connects to devices via two MQTT channels, and the device connection method adopts the "one product one secret" approach.
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "lierda_app_main.h"
#include "liot_datacall.h"
#include "liot_os.h"
#include "liot_type.h"
#include "liot_fs_api.h"
#include "liot_http.h"
#include "liot_mqtt_client.h"
#include "cms_util.h"
#ifdef FEATURE_MQTT_TLS_ENABLE
#include "sha256.h"
#endif
#include "cJSON.h"
#include "liot_nv.h"
#include "liot_fs_api.h"
#include "liot_nw.h"
#include "liot_ssl.h"
#define PRODUCTKEY_MAXLEN (20)
#define DEVICENAME_MAXLEN (32)
#define DEVICESECRET_MAXLEN (65)
#define SIGN_SOURCE_MAXLEN (200)
#define CLIENTID_MAXLEN (150)
#define USERNAME_MAXLEN (64)
#define PASSWORD_MAXLEN (65)
#define ALI_PRODUCT_SECRET "CUCCrRl7NFRo79fK"
#define ALI_PRODUCT_KEY "a1HjUZZgbcB"
#define ALI_DEVICE_NAME1 "IOT-38"
#define ALI_DEVICE_NAME2 "IOT-39"
// #define MQTT_CLIENT_IDENTITY "866179061276528|securemode=2,signmethod=hmacsha1,timestamp=4070880000000|"
// #define MQTT_CLIENT_USER "866179061276528&hf05listj7l"
// #define MQTT_CLIENT_PASS "da4a91865e995fc663fb9286c05d956026c52a0b"
//HTTP request url
#define HTTP_UARL "https://iot-auth.cn-shanghai.aliyuncs.com/auth/register/device"
char* authType = "register";
#define LIOT_NV_MQTT1_CFG "liot_nv_mqtt1_cfg.nvm"
#define LIOT_NV_MQTT2_CFG "liot_nv_mqtt2_cfg.nvm"
#define PRODUCTKEY_MAXLEN (20)
#define DEVICENAME_MAXLEN (32)
#define DEVICESECRET_MAXLEN (65)
#define SIGN_SOURCE_MAXLEN (200)
#define CLIENTID_MAXLEN (150)
#define USERNAME_MAXLEN (64)
#define PASSWORD_MAXLEN (65)
// #define MQTT_CLIENT_IDENTITY "866179061276528|securemode=2,signmethod=hmacsha1,timestamp=4070880000000|"
// #define MQTT_CLIENT_USER "866179061276528&hf05listj7l"
// #define MQTT_CLIENT_PASS "da4a91865e995fc663fb9286c05d956026c52a0b"
// subscribe topic
#define MQTT_ALI_SUB_TOPIC1 "/" ALI_PRODUCT_KEY "/" ALI_DEVICE_NAME1 "/user/1234"
//publish topic
#define MQTT_ALI_PUB_TOPIC1 "/" ALI_PRODUCT_KEY "/" ALI_DEVICE_NAME1 "/user/1234"
//subscribe topic
#define MQTT_ALI_SUB_TOPIC2 "/" ALI_PRODUCT_KEY "/" ALI_DEVICE_NAME2 "/user/1234"
//publish topic
#define MQTT_ALI_PUB_TOPIC2 "/" ALI_PRODUCT_KEY "/" ALI_DEVICE_NAME2 "/user/1234"
// publish topic
#define MQTT_ALI_PUB_PAYLOAD "update"
// mqtt client url
#define MQTT_CLIENT_URL "mqtt://iot-06z00iqw9ampzrw.mqtt.iothub.aliyuncs.com:1883"
//mqtt connect status
static int mqtt1_connected = 0, mqtt2_connected = 0;
//semaphore
static liot_sem_t mqtt_semp1, mqtt_semp2;
liot_mqtt_client_t mqtt_cli1 = 0, mqtt_cli2 = 0;
#define LIOT_HTTPC_ENABLE 1
#if LIOT_HTTPC_ENABLE
#define LIOT_HTTP_DEMO_SSL_TEST 0
unsigned char *recv_buf = NULL;
static liot_sem_t http_semp;
const char *HTTPS_CA_CERT = \
{
\
"-----BEGIN CERTIFICATE-----\r\n"
"MIIDdTCCAl2gAwIBAgILBAAAAAABFUtaw5QwDQYJKoZIhvcNAQEFBQAwVzELMAkG\r\n" \
"A1UEBhMCQkUxGTAXBgNVBAoTEEdsb2JhbFNpZ24gbnYtc2ExEDAOBgNVBAsTB1Jv\r\n" \
"b3QgQ0ExGzAZBgNVBAMTEkdsb2JhbFNpZ24gUm9vdCBDQTAeFw05ODA5MDExMjAw\r\n" \
"MDBaFw0yODAxMjgxMjAwMDBaMFcxCzAJBgNVBAYTAkJFMRkwFwYDVQQKExBHbG9i\r\n" \
"YWxTaWduIG52LXNhMRAwDgYDVQQLEwdSb290IENBMRswGQYDVQQDExJHbG9iYWxT\r\n" \
"aWduIFJvb3QgQ0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDaDuaZ\r\n" \
"jc6j40+Kfvvxi4Mla+pIH/EqsLmVEQS98GPR4mdmzxzdzxtIK+6NiY6arymAZavp\r\n" \
"xy0Sy6scTHAHoT0KMM0VjU/43dSMUBUc71DuxC73/OlS8pF94G3VNTCOXkNz8kHp\r\n" \
"1Wrjsok6Vjk4bwY8iGlbKk3Fp1S4bInMm/k8yuX9ifUSPJJ4ltbcdG6TRGHRjcdG\r\n" \
"snUOhugZitVtbNV4FpWi6cgKOOvyJBNPc1STE4U6G7weNLWLBYy5d4ux2x8gkasJ\r\n" \
"U26Qzns3dLlwR5EiUWMWea6xrkEmCMgZK9FGqkjWZCrXgzT/LCrBbBlDSgeF59N8\r\n" \
"9iFo7+ryUp9/k5DPAgMBAAGjQjBAMA4GA1UdDwEB/wQEAwIBBjAPBgNVHRMBAf8E\r\n" \
"BTADAQH/MB0GA1UdDgQWBBRge2YaRQ2XyolQL30EzTSo//z9SzANBgkqhkiG9w0B\r\n" \
"AQUFAAOCAQEA1nPnfE920I2/7LqivjTFKDK1fPxsnCwrvQmeU79rXqoRSLblCKOz\r\n" \
"yj1hTdNGCbM+w6DjY1Ub8rrvrTnhQ7k4o+YviiY776BQVvnGCv04zcQLcFGUl5gE\r\n" \
"38NflNUVyRRBnMRddWQVDf9VMOyGj/8N7yy5Y0b2qvzfvGn9LhJIZJrglfCm7ymP\r\n" \
"AbEVtQwdpf5pLGkkeB6zpxxxYu7KyJesF12KwvhHhm4qxFYxldBniYUr+WymXUad\r\n" \
"DKqC5JlR3XC321Y9YeRq4VzW9v493kHMB65jUr9TU/Qr6cf9tveCX4XSQRjbgbME\r\n" \
"HMUfpIBvFSDJ3gyICh3WZlXi/EjJKSZp4A==\r\n" \
"-----END CERTIFICATE-----"
};
/
* @brief HmacSha256 algorithm
* @param input Input data to be encrypted
* @param ilen Input data length
* @param output Encrypted string
* @param key Key
* @param keylen Key length
* @return void
*/
static void HmacSha256(const unsigned char *input, int ilen, unsigned char *output,const unsigned char *key, int keylen)
{
int i;
mbedtls_sha256_context ctx;
unsigned char k_ipad[ALI_SHA256_KEY_IOPAD_SIZE] = {0};
unsigned char k_opad[ALI_SHA256_KEY_IOPAD_SIZE] = {0};
memset(k_ipad, 0x36, 64);
memset(k_opad, 0x5C, 64);
if ((NULL == input) || (NULL == key) || (NULL == output)) {
return;
}
if (keylen > ALI_SHA256_KEY_IOPAD_SIZE) {
return;
}
for(i=0; i<keylen; i++)
{
if(i>=ALI_SHA256_KEY_IOPAD_SIZE)
{
break;
}
k_ipad[i] ^=key[i];
k_opad[i] ^=key[i];
}
mbedtls_sha256_init(&ctx);
mbedtls_sha256_starts(&ctx, 0);
mbedtls_sha256_update(&ctx, k_ipad, ALI_SHA256_KEY_IOPAD_SIZE);
mbedtls_sha256_update(&ctx, input, ilen);
mbedtls_sha256_finish(&ctx, output);
mbedtls_sha256_starts(&ctx, 0);
mbedtls_sha256_update(&ctx, k_opad, ALI_SHA256_KEY_IOPAD_SIZE);
mbedtls_sha256_update(&ctx, output, ALI_SHA256_DIGEST_SIZE);
mbedtls_sha256_finish(&ctx, output);
mbedtls_sha256_free(&ctx);
}
/
* @brief HTTP request to get request body
* @param productkey Product key
* @param deviceName Device name
* @param productSecret Product secret
* @param requestdata Request data body
* @return INT32 Request result -1表示失败 0表示成功
*/
INT32 liot_getHttpRequestData(char* productkey, char* deviceName, char* productSecret, char* requestdata)
{
int dataLen = 0;
uint8_t* sign = NULL;
int radomData = 0;
uint8_t sign_input_fmt[128] = {0};
char hexstr[65] = {0};
liot_trace("productkey: %s, deviceName: %s, productSecret: %s", productkey, deviceName, productSecret);
if((NULL == productkey) || (NULL == deviceName) || (NULL == productSecret) || (NULL == requestdata))
{
return -1;
}
radomData = rand();
sign = malloc(SIGN_SOURCE_MAXLEN);
if(NULL == sign)
{
return -1;
}
sprintf((char*)sign_input_fmt, "deviceName%sproductKey%srandom%d", deviceName, productkey, radomData);
liot_trace("sign_input_fmt: %s", sign_input_fmt);
HmacSha256(sign_input_fmt, strlen((char*)sign_input_fmt), sign, (uint8_t*)productSecret, strlen(productSecret));
cmsHexToHexStr(hexstr, 64, sign, 32);
liot_trace("hexstr: %s", hexstr);
sprintf(requestdata, "productKey=%s&deviceName=%s&random=%d&sign=%s&signMethod=HmacSha256", productkey, deviceName, radomData, hexstr);
liot_trace("requestdata: %s", requestdata);
free(sign);
dataLen = strlen(requestdata);
return dataLen;
}
/
* @brief Get device secret
* @param inStr Input json string
* @param deviceSecret Device secret
* @return INT32 Request result -1表示失败 0表示成功
*/
INT32 cjsonGetDeviceSercite(char* inStr, char* deviceSecret)
{
if((NULL == inStr) || (NULL == deviceSecret) )
{
liot_trace("Failed to parse JSON data!.\n");
return -1;
}
cJSON *json = cJSON_Parse(inStr);
if (json == NULL)
{
liot_trace("Failed to parse JSON data.\n");
return -1;
}
cJSON *code = cJSON_GetObjectItem(json, "code");
if(NULL != code)
{
liot_trace("Code: %d\n", code->valueint);
}
cJSON *data = cJSON_GetObjectItem(json, "data");
if(NULL != data)
{
cJSON *device_name = cJSON_GetObjectItem(data, "deviceName");
if(device_name != NULL)
{
liot_trace("Device Name: %s\n", device_name->valuestring);
}
cJSON *device_secret = cJSON_GetObjectItem(data, "deviceSecret");
if(NULL != device_secret)
{
liot_trace("Device Secret: %s\n", device_secret->valuestring);
strcpy(deviceSecret, device_secret->valuestring);
}
cJSON *product_key = cJSON_GetObjectItem(data, "productKey");
if(NULL != product_key)
{
liot_trace("Product Key: %s\n", product_key->valuestring);
}
cJSON *message = cJSON_GetObjectItem(json, "message");
liot_trace("Message: %s\n", message->valuestring);
}
else
{
liot_trace("Failed to parse JSON data!!.\n");
return -1;
}
cJSON_Delete(json);
return 0;
}
/
* @brief http event cb
* @param client http client
* @param evt http event
* @param evt_code http event code
* @param arg http event arg
* @retval none
* @note none
*/
static void http_event_cb(liot_http_client_t *client, int evt, int evt_code, void *arg)
{
liot_trace("===http_event_cb=== evt:%d,evt_code:%d,%p", evt, evt_code, client);
switch (evt)
{
/*http session open*/
case LIOT_HTTPC_SESSION_OPEN:
{
if (evt_code != LIOT_HTTPC_SUCCESS)
{
liot_trace("http session open ERROR");
liot_rtos_semaphore_release(http_semp);
}
}
break;
case LIOT_HTTPC_UPLOAD_START:
{
liot_httpc_user_notify(client, LIOT_HTTPC_READ);
}
break;
case LIOT_HTTPC_UPLOAD_END:
{
}
break;
case LIOT_HTTPC_RESPONSE_STATUS:
{
if (evt_code == LIOT_HTTPC_SUCCESS)
{
int resp_code = 0;
int content_length = 0;
int chunk_encode = 0;
char *location = NULL;
char *date = NULL;
liot_httpc_getinfo(client, LIOT_HTTPC_STATUS_CODE, &resp_code);
liot_httpc_getinfo(client, LIOT_HTTPC_CHUNK_ENCODE, &chunk_encode);
liot_httpc_getinfo(client, LIOT_HTTPC_DATE, &date);
if (date != NULL)
{
liot_trace("===http_event_cb=== Date:%s", date);
liot_rtos_free(date);
}
liot_trace("===http_event_cb=== resp_code:%d,chunk_encode:%d", resp_code, chunk_encode);
if (chunk_encode == 0)
{
liot_httpc_getinfo(client, LIOT_HTTPC_CONTENT_LEN, &content_length);
liot_trace("===http_event_cb=== content_length:%d", content_length);
}
else
{
liot_trace("http chunk encode");
}
if (resp_code >= 300 && resp_code < 400)
{
liot_httpc_getinfo(client, LIOT_HTTPC_LOCATION, &location);
liot_trace("===http_event_cb=== location:%s", location);
liot_rtos_free(location);
}
}
else if(evt_code == LIOT_HTTPC_ERR_SOCKET_FAILURE)
{
liot_rtos_semaphore_release(http_semp);
}
}
break;
case LIOT_HTTPC_RESPONSE_COMPLETE:
{
if (evt_code == LIOT_HTTPC_SUCCESS)
{
liot_trace("http transfer success");
}
else
{
liot_trace("http transfer fail");
}
liot_rtos_semaphore_release(http_semp);
}
break;
case LIOT_HTTPC_RESPONSE_TIMEOUT:
{
liot_trace("http response timeout");
liot_rtos_semaphore_release(http_semp);
}
break;
case LIOT_HTTPC_SESSION_CLOSE:
{
liot_trace("http LIOT_HTTPC_SESSION_CLOSE success");
liot_rtos_semaphore_release(http_semp);
}
break;
}
}
#if LIOT_HTTP_DEMO_SSL_TEST
/
* @brief http ssl param cfg
* @param client http client
* @param verify_level https verify level
* @return true if success, false if failed
*/
bool liot_https_param_cfg(liot_http_client_t *client, liot_https_verify_level_e verify_level)
{
if((*client == 0) || (verify_level > 2))
{
liot_trace("http liot_https_param_cfg failed!!!!");
return false;
}
unsigned short hs_timeout = 300;
liot_httpc_setopt(client, LIOT_HTTP_CLIENT_OPT_SSLCTXID, 1); // config ssl id
liot_httpc_setopt(client, LIOT_HTTP_CLIENT_OPT_SSL_VERSION, LIOT_SSL_VERSION_3); // ssl version
liot_httpc_setopt(client, LIOT_HTTP_CLIENT_OPT_SSL_HS_TIMEOUT, hs_timeout);
// liot_httpc_setopt(client, LIOT_HTTP_CLIENT_OPT_SSL_SNI, 0);
liot_trace("%s:%d verify_level:%d ", __FUNCTION__, __LINE__, verify_level);
if (verify_level == LIOT_HTTPS_VERIFY_SERVER)
{
liot_httpc_setopt(client, LIOT_HTTP_CLIENT_OPT_SSL_VERIFY_LEVEL, LIOT_HTTPS_VERIFY_SERVER);
liot_httpc_setopt(client, LIOT_HTTP_CLIENT_OPT_SSL_CACERT_DATA, HTTPS_CA_CERT);
liot_httpc_setopt(client, LIOT_HTTP_CLIENT_OPT_SSL_IGNORE_INVALID_CERT_SIGN, 1);
liot_httpc_setopt(client, LIOT_HTTP_CLIENT_OPT_SSL_IGNORE_CERT_ITEM, 0xffff);
}
else if (verify_level == LIOT_HTTPS_VERIFY_SERVER_CLIENT)
{
liot_trace(">>>>>verify_level error:%d ", verify_level);
return false;
}
liot_trace(">>>>>verify_level:%d ", verify_level);
return true;
}
#endif
static int http_recv_data_cb(liot_http_client_t *client, void *arg, char *data, int size, unsigned char end)
{
liot_trace(">>>>>client:%p", client);
liot_trace("===http_response_write_data_cb=== size:%d", size);
recv_buf = malloc(size + 1);
if (recv_buf != NULL)
{
memset(recv_buf, 0, size + 1);
memcpy(recv_buf, data, size);
liot_trace("===http_response_write_data_cb=== recv_buf:%s", recv_buf);
}
return size;
}
/
* @brief http request send data cb
* @param client http client
* @param arg http request send data cb arg
* @param data http request send data
* @param size http request send data size
* @return int http request send data size
*/
static int http_request_send_data_cb(liot_http_client_t *client, void *arg, char *data, int size)
{
liot_trace(">>>>>client:%p,%d", client, arg);
if ((client == NULL) || (arg == NULL) || (data == NULL) || (size <= 0))
{
return 0;
}
memset(data, 0, size);
memcpy(data, arg, size);
return size;
}
/
* @brief http init
* @param nSim sim id
* @param cid data call id
* @param client http client
* @param httpHost http host
* @param http_method http method
* @return int
*/
INT32 liot_http_init(uint8_t nSim, uint8_t cid, liot_http_client_t *client, char *httpHost, int http_method)
{
liot_httpc_url_s *url = NULL;
if (client == NULL || httpHost == NULL )
{
liot_trace("http liot_http_requeset_data failed!!!!");
return -1;
}
url = liot_rtos_malloc(sizeof(liot_httpc_url_s));
if(NULL == url)
{
liot_trace("http liot_http_requeset_data memary malloc failed!!!!");
return -1;
}
if(liot_httpc_url_parse(httpHost, url) == false)
{
liot_trace("http liot_http_requeset_data liot_httpc_url_parse failed!!!!");
liot_rtos_free(url);
return -1;
}
liot_trace(">>>>>>url:%s,port:%d,path:%s,scheme:%d", url->host, url->port, url->uri, url->scheme);
liot_httpc_setopt(client, LIOT_HTTP_CLIENT_OPT_SIM_ID, nSim);
liot_httpc_setopt(client, LIOT_HTTP_CLIENT_OPT_PDPCID, cid);
liot_httpc_setopt(client, LIOT_HTTP_CLIENT_OPT_METHOD, http_method);
liot_httpc_setopt(client, LIOT_HTTP_CLIENT_OPT_URL, url);
liot_httpc_setopt(client, LIOT_HTTP_CLIENT_OPT_WRITE_FUNC, http_recv_data_cb);
if (http_method == LIOT_HTTPC_METHOD_POST)
{
liot_httpc_setopt(client, LIOT_HTTP_CLIENT_OPT_READ_FUNC, http_request_send_data_cb);
liot_httpc_setopt(client, LIOT_HTTP_CLIENT_OPT_REQUEST_HEADER, "Content-type: application/x-www-form-urlencoded");
}
#if LIOT_HTTP_DEMO_SSL_TEST
if(url->scheme != LIOT_HTTPC_SCHEME_HTTPS)
{
liot_trace("url->scheme =%s", url->scheme);
return -1;
}
if(liot_https_param_cfg(client, LIOT_HTTPS_VERIFY_SERVER) == false)
{
liot_trace("liot_https_param_cfg failed!!!!");
return -1;
}
#endif
return 0;
}
int liot_http_request(uint8_t nSim, uint8_t cid, char *httpHost, int http_method, char *postData)
{
liot_http_client_t http_client = 0;
int ret = -1;
uint32 postDataLen = 0;
liot_rtos_semaphore_create(&http_semp, 0);
if ((httpHost == NULL) || (postData == NULL))
{
liot_trace("http liot_http_requeset_data failed!!!!");
return -1;
}
if (liot_httpc_new(&http_client, http_event_cb, NULL) != LIOT_HTTPC_SUCCESS)
{
liot_trace("http client create failed!!!!");
return -1;
}
ret = liot_http_init(nSim, cid, &http_client, httpHost, http_method);
if(ret != 0)
{
liot_trace("http liot_http_requeset_data failed!!!!");
return -1;
}
if (http_method == LIOT_HTTPC_METHOD_POST)
{
liot_trace("http liot_http_requeset_data postData:%d", postData);
postDataLen = strlen(postData);
liot_httpc_setopt(&http_client, LIOT_HTTP_CLIENT_OPT_READ_DATA, postData);
liot_httpc_setopt(&http_client, LIOT_HTTP_CLIENT_OPT_UPLOAD_LEN, postDataLen);
}
if(liot_httpc_perform(&http_client) == LIOT_HTTPC_SUCCESS)
{
liot_trace("liot_rtos_semaphore_wait");
liot_rtos_semaphore_wait(http_semp, LIOT_WAIT_FOREVER);
liot_trace("http perform client success");
ret = 0;
}
else
{
liot_trace("http perform client fail");
ret = -1;
}
liot_httpc_stop(&http_client);
liot_rtos_semaphore_wait(http_semp, LIOT_WAIT_FOREVER);
liot_httpc_release(&http_client);
liot_rtos_semaphore_release(http_semp);
liot_rtos_semaphore_delete(http_semp);
http_semp = NULL;
return ret;
}
int liot_get_mqtt_devSercite(int nSim, int cid, char* productkey, char* deviceName, char* productSecret, char *devSercite)
{
int ret = -1;
char *reqData = NULL;
if (devSercite == NULL)
{
liot_trace("http liot_http_requeset_data failed!!!!");
return -1;
}
reqData = liot_rtos_malloc(256);
if (reqData == NULL)
{
liot_trace("http liot_http_requeset_data failed!!!!");
return -1;
}
ret = liot_getHttpRequestData(productkey, deviceName, productSecret, reqData);
if(ret <= 0)
{
liot_trace("http liot_http_requeset_data failed!!!!");
return -1;
}
ret = liot_http_request(nSim, cid, HTTP_UARL, LIOT_HTTPC_METHOD_POST, reqData);
if(ret != 0)
{
liot_trace("http liot_http_requeset_data failed!!!!");
ret = -1;
}
else
{
ret = cjsonGetDeviceSercite((char*)recv_buf, devSercite);
}
if (recv_buf != NULL)
{
liot_rtos_free(recv_buf);
}
if(reqData != NULL)
{
liot_rtos_free(reqData);
}
return ret;
}
/
* @brief Get MQTT client information
* @param productkey Product key
* @param deviceName Device name
* @param deviceSecret Device secret
* @param clientId MQTT client ID
* @param username MQTT login username
* @param password MQTT login password
* @return INT32 Request result -1表示失败 0表示æå
*/
int32_t getMqttClientInfo(char* productkey, char* deviceName, char* deviceSecret,
char* clientId, char* username, char* password)
{
uint8_t* sign = NULL;
// int radomData = 0;
char clientId_in_fmt[256] = {0};
char userName_in_fmt[65] = {0};
uint8_t sign_input_fmt[128] = {0};
if((NULL == productkey) || (NULL == deviceName) || (NULL == deviceSecret) ||
(NULL == clientId) || (NULL == username) || (NULL == password) )
{
return -1;
}
// radomData = rand();
sign = malloc(SIGN_SOURCE_MAXLEN);
if(NULL == sign)
{
return -1;
}
sprintf(clientId_in_fmt, "%s&%s|securemode=3,signmethod=hmacsha256|",deviceName, productkey);
liot_trace("clientId_in_fmt: %s", clientId_in_fmt);
strcpy(clientId, clientId_in_fmt);
sprintf(userName_in_fmt, "%s&%s", deviceName, productkey);
liot_trace("userName_in_fmt: %s", userName_in_fmt);
strcpy(username, userName_in_fmt);
sprintf((char*)sign_input_fmt, "clientId%sdeviceName%sproductKey%s",userName_in_fmt, deviceName, productkey);
liot_trace("sign_input_fmt: %s", sign_input_fmt);
HmacSha256(sign_input_fmt, strlen((char*)sign_input_fmt), sign, (uint8_t*) deviceSecret, strlen(deviceSecret));
cmsHexToHexStr(password, 64, sign, 32);
liot_trace("password: %s", password);
free(sign);
return 0;
}
/
* @brief Get mqtt client information
* @param fileName File name
* @param url Post request URL
* @param productKey Product key
* @param deviceName Device name
* @param productSecret Product secret
* @param clientId Client ID
* @param username Username
* @param password Password
* @return int32_t
*/
int32_t liot_get_mqtt_parma(int nSim, int cid, char* fileName, char* url, char* productKey, char* deviceName, char* productSecret,
char* clientId, char* username, char* password)
{
LFILE fd = 0;
int32_t result = 0;
char deviceSecret[DEVICESECRET_MAXLEN] = {0};
// int32_t deviceSecretLen = 0;
if(clientId == NULL || username == NULL || password == NULL || url == NULL ||
productKey == NULL || deviceName == NULL || productSecret == NULL || fileName == NULL)
{
liot_trace("clientId, username or password is NULL");
return -1;
}
liot_trace("mqtt test start");
// get device secret
if(0 == liot_file_exist(fileName))
{
fd = liot_fopen(fileName, "r");
}
else
{
fd = liot_fopen(fileName, "wb+");
}
result = liot_fread(deviceSecret, sizeof(deviceSecret), 1, fd);
if (result <= 0)
{
result = liot_get_mqtt_devSercite( nSim, cid, productKey, deviceName, productSecret, deviceSecret);
if(result != 0)
{
liot_trace("httpPostData failed!!");
}
else
{
liot_trace(" deviceSecre: %s",deviceSecret);
}
result = liot_fwrite(deviceSecret, strlen(deviceSecret), 1, fd);
if(result != strlen(deviceSecret))
{
liot_trace(" LIOT_NV_MQTT_CFG write fail");
}
result = liot_fclose(fd);
if (result != LIOT_FS_OK)
{
liot_trace("=== close file fail. ret(%d) ===", result);
}
}
result = getMqttClientInfo(productKey, deviceName, deviceSecret, clientId, username, password);
return result;
}
#endif
static void liot_mqtt_event_cb(liot_mqtt_client_t *client, int event_type, void *arg, void *data)
{
liot_trace("===liot_mqtt_event_cb!!!! = %d",event_type);
switch(event_type)
{
case LIOT_MQTT_OPEN_EVENT:
case LIOT_MQTT_CONNECT_EVENT:
{
int status = *(int *)data;
if((*client) == mqtt_cli1)
{
if ( status == 0)
{
mqtt1_connected = 1;
}
liot_trace("mqtt_cli1 connect result = %d", status);
liot_rtos_semaphore_release(mqtt_semp1);
}
else if((*client) == mqtt_cli2)
{
if ( status == 0)
{
mqtt2_connected = 1;
}
liot_trace("mqtt_cli2 connect result = %d", status);
liot_rtos_semaphore_release(mqtt_semp2);
}
liot_trace("MQTT_OPEN_EVENT data=%d", *(int *)data);
}
break;
case LIOT_MQTT_CLOSE_EVENT:
{
if((*client) == mqtt_cli1)
{
mqtt1_connected = 0;
liot_trace("mqtt_cli1 disconnected");
//liot_rtos_semaphore_release(mqtt_semp1);
}
else if((*client) == mqtt_cli2)
{
mqtt2_connected = 0;
liot_trace("mqtt_cli2 disconnected");
//liot_rtos_semaphore_release(mqtt_semp2);
}
liot_trace("MQTT_CLOSE_EVENT data=%d", *(int *)data);
}
break;
case LIOT_MQTT_RECONNECT_EVENT:
{
if((*client) == mqtt_cli1)
{
liot_trace("mqtt_cli1 reconnect");
}
else if((*client) == mqtt_cli2)
{
liot_trace("mqtt_cli2 reconnect");
}
liot_trace("MQTT_RECONNECT_EVENT");
}
break;
case LIOT_MQTT_PUB_EVENT:
case LIOT_MQTT_SUB_EVENT:
case LIOT_MQTT_UNSUB_EVENT:
{
if((*client) == mqtt_cli1)
{
liot_rtos_semaphore_release(mqtt_semp1);
liot_trace("mqtt_requst_result_cb *client = %p", *client);
}
else if((*client) == mqtt_cli2)
{
liot_rtos_semaphore_release(mqtt_semp2);
liot_trace("mqtt_requst_result_cb *client = %p", *client);
}
liot_trace("MQTT_PUB_EVENT pkt_id=%d", *(int *)data);
}
break;
case LIOT_MQTT_DISCONNECT_EVENT:
{
if((*client) == mqtt_cli1)
{
liot_rtos_semaphore_release(mqtt_semp1);
liot_trace("mqtt_disconnect_result_cb *client = %p", *client);
}
else if((*client) == mqtt_cli2)
{
liot_rtos_semaphore_release(mqtt_semp2);
liot_trace("mqtt_disconnect_result_cb *client = %p", *client);
}
liot_trace("MQTT_DISCONNECT_EVENT data=%d", *(int *)data);
}
break;
case LIOT_MQTT_PUBLISH_EVENT:
{
liot_mqtt_recvMessage *msg = (liot_mqtt_recvMessage *)data;
liot_trace("===mqtt_recv_data===== *client = %p", *client);
liot_trace("topic: %s, len: %d", msg->topicName, msg->topicLen);
liot_trace("payload: %s, len: %d", msg->payload, msg->payloadlen);
liot_trace("qos: %d", msg->qos);
liot_trace("retain: %d", msg->retained);
liot_trace("dup: %d", msg->dup);
liot_trace("msgid: %d", msg->id);
}
break;
default:
break;
}
}
/
* @brief Callback function for network indication events.
* This function is a test function for multi - task network callback.
*
* @param nSim SIM card number.
* @param ind_type Indication event type.
* @param ctx Pointer to the context data.
*/
static void liot_nw_ind_callback(uint8_t nSim, unsigned int ind_type, void *ctx)
{
char csq = 99; // Default signal quality value
liot_nw_common_reg_status_info_s *liot_nw_msg = NULL;
liot_nw_nitz_time_info_s *liot_nw_nitz_time_info = NULL;
liot_trace("nSim=%d, ind_type=%x", nSim, ind_type);
switch (ind_type)
{
case LIOT_NW_SIGNAL_QUALITY_IND:
{
csq = *((char *)ctx); // Get the signal quality value
liot_trace("csq=%d", csq);
}
break;
case LIOT_NW_DATA_REG_STATUS_IND:
{
liot_nw_msg = (liot_nw_common_reg_status_info_s *)ctx; // Cast the context pointer
liot_trace("regState=%d, lac=0x%X, cid=0x%X, act=%d",
liot_nw_msg->state,
liot_nw_msg->lac,
liot_nw_msg->cid,
liot_nw_msg->act);
}
break;
case LIOT_NW_NITZ_TIME_UPDATE_IND:
{
liot_nw_nitz_time_info = (liot_nw_nitz_time_info_s *)ctx; // Cast the context pointer
liot_trace(
"nitz_time=%s, abs_time=%ld", liot_nw_nitz_time_info->nitz_time, liot_nw_nitz_time_info->abs_time);
}
break;
}
}
static int datacall_start()
{
int ret = 0;
int profile_idx = 1;
liot_data_call_info_t info;
uint8_t nSim = 0;
// Register network callback
ret = liot_nw_register_cb(liot_nw_ind_callback);
if (LIOT_NW_SUCCESS != ret)
{
liot_trace("liot_nw_register_cb failed, ret is 0x%x.", ret);
}
// Try to wait for network registration, up to 10 attempts
int times = 0;
while (LIOT_DATACALL_SUCCESS != (ret = liot_network_register_wait(nSim, 120)) && times < 10)
{
times++;
liot_rtos_task_sleep_s(1);
}
if (LIOT_DATACALL_SUCCESS != ret)
{
liot_trace("====network register failure!!!!!====");
return -1;
}
liot_trace("===start data call begin====");
ret = liot_start_data_call(nSim, profile_idx, LIOT_DATA_TYPE_IP, "", "", "", LIOT_DATA_AUTH_TYPE_NONE);
if (ret != LIOT_DATACALL_SUCCESS)
{
liot_trace("liot_start_data_call ret: 0x%x.", ret);
return -1;
}
liot_trace("===start data call end====");
liot_rtos_task_sleep_s(5);
ret = liot_get_data_call_info(nSim, profile_idx, &info);
if (ret != 0)
{
liot_stop_data_call(nSim, profile_idx);
liot_trace("liot_get_data_call_info ret: %d", ret);
return -1;
}
liot_trace("info->cid: %d", info.cid);
liot_trace("info->ip_version: %d", info.ip_version);
liot_trace("info->v4.state: %d", info.v4.state);
// inet_ntop(AF_INET, &info.v4.addr.ip, ip4_addr_str, sizeof(ip4_addr_str));
// liot_trace("info.v4.addr.ip: %s", ip4_addr_str);
// inet_ntop(AF_INET, &info.v4.addr.pri_dns, ip4_addr_str, sizeof(ip4_addr_str));
// liot_trace("info.v4.addr.pri_dns: %s", ip4_addr_str);
// inet_ntop(AF_INET, &info.v4.addr.sec_dns, ip4_addr_str, sizeof(ip4_addr_str));
// liot_trace("info.v4.addr.sec_dns: %s", ip4_addr_str);
return 0;
}
void liot_mqtt_ali_client1_thread(void *arg)
{
int ret = 0;
liot_mqtt_client_option options = {0};
char clientid[CLIENTID_MAXLEN] = {0};
char username[USERNAME_MAXLEN] = {0};
char password[PASSWORD_MAXLEN] = {0};
int mqttState = 0;
int cid = 1;
int nSim = 0;
int free_size = liot_xPortGetFreeHeapSize();
liot_trace("free size :%d",free_size);
#if LIOT_HTTPC_ENABLE
ret = liot_get_mqtt_parma(nSim, cid, LIOT_NV_MQTT1_CFG, HTTP_UARL, ALI_PRODUCT_KEY, ALI_DEVICE_NAME1, ALI_PRODUCT_SECRET,
clientid, username, password);
if (ret != 0)
{
liot_trace("get mqtt parma failed");
return;
}
#endif
options.version = LIOT_MQTT_VERSION_4; // 3.1.1
options.pdp_cid = 1;
options.ssl_enable = false,
options.ssl_cfg = NULL;
options.clean_session = 1;
options.kalive_time = 30; //mqtt keepalive time unit: second
options.delivery_time = 5; //mqtt message delivery time
options.delivery_cnt = 3;
options.will_flag = 0;
options.will_qos = 0;
options.will_retain = 0;
options.ping_timeout = 5;
options.client_id = clientid;
options.client_user = username;
options.client_pass = password;
memset(options.will_topic, 0x00, 256);
memset(options.will_message, 0x00, 256);
liot_rtos_semaphore_create(&mqtt_semp1, 0);
int reconnectCnt = 0;
if (liot_mqtt_client_init_ex(&mqtt_cli1, cid, liot_mqtt_event_cb, NULL) != LIOT_MQTTCLIENT_SUCCESS)
{
liot_trace("mqtt client init failed!!!!");
}
ret = liot_mqtt_connect(&mqtt_cli1,
MQTT_CLIENT_URL,
NULL,
NULL,
(liot_mqtt_client_option *)&options,
NULL);
if (ret == LIOT_MQTTCLIENT_WOUNDBLOCK)
{
liot_trace("====wait connect result");
liot_rtos_semaphore_wait(mqtt_semp1, LIOT_WAIT_FOREVER);
if (mqtt1_connected == 0)
{
liot_mqtt_client_deinit(&mqtt_cli1);
}
}
else
{
liot_trace("===mqtt connect failed ,ret = %d", ret);
}
while(1)
{
// app_show_mem(NULL, __LINE__);
mqttState = liot_mqtt_client_state(&mqtt_cli1);
liot_trace("=====MQTT STATE-->>%d", mqttState);
if(mqttState == MQTT_CONN_CONNECTED)
{
if (liot_mqtt_sub_unsub(&mqtt_cli1, MQTT_ALI_SUB_TOPIC1, 0, NULL, NULL, 1) ==
LIOT_MQTTCLIENT_WOUNDBLOCK)
{
liot_trace("======wait subscrible result");
liot_rtos_semaphore_wait(mqtt_semp1, LIOT_WAIT_FOREVER);
}
if (liot_mqtt_publish(&mqtt_cli1,
MQTT_ALI_PUB_TOPIC1,
MQTT_ALI_PUB_PAYLOAD,
strlen(MQTT_ALI_PUB_PAYLOAD),
0,
0,
NULL,
NULL) == LIOT_MQTTCLIENT_WOUNDBLOCK)
{
liot_trace("======wait publish result");
liot_rtos_semaphore_wait(mqtt_semp1, LIOT_WAIT_FOREVER);
}
if (liot_mqtt_sub_unsub(&mqtt_cli1, MQTT_ALI_SUB_TOPIC1, 0, NULL, NULL, 0) ==
LIOT_MQTTCLIENT_WOUNDBLOCK)
{
liot_trace("=====wait unsubscrible result");
liot_rtos_semaphore_wait(mqtt_semp1, LIOT_WAIT_FOREVER);
}
}
//reconnect mqtt
else if((mqttState == MQTT_CONN_DISCONNECTED) || (mqttState == MQTT_CONN_DEFAULT))
{
liot_trace("=====MQTT reconnect");
ret = datacall_start();
if (ret != 0) {
liot_trace("datacall_start failed");
break;
}
reconnectCnt++;
if (liot_mqtt_client_init(&mqtt_cli1, cid) != LIOT_MQTTCLIENT_SUCCESS)
{
liot_trace("mqtt client init failed!!!!");
}
ret = liot_mqtt_connect(&mqtt_cli1,
MQTT_CLIENT_URL,
NULL,
NULL,
(liot_mqtt_client_option *)&options,
NULL);
liot_trace("=====MQTT eeeee-->> ret: %d", ret);
if (ret == LIOT_MQTTCLIENT_WOUNDBLOCK)
{
uint32_t count = 0;
liot_rtos_semaphore_get_cnt(mqtt_semp1, &count);
liot_trace("====wait connect result liot_rtos_semaphore_get_cnt = %d", count);
liot_rtos_semaphore_wait(mqtt_semp1, LIOT_WAIT_FOREVER);
if (mqtt1_connected == 0)
{
liot_mqtt_client_deinit(&mqtt_cli1);
}
else
{
reconnectCnt = 0;
}
}
else
{
liot_trace("===mqtt connect failed ,ret = %d", ret);
}
if (reconnectCnt > 2)
{
liot_trace("===mqtt reconnect failed ,exit");
break;
}
}
else if(mqttState == MQTT_CONN_RECONNECTING)
{
liot_trace("=====MQTT STATE-->> RECONNECTING");
}
else
{
liot_trace("=====MQTT STATE-->> other: %d", mqttState);
}
liot_rtos_task_sleep_s(5);
}
if (mqtt1_connected == 1 && liot_mqtt_disconnect(&mqtt_cli1, NULL, NULL) == LIOT_MQTTCLIENT_WOUNDBLOCK)
{
liot_trace("=====wait disconnect result");
liot_rtos_semaphore_wait(mqtt_semp1, LIOT_WAIT_FOREVER);
}
liot_rtos_task_sleep_s(5);
liot_mqtt_client_deinit(&mqtt_cli1);
mqtt1_connected = 0;
liot_rtos_semaphore_delete(mqtt_semp1);
mqtt_semp1 = NULL;
liot_rtos_task_sleep_s(5);
liot_trace("mqtt test end");
liot_rtos_task_delete(NULL);
return;
}
void liot_mqtt_ali_client2_thread(void *arg)
{
int ret = 0;
liot_mqtt_client_option options = {0};
char clientid[CLIENTID_MAXLEN] = {0};
char username[USERNAME_MAXLEN] = {0};
char password[PASSWORD_MAXLEN] = {0};
int mqttState = 0;
int cid = 1;
int nSim = 0;
int free_size = liot_xPortGetFreeHeapSize();
liot_trace("free size :%d",free_size);
#if LIOT_HTTPC_ENABLE
ret = liot_get_mqtt_parma(nSim, cid, LIOT_NV_MQTT2_CFG, HTTP_UARL, ALI_PRODUCT_KEY, ALI_DEVICE_NAME2, ALI_PRODUCT_SECRET,
clientid, username, password);
if (ret != 0)
{
liot_trace("get mqtt parma failed");
return;
}
#endif
options.version = LIOT_MQTT_VERSION_4; // 3.1.1
options.pdp_cid = 1;
options.ssl_enable = false,
options.ssl_cfg = NULL;
options.clean_session = 1;
options.kalive_time = 30; // mqtt keep alive time, unit: s
options.delivery_time = 5; // mqtt message delivery time, unit: s
options.delivery_cnt = 3;
options.will_flag = 0;
options.will_qos = 0;
options.will_retain = 0;
options.ping_timeout = 5;
options.client_id = clientid;
options.client_user = username;
options.client_pass = password;
memset(options.will_topic, 0x00, 256);
memset(options.will_message, 0x00, 256);
liot_rtos_semaphore_create(&mqtt_semp2, 0);
int reconnectCnt = 0;
if (liot_mqtt_client_init_ex(&mqtt_cli2, cid, liot_mqtt_event_cb, NULL) != LIOT_MQTTCLIENT_SUCCESS)
{
liot_trace("mqtt client init failed!!!!");
}
ret = liot_mqtt_connect(&mqtt_cli2,
MQTT_CLIENT_URL,
NULL,
NULL,
(liot_mqtt_client_option *)&options,
NULL);
if (ret == LIOT_MQTTCLIENT_WOUNDBLOCK)
{
liot_trace("====wait connect result");
liot_rtos_semaphore_wait(mqtt_semp2, LIOT_WAIT_FOREVER);
if (mqtt2_connected == 0)
{
liot_mqtt_client_deinit(&mqtt_cli2);
}
}
else
{
liot_trace("===mqtt connect failed ,ret = %d", ret);
}
while(1)
{
//app_show_mem(NULL, __LINE__);
mqttState = liot_mqtt_client_state(&mqtt_cli2);
liot_trace("=====MQTT STATE-->>%d", mqttState);
if(mqttState == MQTT_CONN_CONNECTED)
{
if (liot_mqtt_sub_unsub(&mqtt_cli2, MQTT_ALI_SUB_TOPIC2, 1, NULL, NULL, 1) ==
LIOT_MQTTCLIENT_WOUNDBLOCK)
{
liot_trace("======wait subscrible result");
liot_rtos_semaphore_wait(mqtt_semp2, LIOT_WAIT_FOREVER);
}
if (liot_mqtt_publish(&mqtt_cli2,
MQTT_ALI_PUB_TOPIC2,
MQTT_ALI_PUB_PAYLOAD,
strlen(MQTT_ALI_PUB_PAYLOAD),
1,
0,
NULL,
NULL) == LIOT_MQTTCLIENT_WOUNDBLOCK)
{
liot_trace("======wait publish result");
liot_rtos_semaphore_wait(mqtt_semp2, LIOT_WAIT_FOREVER);
}
if (liot_mqtt_sub_unsub(&mqtt_cli2, MQTT_ALI_SUB_TOPIC2, 1, NULL, NULL, 0) ==
LIOT_MQTTCLIENT_WOUNDBLOCK)
{
liot_trace("=====wait unsubscrible result");
liot_rtos_semaphore_wait(mqtt_semp2, LIOT_WAIT_FOREVER);
}
}
//reconnect mqtt
else if((mqttState == MQTT_CONN_DISCONNECTED) || (mqttState == MQTT_CONN_DEFAULT))
{
liot_trace("=====MQTT reconnect");
ret = datacall_start();
if (ret != 0) {
liot_trace("datacall_start failed");
break;
}
reconnectCnt++;
if (liot_mqtt_client_init(&mqtt_cli2, cid) != LIOT_MQTTCLIENT_SUCCESS)
{
liot_trace("mqtt client init failed!!!!");
}
ret = liot_mqtt_connect(&mqtt_cli2,
MQTT_CLIENT_URL,
NULL,
NULL,
(liot_mqtt_client_option *)&options,
NULL);
liot_trace("=====MQTT eeeee-->> ret: %d", ret);
if (ret == LIOT_MQTTCLIENT_WOUNDBLOCK)
{
liot_trace("====wait connect result");
liot_rtos_semaphore_wait(mqtt_semp2, LIOT_WAIT_FOREVER);
if (mqtt2_connected == 0)
{
liot_mqtt_client_deinit(&mqtt_cli2);
}
else
{
reconnectCnt = 0;
}
}
else
{
liot_trace("===mqtt connect failed ,ret = %d", ret);
}
if (reconnectCnt > 2)
{
liot_trace("===mqtt reconnect failed ,exit");
break;
}
}
else if(mqttState == MQTT_CONN_DISCONNECTED)
{
liot_trace("=====MQTT STATE-->> CONNECTING");
}
else if(mqttState == MQTT_CONN_RECONNECTING)
{
liot_trace("=====MQTT STATE-->> RECONNECTING");
}
else
{
liot_trace("=====MQTT STATE-->> other: %d", mqttState);
}
liot_rtos_task_sleep_s(5);
}
if (mqtt2_connected == 1 && liot_mqtt_disconnect(&mqtt_cli2, NULL, NULL) == LIOT_MQTTCLIENT_WOUNDBLOCK)
{
liot_trace("=====wait disconnect result");
liot_rtos_semaphore_wait(mqtt_semp2, LIOT_WAIT_FOREVER);
}
liot_rtos_task_sleep_s(5);
liot_mqtt_client_deinit(&mqtt_cli2);
mqtt2_connected = 0;
liot_rtos_semaphore_delete(mqtt_semp2);
mqtt_semp2 = NULL;
liot_rtos_task_sleep_s(5);
liot_trace("mqtt test end");
liot_rtos_task_delete(NULL);
return;
}
liot_task_t mqttClient1Task = NULL, mqttClient2Task = NULL;
#define LIOT_MQTT_TASK_PRIORITY APP_PRIORITY_NORMAL
void liot_mqtt_ali_app_thread(void *arg)
{
int ret = -1;
ret = datacall_start();
if (ret != 0) {
liot_trace("datacall_start failed");
liot_rtos_task_delete(NULL);
return;
}
liot_rtos_task_sleep_s(10);
liot_rtos_task_create(&mqttClient2Task, 1024*6, LIOT_APP_TASK_PRIORITY,"mqtt_ali_client2_thread", &liot_mqtt_ali_client2_thread, NULL);
liot_rtos_task_sleep_s(3);
liot_rtos_task_create(&mqttClient1Task, 1024*6, LIOT_APP_TASK_PRIORITY,"mqtt_ali_client1_thread", &liot_mqtt_ali_client1_thread, NULL);
liot_rtos_task_sleep_s(1);
while(1)
{
liot_rtos_task_sleep_s(20);
//app_show_mem(NULL, __LINE__);
}
}
运行结果：
如上图所示，连接aliyun平台，订阅/取消订阅成功，成功接收到发送的数据。mqtt通讯正常。