请求方式：GET
URL: http://api.heclouds.com/devices/device_id/datapoints
服务器或上位机下发主题报文（控制下位机）：
API函数：
请求方式：POST
URL: http://api.heclouds.com/mqtt?topic=xxx
以上2个网络通讯的API函数至关重要，就是实现常规情况下OneNet物联网开发的关键性技术支持。（情况允许的条件下，建议读者朋友们去好好研读一下技术文档，将会为之后的开发大大助力）
三、下位机外设驱动
3.1 ESP8266模块
作者采用的ESP8266模块为ESP8266NodeMCU，是需要进行烧入AT固件，才能实现目标网络通讯。作为常见的物联网开发模块，ESP8266的出现大大降低了物联网开发的难度系数，也普及了物联网的发展。
AT指令最早在蓝牙模块上接触过，所谓AT指令实质上就是一些起控制作用的特殊字符串。模块可以通过AT指令控制搭配使用源代码API函数开发，总体开发速度快，难度较低。
不同厂商芯片的AT固件可能有所不同，但是指令基本一致（作者使用的是乐鑫的）。
说明：由于篇幅有限，这里就不和大家单独详细介绍AT指令。指令的详细参数及使用说明请参考官方文档：ESP8266 AT指令集。
3.2 OLED模块
本项目中0.96寸OLED模块的使用仅为显示DHT11传感器采集到的温湿度信息，以此来对比是否和服务器端以及上位机APP端的数据一致性。对其使用有不是太了解的读者朋友可以参考，作者另一篇基础教学博客：(2条消息) 【强烈推荐】基于stm32的OLED各种显示实现（含动态图）_混分巨兽龙某某的博客-CSDN博客_oled显示图片程序 【强烈推荐】基于stm32的OLED各种显示实现（含动态图）_混分巨兽龙某某的博客-CSDN博客_oled显示图片程序")
本项目的代码都是基于作者以前基础教学上的项目代码搭建而成，保证读者朋友可以实现快速复现。
3.3 DHT11模块
本项目中DHT11为下位机MCU采集周围环境温度和湿度的传感器，当然，条件允许的情况下还可以附加很多环境传感器（比如：烟雾传感器，环境光传感器，二氧化碳传感器等等）。当然得益于OneNet平台的布局，本项目教学的底层逻辑支持读者朋友的自我DIY，实现自主化的物联网产品设计。
DHT11模块驱动参考博客：基于stm32的太空人温湿度时钟项目——DHT11（HAL库）_混分巨兽龙某某的博客-CSDN博客
3.4 KEY和LED
KEY和LED都是源于作者正点原子精英版开发板上自备的（如果和作者同款开发板移植开发将会特别简单快速），属于最基本的GPIO操作相信各位应该都是掌握的
特别注意：
（1）这里的KEY按键从设计逻辑上就可以看出应该是需要采用外部中断的；
（2）KEY按下之后会改变LED的亮灭状态，为了同步上位机此时的LED状态，所以需要触发串口通讯中断（考虑嵌套中断情况时候中断优先级的安排）。
四、CubeMX配置
1、RCC配置外部高速晶振（精度更高）——HSE；
2、SYS配置：Debug设置成Serial Wire（否则可能导致芯片自锁）；
3、TIM2配置：由上面可知DHT11的使用需要us级的延迟函数,HAL库自带只有ms的，所以需要自己设计一个定时器；
4、I2C2配置：作为OLED的通讯方式；
5、UART1和UART3配置：MCU分别与电脑和ESP8266通讯（记得开启串口通信中断）；
6、设置KEY0按键PE4为外部中断（根据自己的开发板来确定）
7、GPIO配置：PE0设置为DHT11的DATA端，PE5为LED，并且设置ESP8266的EN和RST（PB7和PB9）；
8、时钟树配置
五、代码与解析
5.1 OLED与DHT11模块代码
受篇幅限制OLED与DHT11部分的代码，这里就不展示了。如果有不懂这部分原理与代码的读者朋友可以参考本人的另一篇博客。博客地址：基于stm32的太空人温湿度时钟项目——DHT11（HAL库）_混分巨兽龙某某的博客-CSDN博客
5.2 ESP8266模块代码
ESP8266部分的代码主要是借助串口通讯AT指令与ESP8266模块（刷入AT固件的）与OneNet平台进行信息交互（包含ESP8266初始化、数据发送，指令发送和数据缓存清除等）。
esp8266.h代码：
#ifndef _ESP8266_H_
#define _ESP8266_H_
#include "main.h"
#include "usart.h"
#include<string.h>
#include<stdio.h>
#include<stdbool.h>
#define ESP8266_WIFI_INFO "AT+CWJAP=\"NJUST\",\"768541ly\"\r\n" //连接上自己的wifi热点：WiFi名和密码
#define ESP8266_ONENET_INFO "AT+CIPSTART=\"TCP\",\"183.230.40.39\",6002\r\n" //连接上OneNet的MQTT
#define OK 0 //接收完成标志
#define OUTTIME 1 //接收未完成标志
void ESP8266_Clear(void); //清空缓存
void ESP8266_Init(void); //esp8266初始化
_Bool ESP8266_SendCmd(char *cmd, char *res);//发送数据
unsigned char *ESP8266_GetIPD(unsigned short timeOut);
void ESP8266_SendData(unsigned char *data, unsigned short len);
#endif
esp8266.c代码：
#include "esp8266.h"
unsigned char ESP8266_Buf[128]; //定义一个数组作为esp8266的数据缓冲区
unsigned short esp8266_cnt = 0, esp8266_cntPre = 0; //定义两个计数值：此次和上一次
unsigned char a_esp8266_buf;
/**
* @brief esp8266初始化
* @param 无
* @retval 无
*/
void ESP8266_Init(void)
{
ESP8266_Clear();
printf("1. 测试AT启动\r\n"); //AT：测试AT启动
while(ESP8266_SendCmd("AT\r\n", "OK"))
HAL_Delay(500);
printf("2. 设置WiFi模式（CWMODE）\r\n"); //查询/设置 Wi-Fi 模式：设置WiFi模式为Station模式
while(ESP8266_SendCmd("AT+CWMODE=1\r\n", "OK"))
HAL_Delay(500);
printf("3. AT+CWDHCP\r\n"); //启用/禁用 DHCP
while(ESP8266_SendCmd("AT+CWDHCP=1,1\r\n", "OK"))
HAL_Delay(500);
printf("4. 连接WiFi热点（CWJAP）\r\n");
while(ESP8266_SendCmd(ESP8266_WIFI_INFO, "GOT IP"))
HAL_Delay(500);
printf("5. 建立TCP连接（CIPSTART）\r\n");
while(ESP8266_SendCmd(ESP8266_ONENET_INFO, "CONNECT"))
HAL_Delay(500);
printf("6. ESP8266 Init OK\r\n");
}
/**
* @brief 清空缓存
* @param 无
* @retval 无
*/
void ESP8266_Clear(void)
{
memset(ESP8266_Buf, 0, sizeof(ESP8266_Buf)); //将数组中的元素全部初始化为0，
}
/**
* @brief 等待接收完成
* @param 无
* @retval OK：表示接收完成；OUTTIME：表示接收超时完成
* 进行循环调用，检测接收是否完成
*/
_Bool ESP8266_WaitRecive(void)
{
if(esp8266_cnt == 0) //如果当前接收计数为0 则说明没有处于接收数据中，所以直接跳出，结束函数
return OUTTIME;
if(esp8266_cnt == esp8266_cntPre) //如果上一次的值和这次相同，则说明接收完毕
{
esp8266_cnt = 0; //清0接收计数
return OK; //返回接收完成标志
}
else //如果不相同，则将此次赋值给上一次，并返回接收未完成标志
{
esp8266_cntPre = esp8266_cnt;
return OUTTIME;
}
}
/**
* @brief 发送命令
* @param cmd：表示命令；res：需要检查的返回指令
* @retval 0：表示成功；1：表示失败
*/
_Bool ESP8266_SendCmd(char *cmd, char *res)
{
unsigned char timeOut = 200;
HAL_UART_Transmit(&huart3, (unsigned char *)cmd, strlen((const char *)cmd),0xffff);
while(timeOut--)
{
if(ESP8266_WaitRecive() == OK) //如果收到数据
{
printf("%s",ESP8266_Buf);
if(strstr((const char *)ESP8266_Buf, res) != NULL) //如果检索到关键词,清空缓存
{
ESP8266_Clear();
return 0;
}
}
HAL_Delay(10);
}
return 1;
}
/**
* @brief 数据发送
* @param data：待发送的数据；len：待发送的数据长度
* @retval 无
*/
void ESP8266_SendData(unsigned char *data, unsigned short len)
{
char cmdBuf[32];
ESP8266_Clear(); //清空接收缓存
sprintf(cmdBuf, "AT+CIPSEND=%d\r\n", len); //发送命令,sprintf()函数用于将格式化的数据写入字符串
if(!ESP8266_SendCmd(cmdBuf, ">")) //收到‘>’时可以发送数据
{
HAL_UART_Transmit(&huart3, data, len,0xffff); //发送设备连接请求数据
}
}
/**
* @brief 获取平台返回的数据
* @param 等待的时间
* @retval 平台返回的数据，不同网络设备返回的格式不同，需要进行调试，如：ESP8266的返回格式为："+IPD,x:yyy",x表示数据长度，yyy表示数据内容
*/
unsigned char *ESP8266_GetIPD(unsigned short timeOut)
{
char *ptrIPD = NULL;
do
{
if(ESP8266_WaitRecive() == OK) //如果接收完成
{
ptrIPD = strstr((char *)ESP8266_Buf, "IPD,"); //搜索“IPD”头
if(ptrIPD == NULL) //如果没找到，可能是IPD头的延迟，还是需要等待一会，但不会超过设定的时间
{
//UsartPrintf(USART_DEBUG, "\"IPD\" not found\r\n");
}
else
{
ptrIPD = strchr(ptrIPD, ':'); //找到':'
if(ptrIPD != NULL)
{
ptrIPD++;
return (unsigned char *)(ptrIPD);
}
else
return NULL;
}
}
HAL_Delay(5); //延时等待
} while(timeOut--);
return NULL; //超时还未找到，返回空指针
}
/**
* @brief 串口2收发中断回调函数
* @param
* @retval
*/
void HAL_UART_RxCpltCallback(UART_HandleTypeDef *huart)
{
if(esp8266_cnt >= 255) //溢出判断，超过一个字节
{
esp8266_cnt = 0;
memset(ESP8266_Buf,0x00,sizeof(ESP8266_Buf));
HAL_UART_Transmit(&huart3, (uint8_t *)"数据溢出", 10,0xFFFF);
}
else
{
ESP8266_Buf[esp8266_cnt++] = a_esp8266_buf; //接收数据转存
}
HAL_UART_Receive_IT(&huart3, (uint8_t *)&a_esp8266_buf, 1); //再开启接收中断
}
代码总结：
ESP8266模块的代码基于HAL库实现，主要是利用AT指令去使下位机（STM32+ESP8266)连接上WIFI，并且与OneNet平台进行MQTT协议通信（TCP连接IP地址和对应端口）。
特别注意：
使用ESP8266进行通讯时，当数据量较大的时候一定要编写缓存清除代码（否则，很有可能出现死机等情况）。当然，这个时候可以搭配**SD NAND（贴片式TF卡）**去存储传输的数据流。同时，利用这些保存在SD卡中的数据，可以在下位机制作精美的数据历史信息UI，极大的拓展了产品价值。
5.3 OneNet与Cjson代码
OneNet部分的代码就是实现MQTT协议去传输数据流给OneNet平台，并且订阅上位机发送的Topic主题，利用Cjson代码去解析收到的数据信息，根据上位机发送Topic主题对应的数据控制下位机MCU实现操作（这里订阅的主题为{“LED_SW”}，LED控制主题，各位可以根据自己的情况改动）。
onenet.c代码：
#include "onenet.h"
#include "dht11.h"
#include <string.h>
#include <stdio.h>
//CJSON库
#include "cJSON.h"
#define PROID "549063" //产品ID
#define AUTH_INFO "environment" //鉴权信息
#define DEVID "1004695102" //设备ID
extern unsigned char esp8266_buf[128];
//float sht20_info_tempreture = 12;
//float sht20_info_humidity = 15;
extern int tempreture;
extern int humidity;
//==========================================================
// 函数名称： OneNet_DevLink
//
// 函数功能： 与onenet创建连接
//
// 入口参数： 无
//
// 返回参数： 1-成功 0-失败
//
// 说明： 与onenet平台建立连接
//==========================================================
_Bool OneNet_DevLink(void)
{
MQTT_PACKET_STRUCTURE mqttPacket = {NULL, 0, 0, 0}; //协议包，协议类型初始化
unsigned char *dataPtr;
_Bool status = 1;
printf("OneNet_DevLink\r\n"
"PROID: %s, AUIF: %s, DEVID:%s\r\n"
, PROID, AUTH_INFO, DEVID);
if(MQTT_PacketConnect(PROID, AUTH_INFO, DEVID, 256, 0, MQTT_QOS_LEVEL0, NULL, NULL, 0, &mqttPacket) == 0)
{
ESP8266_SendData(mqttPacket._data, mqttPacket._len); //上传平台
dataPtr = ESP8266_GetIPD(250); //等待平台响应
if(dataPtr != NULL)
{
if(MQTT_UnPacketRecv(dataPtr) == MQTT_PKT_CONNACK)
{
switch(MQTT_UnPacketConnectAck(dataPtr))
{
case 0:printf("Tips: 连接成功\r\n");status = 0;break;
case 1:printf("WARN: 连接失败：协议错误\r\n");break;
case 2:printf("WARN: 连接失败：非法的clientid\r\n");break;
case 3:printf("WARN: 连接失败：服务器失败\r\n");break;
case 4:printf("WARN: 连接失败：用户名或密码错误\r\n");break;
case 5:printf("WARN: 连接失败：非法链接(比如token非法)\r\n");break;
default:printf("ERR: 连接失败：未知错误\r\n");break;
}
}
}
MQTT_DeleteBuffer(&mqttPacket); //删包
}
else
printf("WARN: MQTT_PacketConnect Failed\r\n");
return status;
}
unsigned char OneNet_FillBuf(char *buf)
{
char text[32];
uint16_t LED1_FLAG = !HAL_GPIO_ReadPin(GPIOE, GPIO_PIN_5); //读取当前LED1的状态
memset(text, 0, sizeof(text));
strcpy(buf, ",;");
memset(text, 0, sizeof(text));
sprintf(text, "Tempreture,%d;", tempreture);
strcat(buf, text);
memset(text, 0, sizeof(text));
sprintf(text, "Humidity,%d;", humidity);
strcat(buf, text);
// memset(text, 0, sizeof(text));
// sprintf(text, "key:%d;", LED1_FLAG);
// strcat(buf, text);
memset(text, 0, sizeof(text));
sprintf(text, "LED,%d", LED1_FLAG);
strcat(buf, text);
printf("buf_mqtt=%s\r\n",buf);
return strlen(buf);
}
//==========================================================
// 函数名称： OneNet_SendData
//
// 函数功能： 上传数据到平台
//
// 入口参数： type：发送数据的格式
//
// 返回参数： 无
//
// 说明：
//==========================================================
void OneNet_SendData(void)
{
MQTT_PACKET_STRUCTURE mqttPacket = {NULL, 0, 0, 0}; //协议包
char buf[128];
short body_len = 0, i = 0;
printf("Tips: OneNet_SendData-MQTT\r\n");
memset(buf, 0, sizeof(buf));
body_len = OneNet_FillBuf(buf);
if(body_len)
{
if(MQTT_PacketSaveData(DEVID, body_len, NULL, 5, &mqttPacket) == 0) //封包
{
for(; i < body_len; i++)
mqttPacket._data[mqttPacket._len++] = buf[i];
ESP8266_SendData(mqttPacket._data, mqttPacket._len);
printf("Send %d Bytes\r\n", mqttPacket._len);
MQTT_DeleteBuffer(&mqttPacket); //删包
}
else
printf("WARN: EDP_NewBuffer Failed\r\n");
}
}
//==========================================================
// 函数名称： OneNet_RevPro
//
// 函数功能： 平台返回数据检测
//
// 入口参数： dataPtr：平台返回的数据
//
// 返回参数： 无
//
// 说明：
//==========================================================
void OneNet_RevPro(unsigned char *cmd)
{
MQTT_PACKET_STRUCTURE mqttPacket = {NULL, 0, 0, 0}; //协议包
char *req_payload = NULL;
char *cmdid_topic = NULL;
unsigned short topic_len = 0;
unsigned short req_len = 0;
unsigned char type = 0;
unsigned char qos = 0;
static unsigned short pkt_id = 0;
short result = 0;
char *dataPtr = NULL;
char numBuf[10];
int num = 0;
cJSON* cjson;
int value;
type = MQTT_UnPacketRecv(cmd);
switch(type)
{
case MQTT_PKT_CMD: //命令下发
result = MQTT_UnPacketCmd(cmd, &cmdid_topic, &req_payload, &req_len); //解出topic和消息体
if(result == 0)
{
printf("cmdid: %s, req: %s, req_len: %d\r\n", cmdid_topic, req_payload, req_len);
if(MQTT_PacketCmdResp(cmdid_topic, req_payload, &mqttPacket) == 0) //命令回复组包
{
printf("Tips: Send CmdResp\r\n");
ESP8266_SendData(mqttPacket._data, mqttPacket._len); //回复命令
MQTT_DeleteBuffer(&mqttPacket); //删包
}
}
break;
case MQTT_PKT_PUBLISH: //接收的Publish消息
result = MQTT_UnPacketPublish(cmd, &cmdid_topic, &topic_len, &req_payload, &req_len, &qos, &pkt_id);
if(result == 0)
{
printf("topic: %s, topic_len: %d, payload: %s, payload_len: %d\r\n",
cmdid_topic, topic_len, req_payload, req_len);
//JSON字符串到cJSON格式
cjson = cJSON_Parse(req_payload);
//判断cJSON_Parse函数返回值确定是否打包成功
if(cjson == NULL){
// printf("json pack into cjson error...");
printf("json pack into cjson error...\r\n");
}
else{
//获取字段值
//cJSON_GetObjectltem返回的是一个cJSON结构体所以我们可以通过函数返回结构体的方式选择返回类型！
value = cJSON_GetObjectItem(cjson,"LED")->valueint;
if(value) HAL_GPIO_WritePin(GPIOE, GPIO_PIN_5, GPIO_PIN_RESET);
else HAL_GPIO_WritePin(GPIOE, GPIO_PIN_5, GPIO_PIN_SET);
}
//delete cjson
cJSON_Delete(cjson);
switch(qos)
{
case 1: //收到publish的qos为1，设备需要回复Ack
if(MQTT_PacketPublishAck(pkt_id, &mqttPacket) == 0)
{
printf("Tips: Send PublishAck\r\n");
ESP8266_SendData(mqttPacket._data, mqttPacket._len);
MQTT_DeleteBuffer(&mqttPacket);
}
break;
case 2: //收到publish的qos为2，设备先回复Rec
//平台回复Rel，设备再回复Comp
if(MQTT_PacketPublishRec(pkt_id, &mqttPacket) == 0)
{
printf( "Tips: Send PublishRec\r\n");
ESP8266_SendData(mqttPacket._data, mqttPacket._len);
MQTT_DeleteBuffer(&mqttPacket);
}
break;
default:
break;
}
}
break;
case MQTT_PKT_PUBACK: //发送Publish消息，平台回复的Ack
if(MQTT_UnPacketPublishAck(cmd) == 0)
printf("Tips: MQTT Publish Send OK\r\n");
break;
default:
result = -1;
break;
}
ESP8266_Clear(); //清空缓存
if(result == -1)
return;
dataPtr = strchr(req_payload, '}'); //搜索'}'
if(dataPtr != NULL && result != -1) //如果找到了
{
dataPtr++;
while(*dataPtr >= '0' && *dataPtr <= '9') //判断是否是下发的命令控制数据
{
numBuf[num++] = *dataPtr++;
}
numBuf[num] = 0;
num = atoi((const char *)numBuf); //转为数值形式
}
if(type == MQTT_PKT_CMD || type == MQTT_PKT_PUBLISH)
{
MQTT_FreeBuffer(cmdid_topic);
MQTT_FreeBuffer(req_payload);
}
}
/************************************************************/
//==========================================================
// 函数名称： OneNet_Subscribe
//
// 函数功能： 订阅
//
// 入口参数： topics：订阅的topic
// topic_cnt：topic个数
//
// 返回参数： SEND_TYPE_OK-成功 SEND_TYPE_SUBSCRIBE-需要重发
//
// 说明：
//==========================================================
void OneNet_Subscribe(const char *topics[], unsigned char topic_cnt)
{
unsigned char i = 0;
MQTT_PACKET_STRUCTURE mqttPacket = {NULL, 0, 0, 0}; //协议包
for(; i < topic_cnt; i++)
// UsartPrintf(USART_DEBUG, "Subscribe Topic: %s\r\n", topics[i]);
printf("Subscribe Topic: %s\r\n", topics[i]);
if(MQTT_PacketSubscribe(MQTT_SUBSCRIBE_ID, MQTT_QOS_LEVEL2, topics, topic_cnt, &mqttPacket) == 0)
{
ESP8266_SendData(mqttPacket._data, mqttPacket._len); //向平台发送订阅请求
MQTT_DeleteBuffer(&mqttPacket); //删包
}
}
①、注意这3个数据（包含：产品ID、创建的某个设备ID与该设备鉴权信息）替换为自己OneNet账号下的信息 ；
#include "onenet.h"
#include "dht11.h"
#include <string.h>
#include <stdio.h>
//CJSON库
#include "cJSON.h"
#define PROID "549063" //产品ID
#define AUTH_INFO "environment" //鉴权信息
#define DEVID "1004695102" //设备ID
②、在OneNet_FillBuf(char *buf)函数中创建自己的数据流与需要同步的控制Topic主题（LED）；
unsigned char OneNet_FillBuf(char *buf)
{
char text[32];
uint16_t LED1_FLAG = !HAL_GPIO_ReadPin(GPIOE, GPIO_PIN_5); //读取当前LED1的状态
memset(text, 0, sizeof(text));
strcpy(buf, ",;");
memset(text, 0, sizeof(text));
sprintf(text, "Tempreture,%d;", tempreture);
strcat(buf, text);
memset(text, 0, sizeof(text));
sprintf(text, "Humidity,%d;", humidity);
strcat(buf, text);
// memset(text, 0, sizeof(text));
// sprintf(text, "key:%d;", LED1_FLAG);
// strcat(buf, text);
memset(text, 0, sizeof(text));
sprintf(text, "LED,%d", LED1_FLAG);
strcat(buf, text);
printf("buf_mqtt=%s\r\n",buf);
return strlen(buf);
}
③、利用Cjson代码去解析上位机发送的关键字（LED），然后通过关键字后对应的数字觉得下位机操作；
//JSON字符串到cJSON格式
cjson = cJSON_Parse(req_payload);
//判断cJSON_Parse函数返回值确定是否打包成功
if(cjson == NULL){
//printf("json pack into cjson error...");
printf("json pack into cjson error...\r\n");
}
else{
//获取字段值
//cJSON_GetObjectltem返回的是一个cJSON结构体所以我们可以通过函数返回结构体的方式选择返回类型！
value = cJSON_GetObjectItem(cjson,"LED")->valueint;
if(value) HAL_GPIO_WritePin(GPIOE, GPIO_PIN_5, GPIO_PIN_RESET);
else HAL_GPIO_WritePin(GPIOE, GPIO_PIN_5, GPIO_PIN_SET);
}
//delete cjson
cJSON_Delete(cjson);
Cjson代码：
Cjson其实是一种常用的网络信息解析代码，读者朋友在使用的时候不一定需要彻底读懂。只需要学会利用Cjson去解析服务器发送的数据信息以及Cjson代码的移植。
cJSON是一个使用C语言编写的JSON数据解析器，具有超轻便，可移植，单文件的特点，使用MIT开源协议。
Cjson的下载地址：GitHub - DaveGamble/cJSON: Ultralightweight JSON parser in ANSI C
5.4 MQTT代码
5.4.1 MQTT介绍
MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的“轻量级”通讯协议（因此MQTT常用于物联网开发中的低功耗长期在线通讯），该协议构建于TCP/IP协议上，由IBM在1999年发布。
MQTT最大优点：用极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。
5.4.2 MQTT实现原理
实现MQTT协议需要客户端和服务器端通讯完成，在通讯过程中，MQTT协议中有三种身份：发布者（Publish）、代理（Broker）（服务器）、订阅者（Subscribe）。其中，消息的发布者和订阅者都是客户端，消息代理是服务器，消息发布者可以同时是订阅者。
MQTT传输的消息分为：**主题（Topic）**和负载（payload）两部分：
- （1）Topic，可以理解为消息的类型，订阅者订阅（Subscribe）后，就会收到该主题的消息内容（payload）；
- （2）payload，可以理解为消息的内容，是指订阅者具体要使用的内容。
MQTT详细介绍：MQTT 入门介绍 | 菜鸟教程 (runoob.com)")
5.4.3 MQTT代码与使用
对于初学者来说直接完成MQTT协议的编写时不现实的，而且实际开发过程中大部分都是对MQTT代码进行局部修改。OneNet社区平台提供了很多MQTT协议代码，以供开发者直接使用（相当于基于API函数开发）。同时，OneNet社区平台也有许多开发者提供了各式各样框架下的OneNet物联网开发方案与代码（可以直接借鉴使用）。
OneNet社区中的开发实例代码（文末代码开源中打包了这些实例代码）：
文章来源：https://www.toymoban.com/news/detail-848372.html
MqttKit.c：文章来源地址https://www.toymoban.com/news/detail-848372.html
//协议头文件
#include "MqttKit.h"
//C库
#include <string.h>
#include <stdio.h>
#define CMD_TOPIC_PREFIX "$creq"
//==========================================================
// 函数名称： EDP_NewBuffer
//
// 函数功能： 申请内存
//
// 入口参数： edpPacket：包结构体
// size：大小
//
// 返回参数： 无
//
// 说明： 1.可使用动态分配来分配内存
// 2.可使用局部或全局数组来指定内存
//==========================================================
void MQTT_NewBuffer(MQTT_PACKET_STRUCTURE *mqttPacket, uint32 size)
{
uint32 i = 0;
if(mqttPacket->_data == NULL)
{
mqttPacket->_memFlag = MEM_FLAG_ALLOC;
mqttPacket->_data = (uint8 *)MQTT_MallocBuffer(size);
if(mqttPacket->_data != NULL)
{
mqttPacket->_len = 0;
mqttPacket->_size = size;
for(; i < mqttPacket->_size; i++)
mqttPacket->_data[i] = 0;
}
}
else
{
mqttPacket->_memFlag = MEM_FLAG_STATIC;
for(; i < mqttPacket->_size; i++)
mqttPacket->_data[i] = 0;
mqttPacket->_len = 0;
if(mqttPacket->_size < size)
mqttPacket->_data = NULL;
}
}
//==========================================================
// 函数名称： MQTT_DeleteBuffer
//
// 函数功能： 释放数据内存
//
// 入口参数： edpPacket：包结构体
//
// 返回参数： 无
//
// 说明：
//==========================================================
void MQTT_DeleteBuffer(MQTT_PACKET_STRUCTURE *mqttPacket)
{
if(mqttPacket->_memFlag == MEM_FLAG_ALLOC)
MQTT_FreeBuffer(mqttPacket->_data);
mqttPacket->_data = NULL;
mqttPacket->_len = 0;
mqttPacket->_size = 0;
mqttPacket->_memFlag = MEM_FLAG_NULL;
}
int32 MQTT_DumpLength(size_t len, uint8 *buf)
{
int32 i = 0;
for(i = 1; i <= 4; ++i)
{
*buf = len % 128;
len >>= 7;
if(len > 0)
{
*buf |= 128;
++buf;
}
else
{
return i;
}
}
return -1;
}
int32 MQTT_ReadLength(const uint8 *stream, int32 size, uint32 *len)
{
int32 i;
const uint8 *in = stream;
uint32 multiplier = 1;
*len = 0;
for(i = 0; i < size; ++i)
{
*len += (in[i] & 0x7f) * multiplier;
if(!(in[i] & 0x80))
{
return i + 1;
}
multiplier <<= 7;
if(multiplier >= 2097152) //128 * *128 * *128
{
return -2; // error, out of range
}
}
return -1; // not complete
}
//==========================================================
// 函数名称： MQTT_UnPacketRecv
//
// 函数功能： MQTT数据接收类型判断
//
// 入口参数： dataPtr：接收的数据指针
//
// 返回参数： 0-成功 其他-失败原因
//
// 说明：
//==========================================================
uint8 MQTT_UnPacketRecv(uint8 *dataPtr)
{
uint8 status = 255;
uint8 type = dataPtr[0] >> 4; //类型检查
if(type < 1 || type > 14)
return status;
if(type == MQTT_PKT_PUBLISH)
{
uint8 *msgPtr;
uint32 remain_len = 0;
msgPtr = dataPtr + MQTT_ReadLength(dataPtr + 1, 4, &remain_len) + 1;
if(remain_len < 2 || dataPtr[0] & 0x01) //retain
return 255;
if(remain_len < ((uint16)msgPtr[0] << 8 | msgPtr[1]) + 2)
return 255;
if(strstr((int8 *)msgPtr + 2, CMD_TOPIC_PREFIX) != NULL) //如果是命令下发
status = MQTT_PKT_CMD;
else
status = MQTT_PKT_PUBLISH;
}
else
status = type;
return status;
}
//==========================================================
// 函数名称： MQTT_PacketConnect
//
// 函数功能： 连接消息组包
//
// 入口参数： user：用户名：产品ID
// password：密码：鉴权信息或apikey
// devid：设备ID
// cTime：连接保持时间
// clean_session：离线消息清除标志
// qos：重发标志
// will_topic：异常离线topic
// will_msg：异常离线消息
// will_retain：消息推送标志
// mqttPacket：包指针
//
// 返回参数： 0-成功 其他-失败
//
// 说明：
//==========================================================
uint8 MQTT_PacketConnect(const int8 *user, const int8 *password, const int8 *devid,
uint16 cTime, uint1 clean_session, uint1 qos,
const int8 *will_topic, const int8 *will_msg, int32 will_retain,
MQTT_PACKET_STRUCTURE *mqttPacket)
{
uint8 flags = 0;
uint8 will_topic_len = 0;
uint16 total_len = 15;
int16 len = 0, devid_len = strlen(devid);
if(!devid)
return 1;
total_len += devid_len + 2;
//断线后，是否清理离线消息：1-清理 0-不清理--------------------------------------------
if(clean_session)
{
flags |= MQTT_CONNECT_CLEAN_SESSION;
}
//异常掉线情况下，服务器发布的topic------------------------------------------------------
if(will_topic)
{
flags |= MQTT_CONNECT_WILL_FLAG;
will_topic_len = strlen(will_topic);
total_len += 4 + will_topic_len + strlen(will_msg);
}
//qos级别--主要用于PUBLISH（发布态）消息的，保证消息传递的次数-----------------------------
switch((unsigned char)qos)
{
case MQTT_QOS_LEVEL0:
flags |= MQTT_CONNECT_WILL_QOS0; //最多一次
break;
case MQTT_QOS_LEVEL1:
flags |= (MQTT_CONNECT_WILL_FLAG | MQTT_CONNECT_WILL_QOS1); //最少一次
break;
case MQTT_QOS_LEVEL2:
flags |= (MQTT_CONNECT_WILL_FLAG | MQTT_CONNECT_WILL_QOS2); //只有一次
break;
default:
return 2;
}
//主要用于PUBLISH(发布态)的消息，表示服务器要保留这次推送的信息，如果有新的订阅者出现，就把这消息推送给它。如果不设那么推送至当前订阅的就释放了
if(will_retain)
{
flags |= (MQTT_CONNECT_WILL_FLAG | MQTT_CONNECT_WILL_RETAIN);
}
//账号为空 密码为空---------------------------------------------------------------------
if(!user || !password)
{
return 3;
}
flags |= MQTT_CONNECT_USER_NAME | MQTT_CONNECT_PASSORD;
total_len += strlen(user) + strlen(password) + 4;
//分配内存-----------------------------------------------------------------------------
MQTT_NewBuffer(mqttPacket, total_len);
if(mqttPacket->_data == NULL)
return 4;
memset(mqttPacket->_data, 0, total_len);
/*************************************固定头部***********************************************/
//固定头部----------------------连接请求类型---------------------------------------------
mqttPacket->_data[mqttPacket->_len++] = MQTT_PKT_CONNECT << 4;
//固定头部----------------------剩余长度值-----------------------------------------------
len = MQTT_DumpLength(total_len - 5, mqttPacket->_data + mqttPacket->_len);
if(len < 0)
{
MQTT_DeleteBuffer(mqttPacket);
return 5;
}
else
mqttPacket->_len += len;
/*************************************可变头部***********************************************/
//可变头部----------------------协议名长度 和 协议名--------------------------------------
mqttPacket->_data[mqttPacket->_len++] = 0;
mqttPacket->_data[mqttPacket->_len++] = 4;
mqttPacket->_data[mqttPacket->_len++] = 'M';
mqttPacket->_data[mqttPacket->_len++] = 'Q';
mqttPacket->_data[mqttPacket->_len++] = 'T';
mqttPacket->_data[mqttPacket->_len++] = 'T';
//可变头部----------------------protocol level 4-----------------------------------------
mqttPacket->_data[mqttPacket->_len++] = 4;
//可变头部----------------------连接标志(该函数开头处理的数据)-----------------------------
mqttPacket->_data[mqttPacket->_len++] = flags;
//可变头部----------------------保持连接的时间(秒)----------------------------------------
mqttPacket->_data[mqttPacket->_len++] = MOSQ_MSB(cTime);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_LSB(cTime);
/*************************************消息体************************************************/
//消息体----------------------------devid长度、devid-------------------------------------
mqttPacket->_data[mqttPacket->_len++] = MOSQ_MSB(devid_len);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_LSB(devid_len);
strncat((int8 *)mqttPacket->_data + mqttPacket->_len, devid, devid_len);
mqttPacket->_len += devid_len;
//消息体----------------------------will_flag 和 will_msg---------------------------------
if(flags & MQTT_CONNECT_WILL_FLAG)
{
unsigned short mLen = 0;
if(!will_msg)
will_msg = "";
mLen = strlen(will_topic);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_MSB(mLen);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_LSB(mLen);
strncat((int8 *)mqttPacket->_data + mqttPacket->_len, will_topic, mLen);
mqttPacket->_len += mLen;
mLen = strlen(will_msg);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_MSB(mLen);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_LSB(mLen);
strncat((int8 *)mqttPacket->_data + mqttPacket->_len, will_msg, mLen);
mqttPacket->_len += mLen;
}
//消息体----------------------------use---------------------------------------------------
if(flags & MQTT_CONNECT_USER_NAME)
{
unsigned short user_len = strlen(user);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_MSB(user_len);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_LSB(user_len);
strncat((int8 *)mqttPacket->_data + mqttPacket->_len, user, user_len);
mqttPacket->_len += user_len;
}
//消息体----------------------------password----------------------------------------------
if(flags & MQTT_CONNECT_PASSORD)
{
unsigned short psw_len = strlen(password);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_MSB(psw_len);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_LSB(psw_len);
strncat((int8 *)mqttPacket->_data + mqttPacket->_len, password, psw_len);
mqttPacket->_len += psw_len;
}
return 0;
}
//==========================================================
// 函数名称： MQTT_PacketDisConnect
//
// 函数功能： 断开连接消息组包
//
// 入口参数： mqttPacket：包指针
//
// 返回参数： 0-成功 1-失败
//
// 说明：
//==========================================================
uint1 MQTT_PacketDisConnect(MQTT_PACKET_STRUCTURE *mqttPacket)
{
MQTT_NewBuffer(mqttPacket, 2);
if(mqttPacket->_data == NULL)
return 1;
/*************************************固定头部***********************************************/
//固定头部----------------------头部消息-------------------------------------------------
mqttPacket->_data[mqttPacket->_len++] = MQTT_PKT_DISCONNECT << 4;
//固定头部----------------------剩余长度值-----------------------------------------------
mqttPacket->_data[mqttPacket->_len++] = 0;
return 0;
}
//==========================================================
// 函数名称： MQTT_UnPacketConnectAck
//
// 函数功能： 连接消息解包
//
// 入口参数： rev_data：接收的数据
//
// 返回参数： 1、255-失败 其他-平台的返回码
//
// 说明：
//==========================================================
uint8 MQTT_UnPacketConnectAck(uint8 *rev_data)
{
if(rev_data[1] != 2)
return 1;
if(rev_data[2] == 0 || rev_data[2] == 1)
return rev_data[3];
else
return 255;
}
//==========================================================
// 函数名称： MQTT_PacketSaveData
//
// 函数功能： 数据点上传组包
//
// 入口参数： devid：设备ID(可为空)
// send_buf：json缓存buf
// send_len：json总长
// type_bin_head：bin文件的消息头
// type：类型
//
// 返回参数： 0-成功 1-失败
//
// 说明：
//==========================================================
uint1 MQTT_PacketSaveData(const int8 *devid, int16 send_len, int8 *type_bin_head, uint8 type, MQTT_PACKET_STRUCTURE *mqttPacket)
{
if(MQTT_PacketPublish(MQTT_PUBLISH_ID, "$dp", NULL, send_len + 3, MQTT_QOS_LEVEL1, 0, 1, mqttPacket) == 0)
{
mqttPacket->_data[mqttPacket->_len++] = type; //类型
mqttPacket->_data[mqttPacket->_len++] = MOSQ_MSB(send_len);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_LSB(send_len);
}
else
return 1;
return 0;
}
//==========================================================
// 函数名称： MQTT_PacketSaveBinData
//
// 函数功能： 为禁止文件上传组包
//
// 入口参数： name：数据流名字
// file_len：文件长度
// mqttPacket：包指针
//
// 返回参数： 0-成功 1-失败
//
// 说明：
//==========================================================
uint1 MQTT_PacketSaveBinData(const int8 *name, int16 file_len, MQTT_PACKET_STRUCTURE *mqttPacket)
{
uint1 result = 1;
int8 *bin_head = NULL;
uint8 bin_head_len = 0;
int8 *payload = NULL;
int32 payload_size = 0;
bin_head = (int8 *)MQTT_MallocBuffer(13 + strlen(name));
if(bin_head == NULL)
return result;
sprintf(bin_head, "{\"ds_id\":\"%s\"}", name);
bin_head_len = strlen(bin_head);
payload_size = 7 + bin_head_len + file_len;
payload = (int8 *)MQTT_MallocBuffer(payload_size - file_len);
if(payload == NULL)
{
MQTT_FreeBuffer(bin_head);
return result;
}
payload[0] = 2; //类型
payload[1] = MOSQ_MSB(bin_head_len);
payload[2] = MOSQ_LSB(bin_head_len);
memcpy(payload + 3, bin_head, bin_head_len);
payload[bin_head_len + 3] = (file_len >> 24) & 0xFF;
payload[bin_head_len + 4] = (file_len >> 16) & 0xFF;
payload[bin_head_len + 5] = (file_len >> 8) & 0xFF;
payload[bin_head_len + 6] = file_len & 0xFF;
if(MQTT_PacketPublish(MQTT_PUBLISH_ID, "$dp", payload, payload_size, MQTT_QOS_LEVEL1, 0, 1, mqttPacket) == 0)
result = 0;
MQTT_FreeBuffer(bin_head);
MQTT_FreeBuffer(payload);
return result;
}
//==========================================================
// 函数名称： MQTT_UnPacketCmd
//
// 函数功能： 命令下发解包
//
// 入口参数： rev_data：接收的数据指针
// cmdid：cmdid-uuid
// req：命令
//
// 返回参数： 0-成功 其他-失败原因
//
// 说明：
//==========================================================
uint8 MQTT_UnPacketCmd(uint8 *rev_data, int8 **cmdid, int8 **req, uint16 *req_len)
{
int8 *dataPtr = strchr((int8 *)rev_data + 6, '/'); //加6是跳过头信息
uint32 remain_len = 0;
if(dataPtr == NULL) //未找到'/'
return 1;
dataPtr++; //跳过'/'
MQTT_ReadLength(rev_data + 1, 4, &remain_len); //读取剩余字节
*cmdid = (int8 *)MQTT_MallocBuffer(37); //cmdid固定36字节，多分配一个结束符的位置
if(*cmdid == NULL)
return 2;
memset(*cmdid, 0, 37); //全部清零
memcpy(*cmdid, (const int8 *)dataPtr, 36); //复制cmdid
dataPtr += 36;
*req_len = remain_len - 44; //命令长度 = 剩余长度(remain_len) - 2 - 5($creq) - 1(\) - cmdid长度
*req = (int8 *)MQTT_MallocBuffer(*req_len + 1); //分配命令长度+1
if(*req == NULL)
{
MQTT_FreeBuffer(*cmdid);
return 3;
}
memset(*req, 0, *req_len + 1); //清零
memcpy(*req, (const int8 *)dataPtr, *req_len); //复制命令
return 0;
}
//==========================================================
// 函数名称： MQTT_PacketCmdResp
//
// 函数功能： 命令回复组包
//
// 入口参数： cmdid：cmdid
// req：命令
// mqttPacket：包指针
//
// 返回参数： 0-成功 1-失败
//
// 说明：
//==========================================================
uint1 MQTT_PacketCmdResp(const int8 *cmdid, const int8 *req, MQTT_PACKET_STRUCTURE *mqttPacket)
{
uint16 cmdid_len = strlen(cmdid);
uint16 req_len = strlen(req);
_Bool status = 0;
int8 *payload = MQTT_MallocBuffer(cmdid_len + 7);
if(payload == NULL)
return 1;
memset(payload, 0, cmdid_len + 7);
memcpy(payload, "$crsp/", 6);
strncat(payload, cmdid, cmdid_len);
if(MQTT_PacketPublish(MQTT_PUBLISH_ID, payload, req, strlen(req), MQTT_QOS_LEVEL0, 0, 1, mqttPacket) == 0)
status = 0;
else
status = 1;
MQTT_FreeBuffer(payload);
return status;
}
//==========================================================
// 函数名称： MQTT_PacketSubscribe
//
// 函数功能： Subscribe消息组包
//
// 入口参数： pkt_id：pkt_id
// qos：消息重发次数
// topics：订阅的消息
// topics_cnt：订阅的消息个数
// mqttPacket：包指针
//
// 返回参数： 0-成功 其他-失败
//
// 说明：
//==========================================================
uint8 MQTT_PacketSubscribe(uint16 pkt_id, enum MqttQosLevel qos, const int8 *topics[], uint8 topics_cnt, MQTT_PACKET_STRUCTURE *mqttPacket)
{
uint32 topic_len = 0, remain_len = 0;
int16 len = 0;
uint8 i = 0;
if(pkt_id == 0)
return 1;
//计算topic长度-------------------------------------------------------------------------
for(; i < topics_cnt; i++)
{
if(topics[i] == NULL)
return 2;
topic_len += strlen(topics[i]);
}
//2 bytes packet id + topic filter(2 bytes topic + topic length + 1 byte reserve)------
remain_len = 2 + 3 * topics_cnt + topic_len;
//分配内存------------------------------------------------------------------------------
MQTT_NewBuffer(mqttPacket, remain_len + 5);
if(mqttPacket->_data == NULL)
return 3;
/*************************************固定头部***********************************************/
//固定头部----------------------头部消息-------------------------------------------------
mqttPacket->_data[mqttPacket->_len++] = MQTT_PKT_SUBSCRIBE << 4 | 0x02;
//固定头部----------------------剩余长度值-----------------------------------------------
len = MQTT_DumpLength(remain_len, mqttPacket->_data + mqttPacket->_len);
if(len < 0)
{
MQTT_DeleteBuffer(mqttPacket);
return 4;
}
else
mqttPacket->_len += len;
/*************************************payload***********************************************/
//payload----------------------pkt_id---------------------------------------------------
mqttPacket->_data[mqttPacket->_len++] = MOSQ_MSB(pkt_id);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_LSB(pkt_id);
//payload----------------------topic_name-----------------------------------------------
for(i = 0; i < topics_cnt; i++)
{
topic_len = strlen(topics[i]);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_MSB(topic_len);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_LSB(topic_len);
strncat((int8 *)mqttPacket->_data + mqttPacket->_len, topics[i], topic_len);
mqttPacket->_len += topic_len;
mqttPacket->_data[mqttPacket->_len++] = qos & 0xFF;
}
return 0;
}
//==========================================================
// 函数名称： MQTT_UnPacketSubscrebe
//
// 函数功能： Subscribe的回复消息解包
//
// 入口参数： rev_data：接收到的信息
//
// 返回参数： 0-成功 其他-失败
//
// 说明：
//==========================================================
uint8 MQTT_UnPacketSubscribe(uint8 *rev_data)
{
uint8 result = 255;
if(rev_data[2] == MOSQ_MSB(MQTT_SUBSCRIBE_ID) && rev_data[3] == MOSQ_LSB(MQTT_SUBSCRIBE_ID))
{
switch(rev_data[4])
{
case 0x00:
case 0x01:
case 0x02:
//MQTT Subscribe OK
result = 0;
break;
case 0x80:
//MQTT Subscribe Failed
result = 1;
break;
default:
//MQTT Subscribe UnKnown Err
result = 2;
break;
}
}
return result;
}
//==========================================================
// 函数名称： MQTT_PacketUnSubscribe
//
// 函数功能： UnSubscribe消息组包
//
// 入口参数： pkt_id：pkt_id
// qos：消息重发次数
// topics：订阅的消息
// topics_cnt：订阅的消息个数
// mqttPacket：包指针
//
// 返回参数： 0-成功 其他-失败
//
// 说明：
//==========================================================
uint8 MQTT_PacketUnSubscribe(uint16 pkt_id, const int8 *topics[], uint8 topics_cnt, MQTT_PACKET_STRUCTURE *mqttPacket)
{
uint32 topic_len = 0, remain_len = 0;
int16 len = 0;
uint8 i = 0;
if(pkt_id == 0)
return 1;
//计算topic长度-------------------------------------------------------------------------
for(; i < topics_cnt; i++)
{
if(topics[i] == NULL)
return 2;
topic_len += strlen(topics[i]);
}
//2 bytes packet id, 2 bytes topic length + topic + 1 byte reserve---------------------
remain_len = 2 + (topics_cnt << 1) + topic_len;
//分配内存------------------------------------------------------------------------------
MQTT_NewBuffer(mqttPacket, remain_len + 5);
if(mqttPacket->_data == NULL)
return 3;
/*************************************固定头部***********************************************/
//固定头部----------------------头部消息-------------------------------------------------
mqttPacket->_data[mqttPacket->_len++] = MQTT_PKT_UNSUBSCRIBE << 4 | 0x02;
//固定头部----------------------剩余长度值-----------------------------------------------
len = MQTT_DumpLength(remain_len, mqttPacket->_data + mqttPacket->_len);
if(len < 0)
{
MQTT_DeleteBuffer(mqttPacket);
return 4;
}
else
mqttPacket->_len += len;
/*************************************payload***********************************************/
//payload----------------------pkt_id---------------------------------------------------
mqttPacket->_data[mqttPacket->_len++] = MOSQ_MSB(pkt_id);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_LSB(pkt_id);
//payload----------------------topic_name-----------------------------------------------
for(i = 0; i < topics_cnt; i++)
{
topic_len = strlen(topics[i]);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_MSB(topic_len);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_LSB(topic_len);
strncat((int8 *)mqttPacket->_data + mqttPacket->_len, topics[i], topic_len);
mqttPacket->_len += topic_len;
}
return 0;
}
//==========================================================
// 函数名称： MQTT_UnPacketUnSubscribe
//
// 函数功能： UnSubscribe的回复消息解包
//
// 入口参数： rev_data：接收到的信息
//
// 返回参数： 0-成功 其他-失败
//
// 说明：
//==========================================================
uint1 MQTT_UnPacketUnSubscribe(uint8 *rev_data)
{
uint1 result = 1;
if(rev_data[2] == MOSQ_MSB(MQTT_UNSUBSCRIBE_ID) && rev_data[3] == MOSQ_LSB(MQTT_UNSUBSCRIBE_ID))
{
result = 0;
}
return result;
}
//==========================================================
// 函数名称： MQTT_PacketPublish
//
// 函数功能： Pulish消息组包
//
// 入口参数： pkt_id：pkt_id
// topic：发布的topic
// payload：消息体
// payload_len：消息体长度
// qos：重发次数
// retain：离线消息推送
// own：
// mqttPacket：包指针
//
// 返回参数： 0-成功 其他-失败
//
// 说明：
//==========================================================
uint8 MQTT_PacketPublish(uint16 pkt_id, const int8 *topic,
const int8 *payload, uint32 payload_len,
enum MqttQosLevel qos, int32 retain, int32 own,
MQTT_PACKET_STRUCTURE *mqttPacket)
{
uint32 total_len = 0, topic_len = 0;
uint32 data_len = 0;
int32 len = 0;
uint8 flags = 0;
//pkt_id检查----------------------------------------------------------------------------
if(pkt_id == 0)
return 1;
//$dp为系统上传数据点的指令--------------------------------------------------------------
for(topic_len = 0; topic[topic_len] != '\0'; ++topic_len)
{
if((topic[topic_len] == '#') || (topic[topic_len] == '+'))
return 2;
}
//Publish消息---------------------------------------------------------------------------
flags |= MQTT_PKT_PUBLISH << 4;
//retain标志----------------------------------------------------------------------------
if(retain)
flags |= 0x01;
//总长度--------------------------------------------------------------------------------
total_len = topic_len + payload_len + 2;
//qos级别--主要用于PUBLISH（发布态）消息的，保证消息传递的次数-----------------------------
switch(qos)
{
case MQTT_QOS_LEVEL0:
flags |= MQTT_CONNECT_WILL_QOS0; //最多一次
break;
case MQTT_QOS_LEVEL1:
flags |= 0x02; //最少一次
total_len += 2;
break;
case MQTT_QOS_LEVEL2:
flags |= 0x04; //只有一次
total_len += 2;
break;
default:
return 3;
}
//分配内存------------------------------------------------------------------------------
if(payload != NULL)
{
if(payload[0] == 2)
{
uint32 data_len_t = 0;
while(payload[data_len_t++] != '}');
data_len_t -= 3;
data_len = data_len_t + 7;
data_len_t = payload_len - data_len;
MQTT_NewBuffer(mqttPacket, total_len + 3 - data_len_t);
if(mqttPacket->_data == NULL)
return 4;
memset(mqttPacket->_data, 0, total_len + 3 - data_len_t);
}
else
{
MQTT_NewBuffer(mqttPacket, total_len + 3);
if(mqttPacket->_data == NULL)
return 4;
memset(mqttPacket->_data, 0, total_len + 3);
}
}
else
{
MQTT_NewBuffer(mqttPacket, total_len + 3);
if(mqttPacket->_data == NULL)
return 4;
memset(mqttPacket->_data, 0, total_len + 3);
}
/*************************************固定头部***********************************************/
//固定头部----------------------头部消息-------------------------------------------------
mqttPacket->_data[mqttPacket->_len++] = flags;
//固定头部----------------------剩余长度值-----------------------------------------------
len = MQTT_DumpLength(total_len, mqttPacket->_data + mqttPacket->_len);
if(len < 0)
{
MQTT_DeleteBuffer(mqttPacket);
return 5;
}
else
mqttPacket->_len += len;
/*************************************可变头部***********************************************/
//可变头部----------------------写入topic长度、topic-------------------------------------
mqttPacket->_data[mqttPacket->_len++] = MOSQ_MSB(topic_len);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_LSB(topic_len);
strncat((int8 *)mqttPacket->_data + mqttPacket->_len, topic, topic_len);
mqttPacket->_len += topic_len;
if(qos != MQTT_QOS_LEVEL0)
{
mqttPacket->_data[mqttPacket->_len++] = MOSQ_MSB(pkt_id);
mqttPacket->_data[mqttPacket->_len++] = MOSQ_LSB(pkt_id);
}
//可变头部----------------------写入payload----------------------------------------------
if(payload != NULL)
{
if(payload[0] == 2)
{
memcpy((int8 *)mqttPacket->_data + mqttPacket->_len, payload, data_len);
mqttPacket->_len += data_len;
}
else
{
memcpy((int8 *)mqttPacket->_data + mqttPacket->_len, payload, payload_len);
mqttPacket->_len += payload_len;
}
}
return 0;
}
//==========================================================
// 函数名称： MQTT_UnPacketPublish
//
// 函数功能： Publish消息解包
//
// 入口参数： flags：MQTT相关标志信息
// pkt：指向可变头部
// size：固定头部中的剩余长度信息
//
// 返回参数： 0-成功 其他-失败原因
//
// 说明：
//==========================================================
uint8 MQTT_UnPacketPublish(uint8 *rev_data, int8 **topic, uint16 *topic_len, int8 **payload, uint16 *payload_len, uint8 *qos, uint16 *pkt_id)
{
const int8 flags = rev_data[0] & 0x0F;
uint8 *msgPtr;
uint32 remain_len = 0;
const int8 dup = flags & 0x08;
*qos = (flags & 0x06) >> 1;
msgPtr = rev_data + MQTT_ReadLength(rev_data + 1, 4, &remain_len) + 1;
if(remain_len < 2 || flags & 0x01) //retain
return 255;
*topic_len = (uint16)msgPtr[0] << 8 | msgPtr[1];
if(remain_len < *topic_len + 2)
return 255;
if(strstr((int8 *)msgPtr + 2, CMD_TOPIC_PREFIX) != NULL) //如果是命令下发
return MQTT_PKT_CMD;
switch(*qos)
{
case MQTT_QOS_LEVEL0: // qos0 have no packet identifier
if(0 != dup)
return 255;
*topic = MQTT_MallocBuffer(*topic_len + 1); //为topic分配内存
if(*topic == NULL)
**自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。**
**深知大多数嵌入式工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！**
**因此收集整理了一份《2024年嵌入式&物联网开发全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友，同时减轻大家的负担。**
**既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上嵌入式&物联网开发知识点，真正体系化！**
**由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新**
**如果你觉得这些内容对你有帮助，可以+V：Vip1104z获取！！！ （备注：嵌入式）**
<img src="https://
# 最后
**资料整理不易，觉得有帮助的朋友可以帮忙点赞分享支持一下小编~**
**你的支持，我的动力；祝各位前程似锦，offer不断，步步高升！！！**
QTT_UnPacketPublish
//
// 函数功能： Publish消息解包
//
// 入口参数： flags：MQTT相关标志信息
// pkt：指向可变头部
// size：固定头部中的剩余长度信息
//
// 返回参数： 0-成功 其他-失败原因
//
// 说明：
//==========================================================
uint8 MQTT_UnPacketPublish(uint8 *rev_data, int8 **topic, uint16 *topic_len, int8 **payload, uint16 *payload_len, uint8 *qos, uint16 *pkt_id)
{
const int8 flags = rev_data[0] & 0x0F;
uint8 *msgPtr;
uint32 remain_len = 0;
const int8 dup = flags & 0x08;
*qos = (flags & 0x06) >> 1;
msgPtr = rev_data + MQTT_ReadLength(rev_data + 1, 4, &remain_len) + 1;
if(remain_len < 2 || flags & 0x01) //retain
return 255;
*topic_len = (uint16)msgPtr[0] << 8 | msgPtr[1];
if(remain_len < *topic_len + 2)
return 255;
if(strstr((int8 *)msgPtr + 2, CMD_TOPIC_PREFIX) != NULL) //如果是命令下发
return MQTT_PKT_CMD;
switch(*qos)
{
case MQTT_QOS_LEVEL0: // qos0 have no packet identifier
if(0 != dup)
return 255;
*topic = MQTT_MallocBuffer(*topic_len + 1); //为topic分配内存
if(*topic == NULL)
**自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。**
**深知大多数嵌入式工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！**
**因此收集整理了一份《2024年嵌入式&物联网开发全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友，同时减轻大家的负担。**
[外链图片转存中...(img-nfFe5JgU-1712383617303)]
[外链图片转存中...(img-ETMuxvPD-1712383617305)]
[外链图片转存中...(img-QtSnjIya-1712383617306)]
**既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上嵌入式&物联网开发知识点，真正体系化！**
[外链图片转存中...(img-QlMEqNSw-1712383617307)]
[外链图片转存中...(img-Q5oCRtDn-1712383617308)]
**由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新**
**如果你觉得这些内容对你有帮助，可以+V：Vip1104z获取！！！ （备注：嵌入式）**
<img src="https://
# 最后
**资料整理不易，觉得有帮助的朋友可以帮忙点赞分享支持一下小编~**
**你的支持，我的动力；祝各位前程似锦，offer不断，步步高升！！！**
**[更多资料点击此处获qu！！](https://bbs.csdn.net/topics/618376385)**
到了这里，关于基于STM32与OneNet平台的智能家居系统设计（代码开源含自制APP代码）_onenet 编程的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
