#include "ets_sys.h"
#include "driver/uart.h"
#include "osapi.h"
#include "mqtt.h"
#include "wifi.h"
#include "config.h"
#include "debug.h"
#include "gpio.h"
#include "user_interface.h"
#include "mem.h"
#include "sntp.h"
全局变量
MQTT_Client mqttClient; // MQTT客户端_结构体【此变量非常重要】
typedef struct
{
struct espconn *pCon; // TCP连接结构体指针
uint8_t security; // 安全类型
uint8_t* host; // 服务端域名/地址
uint32_t port; // 网络连接端口号
ip_addr_t ip; // 32位IP地址
mqtt_state_t mqtt_state; // MQTT状态
mqtt_connect_info_t connect_info; // MQTT【CONNECT】报文的连接参数
MqttCallback connectedCb; // MQTT连接成功_回调
MqttCallback disconnectedCb; // MQTT断开连接_回调
MqttCallback publishedCb; // MQTT发布成功_回调
MqttCallback timeoutCb; // MQTT超时_回调
MqttDataCallback dataCb; // MQTT接收数据_回调
ETSTimer mqttTimer; // MQTT定时器
uint32_t keepAliveTick; // MQTT客户端(ESP8266)心跳计数
uint32_t reconnectTick; // 重连等待计时
uint32_t sendTimeout; // 报文发送超时时间
tConnState connState; // ESP8266运行状态
QUEUE msgQueue; // 消息队列
void* user_data; // 用户数据（预留给用户的指针）
} MQTT_Client;
静态变量
static ETSTimer sntp_timer;
typedef struct _ETSTIMER_ {
struct _ETSTIMER_ *timer_next;
uint32_t timer_expire;
uint32_t timer_period;
ETSTimerFunc *timer_func;
void *timer_arg;
} ETSTimer;
void user_init(void)
{
uart_init(BIT_RATE_115200, BIT_RATE_115200); // 串口波特率设为115200
os_delay_us(60000);
PIN_FUNC_SELECT(PERIPHS_IO_MUX_GPIO4_U, FUNC_GPIO4); // GPIO4输出高 #
GPIO_OUTPUT_SET(GPIO_ID_PIN(4),1); // LED初始化 #
CFG_Load(); // 加载/更新系统参数【WIFI参数、MQTT参数】
// 网络连接参数赋值：服务端域名【mqtt_test_jx.mqtt.iot.gz.baidubce.com】、网络连接端口【1883】、安全类型【0：NO_TLS】
MQTT_InitConnection(&mqttClient, sysCfg.mqtt_host, sysCfg.mqtt_port, sysCfg.security);
// MQTT连接参数赋值：客户端标识符【..】、MQTT用户名【..】、MQTT密钥【..】、保持连接时长【120s】、清除会话【1：clean_session】
MQTT_InitClient(&mqttClient, sysCfg.device_id, sysCfg.mqtt_user, sysCfg.mqtt_pass, sysCfg.mqtt_keepalive, 1);
// 设置遗嘱参数(如果云端没有对应的遗嘱主题，则MQTT连接会被拒绝)
// MQTT_InitLWT(&mqttClient, "Will", "ESP8266_offline", 0, 0);
// 设置MQTT相关函数
MQTT_OnConnected(&mqttClient, mqttConnectedCb); // 设置【MQTT成功连接】函数的另一种调用方式
MQTT_OnDisconnected(&mqttClient, mqttDisconnectedCb); // 设置【MQTT成功断开】函数的另一种调用方式
MQTT_OnPublished(&mqttClient, mqttPublishedCb); // 设置【MQTT成功发布】函数的另一种调用方式
MQTT_OnData(&mqttClient, mqttDataCb); // 设置【接收MQTT数据】函数的另一种调用方式
// 连接WIFI：SSID[..]、PASSWORD[..]、WIFI连接成功函数[wifiConnectCb]
WIFI_Connect(sysCfg.sta_ssid, sysCfg.sta_pwd, wifiConnectCb);
INFO("\r\nSystem started ...\r\n");
}
uart_init(BIT_RATE_115200, BIT_RATE_115200); // 串口波特率设为115200
os_delay_us(60000);
CFG_Load();函数来加载或者是调用系统函数，根据持有人标志是否和之前一样来选择更新或者是不更新系统参数。此函数位于config.c文件中。void ICACHE_FLASH_ATTR CFG_Load()
{
INFO("\r\nload ...\r\n");
// 读Flash【0x7C】扇区，存放到【saveFlag】（读出之前的持有人标识）
spi_flash_read((CFG_LOCATION+3)*SPI_FLASH_SEC_SIZE,(uint32 *)&saveFlag, sizeof(SAVE_FLAG));
//根据【参数扇区标志】，读取对应扇区的系统参数【0：系统参数在0x79扇区!0：系统参数在0x7A扇区】
if (saveFlag.flag == 0)
{
spi_flash_read((CFG_LOCATION+0)*SPI_FLASH_SEC_SIZE, (uint32 *)&sysCfg, sizeof(SYSCFG)); // 读出系统参数(1区：0x79)
}
else //saveFlag.flag != 0
{
spi_flash_read((CFG_LOCATION+1)*SPI_FLASH_SEC_SIZE, (uint32 *)&sysCfg, sizeof(SYSCFG)); // 读出系统参数(2区：0x7A)
}
// 只有在【持有人标识和之前不同】的情况下，才会更新系统参数（修改系统参数时，一定要记得修改持有人标识的值）如果觉得每次都修改持有人标识的值比较麻烦，那么可以注释这条语句。那么ESP8266每次上电执行的时候都会将参数更新到flash当中。
if(sysCfg.cfg_holder != CFG_HOLDER) // 持有人标识不同
{
os_memset(&sysCfg, 0x00, sizeof sysCfg); // 参数扇区=0
sysCfg.cfg_holder = CFG_HOLDER; // 更新持有人标识
os_sprintf(sysCfg.device_id, MQTT_CLIENT_ID, system_get_chip_id()); // 【MQTT_CLIENT_ID】MQTT客户端标识符
sysCfg.device_id[sizeof(sysCfg.device_id) - 1] = '\0'; // 最后添'\0'（防止字符串填满数组，指针溢出）
os_strncpy(sysCfg.sta_ssid, STA_SSID, sizeof(sysCfg.sta_ssid)-1); // 【STA_SSID】WIFI名称
os_strncpy(sysCfg.sta_pwd, STA_PASS, sizeof(sysCfg.sta_pwd)-1); // 【STA_PASS】WIFI密码
sysCfg.sta_type = STA_TYPE; // 【STA_TYPE】WIFI类型
os_strncpy(sysCfg.mqtt_host, MQTT_HOST, sizeof(sysCfg.mqtt_host)-1); // 【MQTT_HOST】MQTT服务端域名/IP地址
sysCfg.mqtt_port = MQTT_PORT; // 【MQTT_PORT】网络连接端口号
os_strncpy(sysCfg.mqtt_user, MQTT_USER, sizeof(sysCfg.mqtt_user)-1); // 【MQTT_USER】MQTT用户名
os_strncpy(sysCfg.mqtt_pass, MQTT_PASS, sizeof(sysCfg.mqtt_pass)-1); // 【MQTT_PASS】MQTT密码
sysCfg.security = DEFAULT_SECURITY; /* default non ssl */ // 【DEFAULT_SECURITY】默认安全等级(默认=0，不加密)
sysCfg.mqtt_keepalive = MQTT_KEEPALIVE; // 【MQTT_KEEPALIVE】保持连接时长(宏定义==120)
INFO(" default configuration\r\n");
CFG_Save(); // 将更新后的系统参数烧录到Flash中
}
}
MQTT_InitConnection函数来将网络连接参数赋值比如服务端域名、网络端口号、安全类型。void ICACHE_FLASH_ATTR MQTT_InitConnection(MQTT_Client *mqttClient, uint8_t* host, uint32_t port, uint8_t security)
{
uint32_t temp;
INFO("MQTT_InitConnection\r\n");
os_memset(mqttClient, 0, sizeof(MQTT_Client)); // 【MQTT客户端】结构体 = 0
temp = os_strlen(host); // 服务端域名/IP的字符串长度
mqttClient->host = (uint8_t*)os_zalloc(temp+1); // 申请空间，存放服务端域名/IP地址字符串
os_strcpy(mqttClient->host, host); // 字符串拷贝
mqttClient->host[temp] = 0; // 最后'\0'
mqttClient->port = port; // 网络端口号 = 1883
mqttClient->security = security; // 安全类型 = 0 = NO_TLS
}
MQT在这里插入代码片T_InitClient函数来将MQTT连接的参数赋值等等，客户端标识符赋值、MQTT用户名赋值、MQTT密码赋值、保持连接时长赋值、清除会话标志位赋值。QUEUE_Init，并且对队列进行了初始化MQTT_InitClient(MQTT_Client *mqttClient, uint8_t* client_id, uint8_t* client_user, uint8_t* client_pass, uint32_t keepAliveTime, uint8_t cleanSession)
{
uint32_t temp;
INFO("MQTT_InitClient\r\n");
// MQTT【CONNECT】报文的连接参数 赋值
os_memset(&mqttClient->connect_info, 0, sizeof(mqtt_connect_info_t)); // MQTT【CONNECT】报文的连接参数 = 0
temp = os_strlen(client_id);
mqttClient->connect_info.client_id = (uint8_t*)os_zalloc(temp + 1); // 申请【客户端标识符】的存放内存
os_strcpy(mqttClient->connect_info.client_id, client_id); // 赋值【客户端标识符】
mqttClient->connect_info.client_id[temp] = 0; // 最后'\0'
if (client_user) // 判断是否有【MQTT用户名】
{
temp = os_strlen(client_user);
mqttClient->connect_info.username = (uint8_t*)os_zalloc(temp + 1);
os_strcpy(mqttClient->connect_info.username, client_user); // 赋值【MQTT用户名】
mqttClient->connect_info.username[temp] = 0;
}
if (client_pass) // 判断是否有【MQTT密码】
{
temp = os_strlen(client_pass);
mqttClient->connect_info.password = (uint8_t*)os_zalloc(temp + 1);
os_strcpy(mqttClient->connect_info.password, client_pass); // 赋值【MQTT密码】
mqttClient->connect_info.password[temp] = 0;
}
mqttClient->connect_info.keepalive = keepAliveTime; // 保持连接 = 120s
mqttClient->connect_info.clean_session = cleanSession; // 清除会话 = 1 = clean_session
// 设置mqtt_state部分参数
mqttClient->mqtt_state.in_buffer = (uint8_t *)os_zalloc(MQTT_BUF_SIZE); // 申请in_buffer内存【入站报文缓存区】
mqttClient->mqtt_state.in_buffer_length = MQTT_BUF_SIZE; // 设置in_buffer大小
mqttClient->mqtt_state.out_buffer = (uint8_t *)os_zalloc(MQTT_BUF_SIZE); // 申请out_buffer内存【出站报文缓存区】
mqttClient->mqtt_state.out_buffer_length = MQTT_BUF_SIZE; // 设置out_buffer大小
mqttClient->mqtt_state.connect_info = &(mqttClient->connect_info); // MQTT【CONNECT】报文的连接参数(指针)，赋值给mqttClient->mqtt_state.connect_info
// 初始化MQTT出站报文缓存区
mqtt_msg_init(&mqttClient->mqtt_state.mqtt_connection, mqttClient->mqtt_state.out_buffer, mqttClient->mqtt_state.out_buffer_length);
QUEUE_Init(&mqttClient->msgQueue, QUEUE_BUFFER_SIZE); // 消息队列初始化【队列可以存放一个/多个MQTT报文】
// 创建任务：任务函数【MQTT_Task】、优先级【2】、任务指针【mqtt_procTaskQueue】、消息深度【1】
system_os_task(MQTT_Task, MQTT_TASK_PRIO, mqtt_procTaskQueue, MQTT_TASK_QUEUE_SIZE);
// 安排任务：参数1=任务等级 / 参数2=消息类型 / 参数3=消息参数
system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)mqttClient); // 参数3的类型必须为【os_param_t】型
}
// 设置遗嘱：遗嘱主题【...】、遗嘱消息【...】、遗嘱质量【Will_Qos=0】、遗嘱保持【Will_Retain=0】
void ICACHE_FLASH_ATTR
MQTT_InitLWT(MQTT_Client *mqttClient, uint8_t* will_topic, uint8_t* will_msg, uint8_t will_qos, uint8_t will_retain)
{
uint32_t temp;
temp = os_strlen(will_topic);
mqttClient->connect_info.will_topic = (uint8_t*)os_zalloc(temp + 1); // 申请【遗嘱主题】的存放内存
os_strcpy(mqttClient->connect_info.will_topic, will_topic); // 赋值【遗嘱主题】
mqttClient->connect_info.will_topic[temp] = 0; // 最后'\0'
temp = os_strlen(will_msg);
mqttClient->connect_info.will_message = (uint8_t*)os_zalloc(temp + 1);
os_strcpy(mqttClient->connect_info.will_message, will_msg); // 赋值【遗嘱消息】
mqttClient->connect_info.will_message[temp] = 0;
mqttClient->connect_info.will_qos = will_qos; // 遗嘱质量【Will_Qos=0】
mqttClient->connect_info.will_retain = will_retain; // 遗嘱保持【Will_Retain=0】
}
user_init函数中的语句，例如：执行mqttClient->connectedCb(...)语句就相当于执行mqttConnectedCb(...)
// 执行 mqttClient->connectedCb(...) => mqttConnectedCb(...)
void ICACHE_FLASH_ATTR MQTT_OnConnected(MQTT_Client*mqttClient, MqttCallback connectedCb)
{
mqttClient->connectedCb = connectedCb; // 函数名【mqttConnectedCb】
}
// 执行 mqttClient->disconnectedCb(...) => mqttDisconnectedCb(...)
void ICACHE_FLASH_ATTR MQTT_OnDisconnected(MQTT_Client *mqttClient, MqttCallback disconnectedCb)
{
mqttClient->disconnectedCb = disconnectedCb; // 函数名【mqttDisconnectedCb】
}
// 执行 mqttClient->dataCb(...) => mqttDataCb(...)
void ICACHE_FLASH_ATTR MQTT_OnData(MQTT_Client *mqttClient, MqttDataCallback dataCb)
{
mqttClient->dataCb = dataCb; // 函数名【mqttDataCb】
}
// 执行 mqttClient->publishedCb(...) => mqttPublishedCb(...)
void ICACHE_FLASH_ATTR MQTT_OnPublished(MQTT_Client *mqttClient, MqttCallback publishedCb)
{
mqttClient->publishedCb = publishedCb; // 函数名【mqttPublishedCb】
}
// 执行 mqttClient->timeoutCb(...) => 【...】未定义函数
void ICACHE_FLASH_ATTR MQTT_OnTimeout(MQTT_Client *mqttClient, MqttCallback timeoutCb)
{
mqttClient->timeoutCb = timeoutCb;
}
WIFI_Connect函数来设置WIFI连接的参数，设置ESP8266设置为STA模式，设置STA参数、设置WIFI定时器，接入WIFI。void ICACHE_FLASH_ATTR WIFI_Connect(uint8_t* ssid, uint8_t* pass, WifiCallback cb)
{
struct station_config stationConf;
INFO("WIFI_INIT\r\n");
wifi_set_opmode_current(STATION_MODE); // 设置ESP8266为STA模式
//wifi_station_set_auto_connect(FALSE); // 上电不自动连接已记录的AP(已注释，即：上电自动连接已记录的AP(默认))
wifiCb = cb; // 函数名赋值：wifiCb可以作为函数名使用，wifiCb(..) == wifiConnectCb(..)
// 设置STA参数
os_memset(&stationConf, 0, sizeof(struct station_config)); // STA信息 = 0
os_sprintf(stationConf.ssid, "%s", ssid); // SSID赋值
os_sprintf(stationConf.password, "%s", pass); // 密码赋值
wifi_station_set_config_current(&stationConf); // 设置STA参数
// 设置WiFiLinker定时器
os_timer_disarm(&WiFiLinker); // 定时器：WIFI连接
os_timer_setfn(&WiFiLinker, (os_timer_func_t *)wifi_check_ip, NULL); // wifi_check_ip：检查IP获取情况
os_timer_arm(&WiFiLinker, 1000, 0); // 1秒定时(1次)
//wifi_station_set_auto_connect(TRUE); // 上电自动连接已记录AP
wifi_station_connect(); // ESP8266接入WIFI
}
// 定时函数：检查IP获取情况
static void ICACHE_FLASH_ATTR wifi_check_ip(void *arg)
{
struct ip_info ipConfig;
os_timer_disarm(&WiFiLinker); // 关闭WiFiLinker定时器
wifi_get_ip_info(STATION_IF, &ipConfig); // 获取IP地址
wifiStatus = wifi_station_get_connect_status(); // 获取接入状态
// 获取到IP地址
if (wifiStatus == STATION_GOT_IP && ipConfig.ip.addr != 0)
{
// 获取IP后，每2秒检查一次WIFI连接的正确性【防止WIFI掉线等情况】
os_timer_setfn(&WiFiLinker, (os_timer_func_t *)wifi_check_ip, NULL);
os_timer_arm(&WiFiLinker, 2000, 0);
}
// 未获取到IP地址
else
{
if(wifi_station_get_connect_status() == STATION_WRONG_PASSWORD)
{
INFO("STATION_WRONG_PASSWORD\r\n"); // 密码错误
wifi_station_connect();
}
else if(wifi_station_get_connect_status() == STATION_NO_AP_FOUND)
{
INFO("STATION_NO_AP_FOUND\r\n"); // 未发现对应AP
wifi_station_connect();
}
else if(wifi_station_get_connect_status() == STATION_CONNECT_FAIL)
{
INFO("STATION_CONNECT_FAIL\r\n"); // 连接失败
wifi_station_connect();
}
else
{
INFO("STATION_IDLE\r\n");
}
// 再次开启定时器
os_timer_setfn(&WiFiLinker, (os_timer_func_t *)wifi_check_ip, NULL);
os_timer_arm(&WiFiLinker, 500, 0); // 500Ms定时
}
// 如果WIFI状态改变，则调用[wifiConnectCb]函数
if(wifiStatus != lastWifiStatus)
{
lastWifiStatus = wifiStatus; // WIFI状态更新
if(wifiCb) // 判断是否设置了[wifiConnectCb]函数
wifiCb(wifiStatus); // wifiCb(wifiStatus)=wifiConnectCb(wifiStatus)
}
}
void wifiConnectCb（）函数void wifiConnectCb(uint8_t status)
{
// 成功获取到IP地址
//---------------------------------------------------------------------
if(status == STATION_GOT_IP)
{
ip_addr_t * addr = (ip_addr_t *)os_zalloc(sizeof(ip_addr_t));
// 在官方例程的基础上，增加2个备用服务器
//---------------------------------------------------------------
sntp_setservername(0, "us.pool.ntp.org"); // 服务器_0【域名】
sntp_setservername(1, "ntp.sjtu.edu.cn"); // 服务器_1【域名】
ipaddr_aton("210.72.145.44", addr); // 点分十进制 => 32位二进制
sntp_setserver(2, addr); // 服务器_2【IP地址】
os_free(addr); // 释放addr
sntp_init(); // SNTP初始化
// 设置SNTP定时器[sntp_timer]
//-----------------------------------------------------------
os_timer_disarm(&sntp_timer);
os_timer_setfn(&sntp_timer, (os_timer_func_t *)sntpfn, NULL);
os_timer_arm(&sntp_timer, 1000, 1); // 1s定时
}
// IP地址获取失败
//----------------------------------------------------------------
else
{
MQTT_Disconnect(&mqttClient); // WIFI连接出错，TCP断开连接
}
}
目的：获取当前网络时间
void sntpfn()
{
u32_t ts = 0;
ts = sntp_get_current_timestamp(); // 获取当前的偏移时间
os_printf("current time : %s\n", sntp_get_real_time(ts)); // 获取真实时间
if (ts == 0) // 网络时间获取失败
{
os_printf("did not get a valid time from sntp server\n");
}
else //(ts != 0) // 网络时间获取成功
{
os_timer_disarm(&sntp_timer); // 关闭SNTP定时器
MQTT_Connect(&mqttClient); // 开始MQTT连接
}
}
typedef unsigned long u32_t;
sntp_get_current_timestamp在我之前写的学习日记——ESP8266STNP博客中有详细的介绍。https://blog.csdn.net/quanqueen/article/details/108045372
sntp_get_current_timestamp：功能：查询当前距离基准时间（ 1970.01.01 00： 00： 00 GMT + 8）的时间戳，单位：秒。函数定义：uint32 sntp_get_current_timestamp()此定义在sntp.h文件中已经定义过了，所以我们只需将添加头文件即可使用。返回的是距离基准时间的时间戳。sntp_get_real_time功能：查询实际时间 (GMT + 8)ts = sntp_get_current_timestamp();我们将基准时间相距的时间戳赋值给了ts，之后直接输出真实的时间即可。 os_printf("current time : %s\n", sntp_get_real_time(ts));。void ICACHE_FLASH_ATTR MQTT_Connect（）这个函数，进行MQTT连接准备。void ICACHE_FLASH_ATTR MQTT_Connect(MQTT_Client *mqttClient)
{
//espconn_secure_set_size(0x01,6*1024); // SSL双向认证时才需使用 // try to modify memory size 6*1024 if ssl/tls handshake failed
// 开始MQTT连接前，判断是否存在MQTT的TCP连接。如果有，则清除之前的TCP连接
if (mqttClient->pCon)
{
// Clean up the old connection forcefully - using MQTT_Disconnect
// does not actually release the old connection until the
// disconnection callback is invoked.
mqtt_tcpclient_delete(mqttClient); // 删除TCP连接、释放pCon内存、清除TCP连接指针
}
// TCP连接设置
mqttClient->pCon = (struct espconn *)os_zalloc(sizeof(struct espconn)); // 申请pCon内存
mqttClient->pCon->type = ESPCONN_TCP; // 设为TCP连接
mqttClient->pCon->state = ESPCONN_NONE;
mqttClient->pCon->proto.tcp = (esp_tcp *)os_zalloc(sizeof(esp_tcp)); // 申请esp_tcp内存
mqttClient->pCon->proto.tcp->local_port = espconn_port(); // 获取ESP8266可用端口
mqttClient->pCon->proto.tcp->remote_port = mqttClient->port; // 设置端口号
mqttClient->pCon->reverse = mqttClient; // mqttClient->pCon->reverse 缓存 mqttClient指针
espconn_regist_connectcb(mqttClient->pCon, mqtt_tcpclient_connect_cb); // 注册TCP连接成功的回调函数
espconn_regist_reconcb(mqttClient->pCon, mqtt_tcpclient_recon_cb); // 注册TCP异常中断的回调函数
mqttClient->keepAliveTick = 0; // MQTT客户端(ESP8266)心跳计数
mqttClient->reconnectTick = 0; // 重连等待计时：当进入重连请求状态后，需等待5秒，之后进行重新连接
// 设置MQTT定时(1秒)【功能：心跳计时、重连计时、TCP发送计时】
os_timer_disarm(&mqttClient->mqttTimer);
os_timer_setfn(&mqttClient->mqttTimer, (os_timer_func_t *)mqtt_timer, mqttClient); // mqtt_timer
os_timer_arm(&mqttClient->mqttTimer, 1000, 1); // 1秒定时(重复)
// 打印SSL配置：安全类型[NO_TLS == 0]
os_printf("your ESP SSL/TLS configuration is %d.[0:NO_TLS\t1:TLS_WITHOUT_AUTHENTICATION\t2ONE_WAY_ANTHENTICATION\t3TWO_WAY_ANTHENTICATION]\n",DEFAULT_SECURITY);
// 解析点分十进制形式的IP地址
if (UTILS_StrToIP(mqttClient->host, &mqttClient->pCon->proto.tcp->remote_ip)) // 解析IP地址(点分十进制字符串形式)
{
INFO("TCP: Connect to ip %s:%d\r\n", mqttClient->host, mqttClient->port); // 打印IP地址
// 根据安全类型，调用不同的TCP连接方式
if (mqttClient->security) // 安全类型 != 0
{
#ifdef MQTT_SSL_ENABLE
if(DEFAULT_SECURITY >= ONE_WAY_ANTHENTICATION ) // 单向认证【ONE_WAY_ANTHENTICATION = 2】
{
espconn_secure_ca_enable(ESPCONN_CLIENT,CA_CERT_FLASH_ADDRESS);
}
if(DEFAULT_SECURITY >= TWO_WAY_ANTHENTICATION) // 双向认证【TWO_WAY_ANTHENTICATION = 3】
{
espconn_secure_cert_req_enable(ESPCONN_CLIENT,CLIENT_CERT_FLASH_ADDRESS);
}
espconn_secure_connect(mqttClient->pCon); // 不认证【TLS_WITHOUT_AUTHENTICATION = 1】
#else
INFO("TCP: Do not support SSL\r\n");
#endif
}
else // 安全类型 = 0 = NO_TLS
{
espconn_connect(mqttClient->pCon); // TCP连接(作为Client连接Server)
}
}
// 解析域名
else
{
INFO("TCP: Connect to domain %s:%d\r\n", mqttClient->host, mqttClient->port);
espconn_gethostbyname(mqttClient->pCon, mqttClient->host, &mqttClient->ip, mqtt_dns_found);
}
mqttClient->connState = TCP_CONNECTING; // TCP正在连接
}
void ICACHE_FLASH_ATTR MQTT_Connect（）之后调用uint8_t ICACHE_FLASH_ATTR UTILS_StrToIP()这个函数。uint8_t ICACHE_FLASH_ATTR UTILS_StrToIP(const int8_t* str, void *ip)
{
/* The count of the number of bytes processed. */
int i;
/* A pointer to the next digit to process. */
const char * start;
start = str; // 字符串指针赋值
// IPV4
for (i = 0; i < 4; i++) // 一共四个十进制字符串
{
/* The digit being processed. */
char c;
/* The value of this byte. */
int n = 0;
while (1)
{
c = * start; // 字符串某字符赋值
start++; // 从前往后(从左向右)
if (c >= '0' && c <= '9') // 在“0～9”范围内
{
n *= 10; // 权重
n += c - '0'; // 将'0'～'9'字符转换为对应的数字
}
/* We insist on stopping at "." if we are still parsing
the first, second, or third numbers. If we have reached
the end of the numbers, we will allow any character. */
else if ((i < 3 && c == '.') || i == 3)
{
break; // 遇到'.'则解析下一十进制字符串
}
else
{
return 0; // 解析失败
}
}
if (n >= 256) // n过大，解析失败
{
return 0;
}
((uint8_t*)ip)[i] = n; // 赋值给IP数组
}
return 1; // 解析成功，返回1
}
LOCAL void ICACHE_FLASH_ATTR mqtt_dns_found(const char *name, ip_addr_t *ipaddr, void *arg)
{
struct espconn *pConn = (struct espconn *)arg; // 获取TCP连接指针
MQTT_Client* client = (MQTT_Client *)pConn->reverse; // 获取mqttClient指针
if (ipaddr == NULL) // 域名解析失败
{
INFO("DNS: Found, but got no ip, try to reconnect\r\n");
client->connState = TCP_RECONNECT_REQ; // TCP重连请求(等待5秒)
return;
}
INFO("DNS: found ip %d.%d.%d.%d\n", // 打印域名对应的IP地址
*((uint8 *) &ipaddr->addr),
*((uint8 *) &ipaddr->addr + 1),
*((uint8 *) &ipaddr->addr + 2),
*((uint8 *) &ipaddr->addr + 3));
// 判断IP地址是否正确(?=0)
if (client->ip.addr == 0 && ipaddr->addr != 0) // 未保存IP地址：mqttClient->ip.addr == 0
{
os_memcpy(client->pCon->proto.tcp->remote_ip, &ipaddr->addr, 4); // IP赋值
// 根据安全类型，调用不同的TCP连接方式
if (client->security) // 安全类型 != 0
{
#ifdef MQTT_SSL_ENABLE
if(DEFAULT_SECURITY >= ONE_WAY_ANTHENTICATION ) // 单向认证【ONE_WAY_ANTHENTICATION = 2】
{
espconn_secure_ca_enable(ESPCONN_CLIENT,CA_CERT_FLASH_ADDRESS);
}
if(DEFAULT_SECURITY >= TWO_WAY_ANTHENTICATION) // 双向认证【TWO_WAY_ANTHENTICATION = 3】
{ espconn_secure_cert_req_enable(ESPCONN_CLIENT,CLIENT_CERT_FLASH_ADDRESS);
}
espconn_secure_connect(client->pCon); // 不认证【TLS_WITHOUT_AUTHENTICATION = 1】
#else
INFO("TCP: Do not support SSL\r\n");
#endif
}
else // 安全类型 = 0 = NO_TLS
{
espconn_connect(client->pCon); // TCP连接(作为Client连接Server)
}
client->connState = TCP_CONNECTING; // TCP正在连接
INFO("TCP: connecting...\r\n");
}
system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client); // 安排任务MQTT_Task
}
espconn_connect函数来将ESP8266作为TCPclient连接到TCPServer。void ICACHE_FLASH_ATTR mqtt_tcpclient_connect_cb(void *arg)
{
struct espconn *pCon = (struct espconn *)arg; // 获取TCP连接指针
MQTT_Client* client = (MQTT_Client *)(pCon->reverse);// 获取mqttClient指针
// 注册回调函数
espconn_regist_disconcb(client->pCon, mqtt_tcpclient_discon_cb); // TCP断开成功_回调
espconn_regist_recvcb(client->pCon, mqtt_tcpclient_recv); // TCP接收成功_回调
espconn_regist_sentcb(client->pCon, mqtt_tcpclient_sent_cb); // TCP发送成功_回调
INFO("MQTT: Connected to broker %s:%d\r\n", client->host, client->port);
// 【CONNECT】报文发送准备
// 初始化MQTT报文缓存区
mqtt_msg_init(&client->mqtt_state.mqtt_connection, client->mqtt_state.out_buffer, client->mqtt_state.out_buffer_length);
// 配置【CONNECT】控制报文，并获取【CONNECT】报文[指针]、[长度]
client->mqtt_state.outbound_message = mqtt_msg_connect(&client->mqtt_state.mqtt_connection, client->mqtt_state.connect_info);
// 获取待发送的报文类型(此处是【CONNECT】报文)
client->mqtt_state.pending_msg_type = mqtt_get_type(client->mqtt_state.outbound_message->data);
// 获取待发送报文中的【报文标识符】（【CONNECT】报文中没有）
client->mqtt_state.pending_msg_id = mqtt_get_id(client->mqtt_state.outbound_message->data,client->mqtt_state.outbound_message->length);
// TCP发送成功/报文发送5秒计时结束 => 报文发送结束(sendTimeout=0)
client->sendTimeout = MQTT_SEND_TIMOUT; // 发送MQTT报文时，sendTimeout=5
INFO("MQTT: Sending, type: %d, id: %04X\r\n", client->mqtt_state.pending_msg_type, client->mqtt_state.pending_msg_id);
// TCP：发送【CONNECT】报文
if (client->security) // 安全类型 != 0
{
#ifdef MQTT_SSL_ENABLE
espconn_secure_send(client->pCon, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length);
#else
INFO("TCP: Do not support SSL\r\n");
#endif
}
else // 安全类型 = 0 = NO_TLS
{
// TCP发送：数据=[client->mqtt_state.outbound_message->data]、长度=[client->mqtt_state.outbound_message->length]
espconn_send(client->pCon, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length);
}
client->mqtt_state.outbound_message = NULL; // 报文发送完后，清除出站报文指针
client->connState = MQTT_CONNECT_SENDING; // 状态设为：MQTT【CONNECT】报文发送中【MQTT_CONNECT_SENDING】
system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client); // 安排任务MQTT_Task
}
// mqtt_msg_connect(&client->mqtt_state.mqtt_connection, client->mqtt_state.connect_info)
mqtt_message_t* ICACHE_FLASH_ATTR mqtt_msg_connect(mqtt_connection_t* connection, mqtt_connect_info_t* info)
{
struct mqtt_connect_variable_header* variable_header; // 【CONNECT】报文的【可变报头】指针
init_message(connection); // 设置报文长度 = 3(暂时设为【固定报头】长度(3)，之后添加【可变报头】、【有效载荷】)
// 判断消息长度是否超过缓存区长度 // 【注：[message.length]是指TCP传输的整个MQTT报文长度】
if(connection->message.length + sizeof(*variable_header) > connection->buffer_length) // 判断MQTT报文长度
return fail_message(connection);
// 跳过了对【固定报头】的赋值，只为【固定报头】保留了3个字节的空间。 注：剩余长度最多占两字节。
// 获取【可变报头】指针，并更新报文长度
variable_header = (void*)(connection->buffer + connection->message.length); // 【可变报头】指针 = 报文缓存区指针+3(固定报头)
connection->message.length += sizeof(*variable_header); // 报文长度 == 固定报头 + 可变报头
// 协议名、协议级别赋值
variable_header->lengthMsb = 0; // lengthMsb
#if defined(PROTOCOL_NAMEv31)
variable_header->lengthLsb = 6; // lengthLsb
memcpy(variable_header->magic, "MQIsdp", 6);
variable_header->version = 3; // v31版本 = 3
#elif defined(PROTOCOL_NAMEv311)
variable_header->lengthLsb = 4; // lengthLsb
memcpy(variable_header->magic, "MQTT", 4);
variable_header->version = 4; // v311版本 = 4
#else
#error "Please define protocol name"
#endif
variable_header->flags = 0; // 连接标志字节 = 0（暂时清0，待会赋值）
// 保持连接时长赋值
variable_header->keepaliveMsb = info->keepalive >> 8; // 赋值高字节
variable_header->keepaliveLsb = info->keepalive & 0xff; // 赋值低字节
// clean_session = 1：客户端和服务端必须丢弃之前的任何会话并开始一个新的会话
if(info->clean_session)
variable_header->flags |= MQTT_CONNECT_FLAG_CLEAN_SESSION; //clean_session=1
// 判断是否存在[client_id]，存在则设置[client_id]字段
if(info->client_id != NULL && info->client_id[0] != '\0')
{
// 将[client_id]字段添加到报文缓存区，报文长度+=[client_id]所占长度
if(append_string(connection, info->client_id, strlen(info->client_id)) < 0)
return fail_message(connection);
}
else
return fail_message(connection); // 报文出错
// 判断是否存在[will_topic]
if(info->will_topic != NULL && info->will_topic[0] != '\0')
{
// 将[will_topic]字段添加到报文缓存区，报文长度+=[will_topic]所占长度
if(append_string(connection, info->will_topic,strlen(info->will_topic))<0)
return fail_message(connection);
// 将[will_message]字段添加到报文缓存区，报文长度+=[will_message]所占长度
if(append_string(connection,info->will_message,strlen(info->will_message))<0)
return fail_message(connection);
// 设置【CONNECT】报文中的Will标志位：[Will Flag]、[Will QoS]、[Will Retain]
variable_header->flags |= MQTT_CONNECT_FLAG_WILL; // 遗嘱标志位 = 1
if(info->will_retain)
variable_header->flags |= MQTT_CONNECT_FLAG_WILL_RETAIN;// WILL_RETAIN = 1
variable_header->flags |= (info->will_qos & 3) << 3; // will质量赋值
}
append_string这个函数来将参数字符串添加到出栈报文缓存区。在字符串前添加两个字符串得前缀来表示这个字符串得长度，之后再返回参数字符串在MQTT控制报文中所占得长度。static int ICACHE_FLASH_ATTR append_string(mqtt_connection_t* connection, const char* string, int len)
{
if(connection->message.length + len + 2 > connection->buffer_length) // 判断报文是否过长
return -1;
// 设置字符串前的两字节前缀，表示此字符串的长度
connection->buffer[connection->message.length++] = len >> 8; // 高八位
connection->buffer[connection->message.length++] = len & 0xff; // 低八位
memcpy(connection->buffer+connection->message.length, string, len); // 将[参数字符串]添加到报文缓存区
connection->message.length += len; // 报文长度 += [参数字符串]所占长度
return len + 2; // 返回[参数字符串]在MQTT控制报文中所占长度
}
在。最后，调用fini_message函数，这是CONNECT控制报的固定报头。
static mqtt_message_t* ICACHE_FLASH_ATTR fini_message(mqtt_connection_t* connection, int type, int dup, int qos, int retain)
{
int remaining_length = connection->message.length - MQTT_MAX_FIXED_HEADER_SIZE; // 获取【可变报头】+【有效载荷】长度
// 设置固定报头(固定头中的剩余长度使用变长度编码方案，详情请参考MQTT协议手册)
if(remaining_length > 127) // 剩余长度占2字节
{
connection->buffer[0] = ((type&0x0f)<<4)|((dup&1)<<3)|((qos&3)<<1)|(retain&1); // 固定头的首字节赋值
connection->buffer[1] = 0x80 | (remaining_length % 128); // 剩余长度的第一个字节
connection->buffer[2] = remaining_length / 128; // 剩余长度的第二个字节
connection->message.length = remaining_length + 3; // 报文的整个长度
connection->message.data = connection->buffer; // MQTT报文指针 -> 出站报文缓存区首地址
}
else //if(remaining_length<= 127) // 剩余长度占1字节
{
// buffer[0] = 无
connection->buffer[1] = ((type&0x0f)<<4)|((dup&1)<<3)|((qos&3)<<1)|(retain&1); // 固定头的首字节赋值
connection->buffer[2] = remaining_length; // 固定头中的[剩余长度](可变报头+负载数据)
connection->message.length = remaining_length + 2; // 报文的整个长度
connection->message.data = connection->buffer + 1; // MQTT报文指针 -> 出站报文缓存区首地址+1
}
return &connection->message; // 返回报文首地址【报文数据、报文整体长度】
}
int type参数是报文类型int dup, int qos, int retain这三个参数是报文类型标志位。也就是固定报文的第一个字节的第四位，除了PUBLISH报文的报文类型标志位是由重复分发标志[dup][Bit3]、服务质量[qos][Bit2～1]、报文保留标志[retain][Bit1=0]组成。其余类型报文的报文类型标志位是固定的。mqtt_tcpclient_recv网络回调函数，在回调函数中会判断当前接收到的MQTT报文是否是CONNACK请求报文。并且判断ESP8266是否是请求MATT连接状态。如果是，那么将ESP8266设置为MQTT_DATA状态，并且执行MQTT连接成功的函数。void mqttConnectedCb(uint32_t *args)
{
MQTT_Client* client = (MQTT_Client*)args; // 获取mqttClient指针
INFO("MQTT: Connected\r\n");
// 【参数2：主题过滤器 / 参数3：订阅Qos】
MQTT_Subscribe(client, "SW_LED", 0); // 订阅主题"SW_LED"，QoS=0
// MQTT_Subscribe(client, "SW_LED", 1);
// MQTT_Subscribe(client, "SW_LED", 2);
// 【参数2：主题名 / 参数3：发布消息的有效载荷 / 参数4：有效载荷长度 / 参数5：发布Qos / 参数6：Retain】
MQTT_Publish(client, "SW_LED", "ESP8266_Online", strlen("ESP8266_Online"), 0, 0); // 向主题"SW_LED"发布"ESP8266_Online"，Qos=0、retain=0
// MQTT_Publish(client, "SW_LED", "ESP8266_Online", strlen("ESP8266_Online"), 1, 0);
// MQTT_Publish(client, "SW_LED", "ESP8266_Online", strlen("ESP8266_Online"), 2, 0);
}
I16 ICACHE_FLASH_ATTR PROTO_AddRb(RINGBUF *rb, const U8 *packet, I16 len)
{
U16 i = 2;
if(RINGBUF_Put(rb,0x7E)==-1) return -1; // 向当前队列写指针指向处写入【起始码：0x7E】
while (len--) // 循环[len]次(报文所有字节)
{
switch (*packet) // 获取当前数据包的一个字节
{
case 0x7D: // 判断数据 ?= 【0x7D】/【0x7E】/【0x7F】
case 0x7E:
case 0x7F:
// 如果数据==[0x7D]||[0x7E]||[0x7F]，都在此数据前写入[0x7D]【因为[0x7E]==起始码、[0x7F]==结束码】
if(RINGBUF_Put(rb, 0x7D) == -1) return -1; // 在此数据前写入[0x7D]
if(RINGBUF_Put(rb, *packet++^0x20) == -1) return -1; // 【0x7D/0x7E/0x7F】^=0x20，写入队列(注：a^b^b == a)
i += 2; // 写入队列的字节数+2
break;
// 数据包当前数据不是特殊码，则正常写入
default:
if(RINGBUF_Put(rb, *packet++) == -1) return -1; // 写入数据包指针对应值
i++; // 写入队列的字节数+1
break;
}
}
if(RINGBUF_Put(rb, 0x7F) == -1) return -1; // 向当前队列写指针指向处写入[结束码：0x7F]
return i; // 返回写入数量(包括起始码、结束码)
}
BOOL ICACHE_FLASH_ATTR MQTT_Subscribe(MQTT_Client *client, char* topic, uint8_t qos)
{
uint8_t dataBuffer[MQTT_BUF_SIZE]; // 解析后报文缓存(1204字节)
uint16_t dataLen; // 解析后报文长度
// 配置【SUBSCRIBE】报文，并获取【SUBSCRIBE】报文[指针]、[长度]
client->mqtt_state.outbound_message=mqtt_msg_subscribe(&client->mqtt_state.mqtt_connection,topic, qos,&client->mqtt_state.pending_msg_id);
INFO("MQTT: queue subscribe, topic\"%s\", id: %d\r\n", topic, client->mqtt_state.pending_msg_id);
// 将报文写入队列，并返回写入字节数(包括特殊码)
while(QUEUE_Puts(&client->msgQueue, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length) == -1)
{
INFO("MQTT: Queue full\r\n");
// 解析队列中的报文
if (QUEUE_Gets(&client->msgQueue, dataBuffer, &dataLen, MQTT_BUF_SIZE) == -1) // 解析失败 = -1
{
INFO("MQTT: Serious buffer error\r\n");
return FALSE;
}
}
system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client); // 安排任务MQTT_Task
return TRUE;
}
QUEUE_Puts函数将订阅主题主题报文写入队列当中，之后安排系统任务，订阅主题函数得参数2是主题过滤器，参数3是订阅质量。BOOL ICACHE_FLASH_ATTR MQTT_Publish(MQTT_Client *client, const char* topic, const char* data, int data_length, int qos, int retain)
{
uint8_t dataBuffer[MQTT_BUF_SIZE]; // 解析后报文缓存(1204字节)
uint16_t dataLen; // 解析后报文长度
// 配置【PUBLISH】报文，并获取【PUBLISH】报文[指针]、[长度]
client->mqtt_state.outbound_message = mqtt_msg_publish(&client->mqtt_state.mqtt_connection,
topic, data, data_length,
qos, retain,
&client->mqtt_state.pending_msg_id);
if (client->mqtt_state.outbound_message->length == 0) // 判断报文是否正确
{
INFO("MQTT: Queuing publish failed\r\n");
return FALSE;
}
// 串口打印：【PUBLISH】报文长度,(队列装填数量/队列大小)
INFO("MQTT: queuing publish, length: %d, queue size(%d/%d)\r\n", client->mqtt_state.outbound_message->length, client->msgQueue.rb.fill_cnt, client->msgQueue.rb.size);
// 将报文写入队列，并返回写入字节数(包括特殊码)
while (QUEUE_Puts(&client->msgQueue, client->mqtt_state.outbound_message->data, client->mqtt_state.outbound_message->length) == -1)
{
INFO("MQTT: Queue full\r\n"); // 队列已满
// 解析队列中的数据包
if (QUEUE_Gets(&client->msgQueue, dataBuffer, &dataLen, MQTT_BUF_SIZE) == -1) // 解析失败 = -1
{
INFO("MQTT: Serious buffer error\r\n");
return FALSE;
}
}
system_os_post(MQTT_TASK_PRIO, 0, (os_param_t)client); // 安排任务
return TRUE;
}
// MQTT任务函数【任务：根据ESP8266运行状态，执行相应操作】
// TCP_RECONNECT_REQ TCP重连请求(等待5秒) 退出Tsak(5秒后，进入TCP_RECONNECT状态)
// TCP_RECONNECT TCP重新连接 执行MQTT连接准备，并设置ESP8266状态
// MQTT_DELETING MQTT正在删除 TCP断开连接
// TCP_DISCONNECTING TCP正在断开
// TCP_RECONNECT_DISCONNECTING TCP暂时断开(断开后会重连)
// TCP_DISCONNECTED TCP成功断开 删除TCP连接，并释放pCon内存
// MQTT_DELETED MQTT已删除 删除MQTT客户端，并释放相关内存
// MQTT_KEEPALIVE_SEND MQTT心跳 向服务器发送心跳报文
// MQTT_DATA MQTT数据传输 TCP发送队列中的报文
void ICACHE_FLASH_ATTR MQTT_Task(os_event_t *e) // 不判断消息类型
{
INFO("\r\n------------- MQTT_Task -------------\r\n");
MQTT_Client* client = (MQTT_Client*)e->par; // 【e->par】 == 【mqttClient指针的值】,所以需类型转换
uint8_t dataBuffer[MQTT_BUF_SIZE]; // 数据缓存区(1204字节)
uint16_t dataLen; // 数据长度
if (e->par == 0) // 没有mqttClient指针，错误
return;
// 根据ESP8266运行状态，执行相应操作
switch (client->connState)
{
// TCP重连请求(等待5秒)，退出Tsak
case TCP_RECONNECT_REQ: break;
// TCP重新连接：执行MQTT连接准备，并设置ESP8266状态
case TCP_RECONNECT:
mqtt_tcpclient_delete(client); // 删除TCP连接、释放pCon内存、清除TCP连接指针
MQTT_Connect(client); // MQTT连接准备：TCP连接、域名解析等
INFO("TCP: Reconnect to: %s:%d\r\n", client->host, client->port);
client->connState = TCP_CONNECTING; // TCP正在连接
break;
// MQTT正在删除、TCP正在断开、【心跳请求】报文发送失败：TCP断开连接
case MQTT_DELETING:
case TCP_DISCONNECTING:
case TCP_RECONNECT_DISCONNECTING:
if (client->security) // 安全类型 != 0
{
#ifdef MQTT_SSL_ENABLE
espconn_secure_disconnect(client->pCon);
#else
INFO("TCP: Do not support SSL\r\n");
#endif
}
else // 安全类型 = 0 = NO_TLS
{
espconn_disconnect(client->pCon); // TCP断开连接
}
break;
// TCP成功断开
case TCP_DISCONNECTED:
INFO("MQTT: Disconnected\r\n");
mqtt_tcpclient_delete(client); // 删除TCP连接、释放pCon内存、清除TCP连接指针
break;
// MQTT已删除：ESP8266的状态为[MQTT已删除]后，将MQTT相关内存释放
case MQTT_DELETED:
INFO("MQTT: Deleted client\r\n");
mqtt_client_delete(client); // 删除MQTT客户端，并释放相关内存
break;
// MQTT客户端存活报告
case MQTT_KEEPALIVE_SEND:
mqtt_send_keepalive(client); // 向MQTT服务器发送【心跳】报文
break;
// MQTT传输数据状态
case MQTT_DATA:
if (QUEUE_IsEmpty(&client->msgQueue) || client->sendTimeout != 0)
{
break; // 【队列为空 || 发送未结束】，不执行操作
}
// 【队列非空 && 发送结束】：解析并发送 队列中的报文
if (QUEUE_Gets(&client->msgQueue, dataBuffer, &dataLen, MQTT_BUF_SIZE) == 0) // 解析成功 = 0
{
client->mqtt_state.pending_msg_type = mqtt_get_type(dataBuffer); // 获取报文中的【报文类型】
client->mqtt_state.pending_msg_id = mqtt_get_id(dataBuffer, dataLen); // 获取报文中的【报文标识符】
client->sendTimeout = MQTT_SEND_TIMOUT; // 发送MQTT报文时，sendTimeout=5
INFO("MQTT: Sending, type: %d, id: %04X\r\n", client->mqtt_state.pending_msg_type, client->mqtt_state.pending_msg_id);
// 发送报文
if (client->security) // 安全类型 != 0
{
#ifdef MQTT_SSL_ENABLE
espconn_secure_send(client->pCon, dataBuffer, dataLen);
#else
INFO("TCP: Do not support SSL\r\n");
#endif
}
else // 安全类型 = 0 = NO_TLS
{
espconn_send(client->pCon, dataBuffer, dataLen); // TCP发送数据包
}
client->mqtt_state.outbound_message = NULL; // 报文发送完后，清除出站报文指针
break;
}
break;
}
} // 函数【MQTT_Task】结束
QUEUE_IsEmpty函数来判断队列是否为空，如果队列不为空，则调用QUEUE_Gets来解析一个完整得报文，
https://www.bilibili.com/video/BV1dJ411S723?p=53
http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html