MQTT移植笔记及源码分析（基于w601）
⼀、MQTT嵌⼊式源码获取
eclip源码中包含linux和freertos的移植⽂件，使⽤起来⽐较⽅便。
⼆、mqtt源码移植需要实现的函数接⼝
mqtt移植需要实现的函数接⼝主要有两类，⼀类是倒计时相关的接⼝函数，⼀类是⽹络连接相关的函数。
typedef struct Timer_{
struct timeval end_time;
}Timer;
void TimerInit(Timer*);//初始化时间
char TimerIsExpired(Timer*);//判断时间是否⽤尽
void TimerCountdownMS(Timer*,unsigned int);//以ms为单位进⾏倒计时
void TimerCountdown(Timer*,unsigned int);//以秒为单位进⾏倒计时
int TimerLeftMS(Timer*);//查看倒计时剩余时间
⽹络相关的接⼝函数
int linux_read(Network*,unsigned char*,int,int);//读socket数据
int linux_write(Network*,unsigned char*,int,int);//向socket写数据
void NetworkInit(Network*);//⽹络初始化
int NetworkConnect(Network*,char*,int);//连接mqtt服务器
void NetworkDisconnect(Network*);//断开socket连接
三、移植踩过的坑
3.1、tcp recv返回ENOTCONN和EAGAIN错误
1. 初次移植好mqtt之后，mqtt连接上服务器⼤约20s左右，recv函数就会报ENOTCONN，后来才发现mqtt⼼跳开关c-
>keepAliveInterval为0。在keepalive函数中，如果这个值为0，系统是不会向服务器发送⼼跳包的。修改c->keepAliveInterval的值即可；
2. EAGAIN是因为recv超时，属于正常情况，这时linux_read需要返回0；
3.2、mqtt订阅消息回调打印消息内容时间太长
1. 出现这个问题的主要原因是w601的printf函数在mqtt的回调函数中使⽤有问题，导致系统在topic的回调函数中停留了5s左右的时
间，使⽤rtthread内容的rt_kprintf函数可以避免这个问题。
3.3、发布qos⼤于0的消息时，publish函数经常返回-1
儿歌童谣顺口溜
这时因为我在主函数中有如下代码：
while(!toStop)
{
MQTTYield(&c,1000);//轮训接收和处理mqtt消息
}
MQTTYield会不停的调⽤lwip的recv函数读取⽹络中的消息，，如果qos⼤于0，publish函数会调⽤waitfor等待服务器的回复，⽽waitfor 中也会调⽤lwip的recv函数等待⽹络消息，这就会产⽣两个线程同时访问同⼀个socket的问题。这个问题的解决⽅法有很多，⽐如可以使⽤mqtt源码本⾝提供的加锁功能，也可以修改源码，使⽤队列或邮箱的⽅式通知publish的结果。
3.4、源码本⾝存在⼀些bug
1. 源码在处理mqtt的ping消息回复时，每⼀次调⽤MQTTYield都会检查服务器是否回复了ping消息。有时候因为⽹络不好等因素，服
务器回复ping包的时间会⼤于⼀次MQTTYield的时间，这时代码中会认为ping失败，⽽实际上这种情况属于正常现象。解决办法是对源代码的⼏处代码进⾏简单的修改即可。
四、MQTT源代码分析
1. ⾸先我们看下官⽅提供的基于linux的例程，此处省略了部分⽆关代码：
Network n;
MQTTClient c;
NetworkInit(&n);//初始化mqtt客户端的⽹络参数
NetworkConnect(&n, opts.host, opts.port);//连接mqtt服务器
MQTTClientInit(&c,&n,1000, buf,100, readbuf,100);//初始化mqtt的MQTTClient结构体，以及接收发送数据的buffer
MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
data.willFlag =0;//是否使⽤遗嘱消息
data.MQTTVersion =3;//mqtt版本号
data.clientID.cstring = opts.clientid;//客户端ID
data.urname.cstring = opts.urname;//⽤户名
data.password.cstring = opts.password;//密码
data.keepAliveInterval =10;//mqtt⼼跳时间间隔
data.cleanssion =1;
rc =MQTTConnect(&c,&data);//发送mqtt的CONNECT数据包到服务器
rc =MQTTSubscribe(&c, topic, opts.qos, messageArrived);//订阅mqtt消息
while(!toStop)
{
MQTTYield(&c,1000);//轮训接收和处理mqtt消息
}
MQTTDisconnect(&c);//断开mqtt连接
NetworkDisconnect(&n);//断开tcp连接
2. MQTTPacket_connectData结构体
typedef struct
一抹香
{
/** The eyecatcher for this structure. must be MQTC. */
char struct_id[4];
/** The version number of this structure. Must be 0 */
int struct_version;
/** Version of MQTT to be ud. 3 = 3.1 4 = 3.1.1
*/
unsigned char MQTTVersion;
MQTTString clientID;
unsigned short keepAliveInterval;
unsigned char cleanssion;
unsigned char willFlag;
MQTTPacket_willOptions will;
MQTTString urname;
MQTTString password;
} MQTTPacket_connectData;
clientID 客户端ID，唯⼀标识⼀个客户端，每⼀台设备的clientID必须不同，这个ID通常可以⽤设备的mac地址组合得到；
keepAliveInterval⼼跳间隔，表⽰每隔多长时间向服务器发送⼀次⼼跳包，单位是s；
cleanssion客户端连接断开是否清除服务器端关于客户端的连接信息；
willFlag 遗嘱消息
MQTTPacket_willOptions遗嘱消息的内容
urname和password⽤户名和密码
3. 订阅和发布消息
什么灯不会亮mqtt订阅和发布消息主要有三个函数：
int MQTTSubscribe(MQTTClient* client,const char* topicFilter,enum QoS, messageHandler);
int MQTTUnsubscribe(MQTTClient* client,const char* topicFilter);
int MQTTPublish(MQTTClient* client,const char*, MQTTMessage*);
MQTTSubscribe⽤来订阅名为topicFilter的主题，指定Qos等级，服务器发布的消息会在messageHandler函数中得到处理；
MQTTUnsubscribe取消订阅
MQTTPublish发布mqtt消息
5. MQTTYield函数分析
int MQTTYield(MQTTClient* c,int timeout_ms)
{
int rc = SUCCESS;
冷笑话谜语大全
Timer timer;
TimerInit(&timer);//初始化定时器
大学多久开学
TimerCountdownMS(&timer, timeout_ms);//给定时器的计数器赋初始值
do{
if(cycle(c,&timer)<0)
{
rc = FAILURE;
break;
}
}while(!TimerIsExpired(&timer));
新娘礼服return rc;
}
int cycle(MQTTClient* c, Timer* timer)
{
int len =0,
rc = SUCCESS;
int packet_type =readPacket(c, timer);//调⽤tcp-ip的recv函数从服务器读取数据，阻塞接收 /* read the socket, e what work is due */
switch(packet_type)//判断数据包的类型
{
default:
/* no more data to read, unrecoverable. Or read packet fails due to unexpected network error */
packet_type = SUCCESS;
梦幻西游法宝位置goto exit;
ca0:/* timed out reading packet */
break;
ca CONNACK://服务器响应连接请求
ca PUBACK://服务器响应客户端发布的消息
ca SUBACK://服务器响应客户端订阅的消息
ca UNSUBACK://服务器响应客户端取消订阅的消息
break;
ca PUBLISH://客户端收到其他客户端或服务器发布的消息
{
{
printf("packet type:%d\r\n",packet_type);
MQTTString topicName;
MQTTMessage msg;
int intQoS;
msg.payloadlen =0;/* this is a size_t, but derialize publish ts this as int */
if(MQTTDerialize_publish(&msg.dup,&intQoS,&ained,&msg.id,&topicName,
(unsigned char**)&msg.payload,(int*)&msg.payloadlen, c->readbuf, c->readbuf_size)!=1)
goto exit;
msg.qos =(enum QoS)intQoS;
deliverMessage(c,&topicName,&msg);//查询是否有符合的topic
if(msg.qos != QOS0)
{
if(msg.qos == QOS1)
len =MQTTSerialize_ack(c->buf, c->buf_size, PUBACK,0, msg.id);
el if(msg.qos == QOS2)
len =MQTTSerialize_ack(c->buf, c->buf_size, PUBREC,0, msg.id);
if(len <=0)
rc = FAILURE;
el
rc =ndPacket(c, len, timer);
if(rc == FAILURE)
goto exit;// there was a problem
}
break;
}
ca PUBREC:
ca PUBREL:
{
unsigned short mypacketid;
unsigned char dup, type;
if(MQTTDerialize_ack(&type,&dup,&mypacketid, c->readbuf, c->readbuf_size)!=1)
rc = FAILURE;
el if((len =MQTTSerialize_ack(c->buf, c->buf_size,
(packet_type == PUBREC)? PUBREL : PUBCOMP,0, mypacketid))<=0)
rc = FAILURE;
el if((rc =ndPacket(c, len, timer))!= SUCCESS)// nd the PUBREL packet
rc = FAILURE;// there was a problem
if(rc == FAILURE)
goto exit;// there was a problem
break;
}
ca PUBCOMP:
break;
ca PINGRESP://收到服务器回复的ping数据包，对应下⾯的keepalive
printf("recv mqtt ping respon!\r\n");
c->ping_outstanding =0;
break;
}
//⼼跳处理函数
if(keepalive(c)!= SUCCESS){
//check only keepalive FAILURE status so that previous FAILURE status can be considered as FAULT rc = FAILURE;
牛牯嶂printf("keep alive fail\r\n");
}
exit://这⾥我做了修改
if(rc == SUCCESS)
rc = packet_type;
//el if (c->isconnected)
// MQTTCloSession(c);
return rc;
}
