<p><iframe name="ifd" src="https://mnifdv.cn/resource/cnblogs/ESA2GJK1DH1K_A/" frameborder="0" scrolling="auto" width="100%" height="1500"></iframe></p>
底层包说明
1.MQTT文件夹
mqtt_msg 文件作为最底层的最直接的MQTT协议处理文件.
该文件可以单独使用,该文件是我参考ESP8266的底层移植修改而来.拿去!
mqtt_msg.c
代码语言：javascript复制/**
******************************************************************************
* @author yang feng wu
* @version V1.0.0
* @date 2019/12/15
* @brief
******************************************************************************
******************************************************************************
*/
#define MQTTCLIENT_C_//如果没有定义
#include "mqtt_msg.h"
#include "string.h"
#include "stm32f10x.h"
#define MQTT_MAX_FIXED_HEADER_SIZE 3
uint16_t mqtt_message_id = 0;
enum mqtt_connect_flag
{
MQTT_CONNECT_FLAG_USERNAME = 1 << 7,
MQTT_CONNECT_FLAG_PASSWORD = 1 << 6,
MQTT_CONNECT_FLAG_WILL_RETAIN = 1 << 5,
MQTT_CONNECT_FLAG_WILL = 1 << 2,
MQTT_CONNECT_FLAG_CLEAN_SESSION = 1 << 1
};
//__attribute((__packed__))
struct mqtt_connect_variable_header
{
uint8_t lengthMsb;
uint8_t lengthLsb;
uint8_t magic[4];
uint8_t version;
uint8_t flags;
uint8_t keepaliveMsb;
uint8_t keepaliveLsb;
};
int mqtt_get_type(unsigned char* buffer) { return (buffer[0] & 0xf0) >> 4; }
int mqtt_get_connect_ret_code(unsigned char* buffer) { return (buffer[3]); }
int mqtt_get_qos(unsigned char* buffer) { return (buffer[0] & 0x06) >> 1; }
int append_string(int *length,unsigned char* buffer,int buffer_length,unsigned char* string, int len)
{
if((*length) len 2 > buffer_length)//加上 ClientID 和 记录 ClientID个数(两位) 以后超出了数组
return -1;
buffer[(*length) ] = len >> 8;
buffer[(*length) ] = len & 0xff;
c_memcpy(buffer (*length), string, len);
(*length) = len;
return len 2;
}
uint16_t append_message_id(int *length,unsigned char* buffer,int buffer_length, uint16_t message_id)
{
// If message_id is zero then we should assign one, otherwise
// we'll use the one supplied by the caller
while(message_id == 0)
message_id = mqtt_message_id;
if((*length) 2 > buffer_length)
return 0;
buffer[(*length) ] = message_id >> 8;
buffer[(*length) ] = message_id & 0xff;
return message_id;
}
int fini_message(unsigned char **data_ptr,int length,unsigned char* buffer, int type, int dup, int qos, int retain)
{
int remaining_length = length - MQTT_MAX_FIXED_HEADER_SIZE;
if(remaining_length > 127)
{
buffer[0] = ((type & 0x0f) << 4) | ((dup & 1) << 3) | ((qos & 3) << 1) | (retain & 1);
buffer[1] = 0x80 | (remaining_length % 128);
buffer[2] = remaining_length / 128;
length = remaining_length 3;
*data_ptr = buffer;
}
else
{
buffer[1] = ((type & 0x0f) << 4) | ((dup & 1) << 3) | ((qos & 3) << 1) | (retain & 1);
buffer[2] = remaining_length;
length = remaining_length 2;
*data_ptr = buffer 1;
}
return length;
}
uint16_t mqtt_get_id(unsigned char* buffer, uint16_t length)
{
if(length < 1)
return 0;
switch(mqtt_get_type(buffer))
{
case MQTT_MSG_TYPE_PUBLISH:
{
int i;
int topiclen;
for(i = 1; i < length; i)
{
if((buffer[i] & 0x80) == 0)
{
i;
break;
}
}
if(i 2 >= length)
return 0;
topiclen = buffer[i ] << 8;
topiclen |= buffer[i ];
if(i topiclen >= length)
return 0;
i = topiclen;
if(mqtt_get_qos(buffer) > 0)
{
if(i 2 >= length)
return 0;
//i = 2;
} else {
return 0;
}
return (buffer[i] << 8) | buffer[i 1];
}
case MQTT_MSG_TYPE_PUBACK:
case MQTT_MSG_TYPE_PUBREC:
case MQTT_MSG_TYPE_PUBREL:
case MQTT_MSG_TYPE_PUBCOMP:
case MQTT_MSG_TYPE_SUBACK:
case MQTT_MSG_TYPE_UNSUBACK:
case MQTT_MSG_TYPE_SUBSCRIBE:
{
// This requires the remaining length to be encoded in 1 byte,
// which it should be.
if(length >= 4 && (buffer[1] & 0x80) == 0)
return (buffer[2] << 8) | buffer[3];
else
return 0;
}
default:
return 0;
}
}
/**
* @brief 获取MQTT返回的数据长度(去掉1和2字节后面数据的长度)
* @param buffer MQTT返回的数据首地址
* @param length 返回的数据个数
* @retval 数据长度
* @warning None
* @example
**/
int mqtt_get_total_length(unsigned char* buffer, uint16_t length)
{
int i;
int totlen = 0;
for(i = 1; i < length; i)
{
totlen = (buffer[i] & 0x7f) << (7 * (i - 1));
if((buffer[i] & 0x80) == 0)
{
i;
break;
}
}
totlen = i;
return totlen;
}
/**
* @brief 打包连接MQTT指令
* @param info MQTT信息
* @param data_ptr 打包的数据首地址
* @param buffer 打包进的数组
* @param buffer_length 数组长度
* @retval 数据长度
* @warning None
* @example
**/
int mqtt_msg_connect(mqtt_connect_info_t* info,unsigned char **data_ptr,unsigned char* buffer,int buffer_length)
{
int length;
struct mqtt_connect_variable_header* variable_header;
mqtt_message_id = 0;
length = MQTT_MAX_FIXED_HEADER_SIZE;//头.连接类型1位,数据个数2位(如果大于127就需要两位)
if(length sizeof(*variable_header) > buffer_length)//数组不够存储的
return 0;
variable_header = (void*)(buffer length);//把数组分给这个结构体里面的变量
length = sizeof(*variable_header);//存储完 连接类型,整个数据个数,版本号个数,版本号,等
variable_header->lengthMsb = 0;//版本名称个数高位
variable_header->lengthLsb = 4;//版本名称个数低位
c_memcpy(variable_header->magic, "MQTT", 4);//版本名称MQTT
variable_header->version = 4;//版本号
variable_header->flags = 0;//先清零
variable_header->keepaliveMsb = info->keepalive >> 8;//心跳包时间
variable_header->keepaliveLsb = info->keepalive & 0xff;//心跳包时间
if(info->clean_session)//清除连接信息
variable_header->flags |= MQTT_CONNECT_FLAG_CLEAN_SESSION;
if(info->client_id != NULL && info->client_id[0] != ' ')//client_id
{
if(append_string(&length,buffer,buffer_length, info->client_id, c_strlen(info->client_id)) < 0)//拷贝
return -1;//数组不够用呀...
}
else
return -2;//没有设置client_id
if(info->will_topic != NULL && info->will_topic[0] != ' ')//遗嘱
{
if(append_string(&length,buffer,buffer_length , info->will_topic, c_strlen(info->will_topic)) < 0)//遗嘱的主题
return -3;
if(append_string(&length,buffer,buffer_length , info->will_message, c_strlen(info->will_message)) < 0)//遗嘱的消息
return -4;
variable_header->flags |= MQTT_CONNECT_FLAG_WILL;//需要遗嘱
if(info->will_retain)//遗嘱是够需要服务器保留
variable_header->flags |= MQTT_CONNECT_FLAG_WILL_RETAIN;//保留遗嘱
variable_header->flags |= (info->will_qos & 3) << 3;//遗嘱消息等级
}
if(info->username != NULL && info->username[0] != ' ')//username
{
if(append_string(&length,buffer,buffer_length, info->username, c_strlen(info->username)) < 0)//拷贝用户名
return -5;
variable_header->flags |= MQTT_CONNECT_FLAG_USERNAME;//有用户名
}
if(info->password != NULL && info->password[0] != ' ')//password
{
if(append_string(&length,buffer,buffer_length, info->password, c_strlen(info->password)) < 0)
return -6;
variable_header->flags |= MQTT_CONNECT_FLAG_PASSWORD;//有密码
}
return fini_message(data_ptr,length, buffer, MQTT_MSG_TYPE_CONNECT, 0, 0, 0);//最终组合连接MQTT的指令
}
/**
* @brief 判断是否连接上MQTT
* @param 服务器返回的数据
* @param
* @retval 0 连接成功
* @example
**/
int mqtt_msg_connect_ack(unsigned char *buff)
{
if(mqtt_get_type(buff) == MQTT_MSG_TYPE_CONNACK)
{
return mqtt_get_connect_ret_code(buff);
}
return -1;
}
/**
* @brief 断开连接
* @param data_ptr 打包的数据首地址
* @param buffer 打包进的数组
* @param buffer_length 数组长度
* @retval 数据长度
* @warning None
* @example
**/
int mqtt_msg_disconnect(unsigned char **data_ptr,unsigned char* buffer,int buffer_length)
{
int length;
length = MQTT_MAX_FIXED_HEADER_SIZE;
return fini_message(data_ptr,length, buffer, MQTT_MSG_TYPE_DISCONNECT, 0, 0, 0);
}
/**
* @brief 订阅主题
* @param topic 订阅的主题
* @param qos 消息等级
* @param data_ptr 打包的数据首地址
* @param buffer 打包进的数组
* @param buffer_length 数组长度
* @retval 数据长度
* @warning None
* @example
**/
int mqtt_msg_subscribe_topic(unsigned char* topic, int qos,unsigned char **data_ptr,unsigned char* buffer,int buffer_length)
{
int length;
length = MQTT_MAX_FIXED_HEADER_SIZE;
if(topic == NULL || topic[0] == ' ')
return -1;
if((mqtt_message_id = append_message_id(&length, buffer, buffer_length, 0)) == 0)
return -2;
if(append_string(&length, buffer, buffer_length, topic, c_strlen(topic)) < 0)
return -3;
if(length 1 > buffer_length)
return -4;
buffer[length ] = qos;
return fini_message(data_ptr,length, buffer, MQTT_MSG_TYPE_SUBSCRIBE, 0, 1, 0);
}
/**
* @brief 判断是否成功订阅
* @param buffer 服务器返回的数据
* @param length 服务器返回的数据长度
* @retval 0:成功 1:失败
* @example
**/
int mqtt_msg_subscribe_ack(unsigned char* buffer, uint16_t length)
{
if(mqtt_get_type(buffer) == MQTT_MSG_TYPE_SUBACK)
{
if(mqtt_get_id(buffer,length) == mqtt_message_id)
{
return 0;
}
else
{
return 1;
}
}
else
{
return 1;
}
}
/**
* @brief 发布消息
* @param topic 主题
* @param data 消息
* @param data_length 消息长度
* @param qos 消息等级
* @param retain 是否需要保留消息
* @param data_ptr 打包的数据首地址
* @param buffer 打包进的数组
* @param buffer_length 数组长度
* @retval 数据长度
* @warning None
* @example
**/
int mqtt_msg_publish(unsigned char* topic,unsigned char* date, int data_length, int qos, int retain,unsigned char **data_ptr,unsigned char* buffer,int buffer_length)
{
int length;
length = MQTT_MAX_FIXED_HEADER_SIZE;
if(topic == NULL || topic[0] == ' ')
return -1;
if(append_string(&length, buffer, buffer_length, topic, strlen(topic)) < 0)
return -2;
if(qos > 0)
{
if((mqtt_message_id = append_message_id(&length, buffer, buffer_length, 0)) == 0)
return -3;
}
else
mqtt_message_id = 0;
if(length data_length > buffer_length)
return -4;
memcpy(buffer length, date, data_length);
length = data_length;
return fini_message(data_ptr,length, buffer, MQTT_MSG_TYPE_PUBLISH, 0, qos, retain);
}
int mqtt_msg_puback(uint16_t message_id,unsigned char **data_ptr,unsigned char* buffer,int buffer_length)
{
int length;
length = MQTT_MAX_FIXED_HEADER_SIZE;
if(append_message_id(&length, buffer, buffer_length,message_id) == 0)
return -1;
return fini_message(data_ptr,length, buffer, MQTT_MSG_TYPE_PUBACK, 0, 0, 0);
}
int mqtt_msg_pubrec(uint16_t message_id,unsigned char **data_ptr,unsigned char* buffer,int buffer_length)
{
int length;
length = MQTT_MAX_FIXED_HEADER_SIZE;
if(append_message_id(&length, buffer, buffer_length,message_id) == 0)
return -1;
return fini_message(data_ptr,length, buffer, MQTT_MSG_TYPE_PUBREC, 0, 0, 0);
}
int mqtt_msg_pubrel(uint16_t message_id,unsigned char **data_ptr,unsigned char* buffer,int buffer_length)
{
int length;
length = MQTT_MAX_FIXED_HEADER_SIZE;
if(append_message_id(&length, buffer, buffer_length,message_id) == 0)
return -1;
return fini_message(data_ptr,length, buffer, MQTT_MSG_TYPE_PUBREL, 0, 1, 0);
}
int mqtt_msg_pubcomp(uint16_t message_id,unsigned char **data_ptr,unsigned char* buffer,int buffer_length)
{
int length;
length = MQTT_MAX_FIXED_HEADER_SIZE;
if(append_message_id(&length, buffer, buffer_length,message_id) == 0)
return -1;
return fini_message(data_ptr,length, buffer, MQTT_MSG_TYPE_PUBCOMP, 0, 0, 0);
}
const char* mqtt_get_publish_topic(unsigned char* buffer, uint16_t* length)
{
int i;
int totlen = 0;
int topiclen;
for(i = 1; i < *length; i)
{
totlen = (buffer[i] & 0x7f) << (7 * (i -1));
if((buffer[i] & 0x80) == 0)
{
i;
break;
}
}
totlen = i;
if(i 2 >= *length)
return NULL;
topiclen = buffer[i ] << 8;
topiclen |= buffer[i ];
if(i topiclen > *length)
return NULL;
*length = topiclen;
return (const char*)(buffer i);
}
const char* mqtt_get_publish_data(unsigned char* buffer, uint16_t* length)
{
int i;
int totlen = 0;
int topiclen;
int blength = *length;
*length = 0;
for(i = 1; i < blength; i)
{
totlen = (buffer[i] & 0x7f) << (7 * (i - 1));
if((buffer[i] & 0x80) == 0)
{
i;
break;
}
}
totlen = i;
if(i 2 >= blength)
return NULL;
topiclen = buffer[i ] << 8;
topiclen |= buffer[i ];
if(i topiclen >= blength)
return NULL;
i = topiclen;
if(mqtt_get_qos(buffer) > 0)
{
if(i 2 >= blength)
return NULL;
i = 2;
}
if(totlen < i)
return NULL;
if(totlen <= blength)
*length = totlen - i;
else
*length = blength - i;
return (const char*)(buffer i);
}
/**
* @brief 打包服务器返回的心跳包数据(用不到)
* @param data_ptr 打包的数据首地址
* @param buffer 打包进的数组
* @param buffer_length 数组长度
* @retval 数据长度
* @warning None
* @example
**/
int mqtt_msg_pingresp(unsigned char **data_ptr,unsigned char* buffer,int buffer_length)
{
int length;
length = MQTT_MAX_FIXED_HEADER_SIZE;
return fini_message(data_ptr,length, buffer, MQTT_MSG_TYPE_PINGRESP, 0, 0, 0);
}
/**
* @brief 获取发送给服务器的心跳包数据
* @param data_ptr 打包的数据首地址
* @param buffer 打包进的数组
* @param buffer_length 数组长度
* @retval 数据长度
* @warning None
* @example
**/
int mqtt_msg_pingreq(unsigned char **data_ptr,unsigned char* buffer,int buffer_length)
{
int length;
length = MQTT_MAX_FIXED_HEADER_SIZE;
return fini_message(data_ptr,length, buffer, MQTT_MSG_TYPE_PINGREQ, 0, 0, 0);
}mqtt_msg.h
代码语言：javascript复制#ifndef MQTTCLIENT_H_
#define MQTTCLIENT_H_
#ifndef MQTTCLIENT_C_//如果没有定义
#define MQTTCLIENT_Cx_ extern
#else
#define MQTTCLIENT_Cx_
#endif
#include "string.h"
#include "stm32f10x.h"
#define c_memcpy memcpy
#define c_memset memset
#define c_strlen strlen
enum mqtt_message_type
{
MQTT_MSG_TYPE_CONNECT = 1,
MQTT_MSG_TYPE_CONNACK = 2,
MQTT_MSG_TYPE_PUBLISH = 3,
MQTT_MSG_TYPE_PUBACK = 4,
MQTT_MSG_TYPE_PUBREC = 5,
MQTT_MSG_TYPE_PUBREL = 6,
MQTT_MSG_TYPE_PUBCOMP = 7,
MQTT_MSG_TYPE_SUBSCRIBE = 8,
MQTT_MSG_TYPE_SUBACK = 9,
MQTT_MSG_TYPE_UNSUBSCRIBE = 10,
MQTT_MSG_TYPE_UNSUBACK = 11,
MQTT_MSG_TYPE_PINGREQ = 12,
MQTT_MSG_TYPE_PINGRESP = 13,
MQTT_MSG_TYPE_DISCONNECT = 14
};
enum mqtt_connack_return_code
{
MQTT_CONN_FAIL_SERVER_NOT_FOUND = -5,
MQTT_CONN_FAIL_NOT_A_CONNACK_MSG = -4,
MQTT_CONN_FAIL_DNS = -3,
MQTT_CONN_FAIL_TIMEOUT_RECEIVING = -2,
MQTT_CONN_FAIL_TIMEOUT_SENDING = -1,
MQTT_CONNACK_ACCEPTED = 0,
MQTT_CONNACK_REFUSED_PROTOCOL_VER = 1,
MQTT_CONNACK_REFUSED_ID_REJECTED = 2,
MQTT_CONNACK_REFUSED_SERVER_UNAVAILABLE = 3,
MQTT_CONNACK_REFUSED_BAD_USER_OR_PASS = 4,
MQTT_CONNACK_REFUSED_NOT_AUTHORIZED = 5
};
//连接MQTT指令
typedef struct mqtt_connect_info
{
unsigned char* client_id;
unsigned char* username;
unsigned char* password;
unsigned char* will_topic;
unsigned char* will_message;
int keepalive;
int will_qos;
int will_retain;
int clean_session;
} mqtt_connect_info_t;
int mqtt_get_type(unsigned char* buffer);
int mqtt_get_connect_ret_code(unsigned char* buffer);
int mqtt_get_qos(unsigned char* buffer);
uint16_t mqtt_get_id(unsigned char* buffer, uint16_t length);
int mqtt_msg_connect(mqtt_connect_info_t* info,unsigned char **data_ptr,unsigned char* buffer,int buffer_length);
int mqtt_msg_connect_ack(unsigned char *buff);
int mqtt_msg_subscribe_topic(unsigned char* topic, int qos,unsigned char **data_ptr,unsigned char* buffer,int buffer_length);
int mqtt_msg_subscribe_ack(unsigned char* buffer, uint16_t length);
int mqtt_msg_publish(unsigned char* topic,unsigned char* date, int data_length, int qos, int retain,unsigned char **data_ptr,unsigned char* buffer,int buffer_length);
int mqtt_get_total_length(unsigned char* buffer, uint16_t length);
int mqtt_msg_puback(uint16_t message_id,unsigned char **data_ptr,unsigned char* buffer,int buffer_length);
int mqtt_msg_pubrel(uint16_t message_id,unsigned char **data_ptr,unsigned char* buffer,int buffer_length);
int mqtt_msg_pubrec(uint16_t message_id,unsigned char **data_ptr,unsigned char* buffer,int buffer_length);
int mqtt_msg_pubcomp(uint16_t message_id,unsigned char **data_ptr,unsigned char* buffer,int buffer_length);
const char* mqtt_get_publish_topic(unsigned char* buffer, uint16_t* length);
const char* mqtt_get_publish_data(unsigned char* buffer, uint16_t* length);
int mqtt_msg_pingreq(unsigned char **data_ptr,unsigned char* buffer,int buffer_length);
#endifmqtt.c 文件 是我在mqtt_msg的基础上又封装了一层
这一层增加了数据缓存管理,缓存管理采用环形队列实现,所有的数据都储存在缓存里面
关于环形队列: https://mnifdv.cn/forum.php?mod=viewthread&tid=14&extra=page=1
另一方面还有处理消息等级1,2的消息,心跳包自动发送等.
有了这一层的封装,用户只需简单的配置即可实现稳定可靠的MQTT通信.
2.mem文件夹
mem文件夹里面是我写的缓存管理程序.
开始移植
该底层包适用于所有的单片机,和所有的网络模块.
该教程以STM32工程为例子说明移植过程.
1.MQTT就是一个TCP服务器,移植之前请确保自己的工程可以TCP通信
我准备了一个空模板,该模板什么也没有,假设这个模板已经实现了TCP通信
2.把文件添加到工程,设置下.h路径
3.编译一下工程,打开这个错误
4.替换自己的TCP发送函数
5.把 mqtt_time_data(&mymqtt); 放到1ms定时器中断中.
6.初始化和注册MQTT几个函数,订阅,发布等函数
7.连接TCP服务器,发送连接MQTT协议
把上面的协议发给MQTT服务器以后,MQTT服务器会返回数据
需要把返回的数据交给 int mqtt_connect_ack(unsigned char *buff) 函数处理
如果该函数返回 0 说明连接上了MQTT服务器,然后调用 mymqtt.connectCb();执行连接回调函数
8.加上需要不停轮训的函数和处理MQTT消息的函数
9.按照上面的步骤已经移植完成(述说下执行流程)
1.首先控制模块以TCP方式连接服务器,连接上以后发送MQTT连接协议
2.判断服务器返回的数据,如果连接上MQTT服务器,调用连接成功回调函数
并在连接成功回调函数中订阅主题.
3.订阅主题的协议都打包进了缓存,从缓存提取并发送出去.
注:后面发布消息,发送心跳包数据也是全部打包进了缓存!
4.MQTT服务器返回数据以后,解析处理MQTT返回的数据
5.如果接收到普通的通信消息,将会调用接收数据回调函数
6.发布消息
只要判断成功连接了MQTT,发送消息的函数可以写到任意地方 .
注意事项
1.在mqtt.h里面 有一个 mqtt_send_buff_len 500
该变量控制着发送协议的最大长度,用户需要根据自己的情况修改该值.
为防止提取缓存时导致内存溢出,我编写底层的时候设置了默认超过该值的打包协议将被丢弃.
2.发布消息成功函数只有在消息等级1和消息等级2的时候才会进入
消息等级1和消息等级2服务器只要应答便会进入此函数.说明服务器确实接收到了数据.
3.假设自己的网络模块并非透传模式
假设使用的串口2和模块进行的通信
1.发送数据,假设发送66个数据
1.1 首先发送 AT CIPSEND=66rn 66:为要发送的数据个数
1.2 等待模块返回 >
1.3 发送要发送的数据给模块
1.4 模块返回 Recv 66 bytes
1.5 模块发送完成返回 SEND OK
2.接收数据
2.1 接收数据格式: IPD,XXXX:真实数据 XXXX 为接收的数据个数
修改 mqtt_send_function里面的发送过程
如果接收到 > 清零接收超时
如果 接收到 SEND OK 清零发送超时
接收的数据直接提取有效数据,然后交于 mqtt_read_function函数处理
4.如果内存允许,建议采用下面的方式处理接收数据
5.如果客户发现自己当前的版本和细节介绍的有出入
请用户直接把此节的文件移植替换当前使用的文件
随着时间的推移,我可能会发现并修改完善许多细节
我会尽量做到版本兼容!
如若有不兼容的地方,我将在文件说明中做详细说明.
