物联网核心之MQTT移植

蓝凌风

    在上一篇文章中，只是讲了MQTT的主要内容，至于怎么移植到STM32上，怎么使用才是最重要的关键。这里使用的平台是RT8711的WIFI SOC，使用的LWIP跟FreeRTOS，移植使用跟STM32+LWIP是没什么区别的。

   先在Github上找到Eclipse的开源MQTT客户端程序https://github.com/eclipse/paho.mqtt.embedded-c.git，并把源码下载起来

   解压源码，再进入MQTTPacket文件夹，里面有三个文件夹

   把src里面的所有文件和samples下的transport.c、transport.h两个文件复制到工程目录下。这里我们主要的移植工作就在transport里面。打开transport.c文件，这个是MQTT连接，发送，接收的接口，源码是Linux跟Windows平台，用的标准的Socket接口函数，我们这里的移植工作量很小，因为LWIP也是支持标准的Socket接口函数，只不过里面有些函数接口是LWIP不支持的，主要就是transport_open这个连接函数有区别。把原来的transport_open函数注释掉，重新写一个。

1. int transport_open(char* addr, int port)
   
2. {
   
3.         int* sock = &mysock;
   
4.         struct hostent *server;
   
5.     struct sockaddr_in serv_addr;
   
6.         static struct timeval tv;
   
7.         int timeout = 1000;
   
8.         fd_set readset;
   
9.         fd_set writeset;
   
10. 
    
11.     *sock = socket(AF_INET, SOCK_STREAM, 0);
    
12.     if(*sock < 0)
    
13.         DiagPrintf("[ERROR] Create socket failed\n");
    
14.    
    
15.     server = gethostbyname(addr);
    
16.     if(server == NULL)
    
17.         DiagPrintf("[ERROR] Get host ip failed\n");
    
18.    
    
19.     memset(&serv_addr,0,sizeof(serv_addr));
    
20.     serv_addr.sin_family = AF_INET;
    
21.     serv_addr.sin_port = htons(port);
    
22.     memcpy(&serv_addr.sin_addr.s_addr,server->h_addr,server->h_length);
    
23.    
    
24.     if (connect(*sock,(struct sockaddr *)&serv_addr,sizeof(serv_addr)) < 0){
    
25.         DiagPrintf("[ERROR] connect failed\n");
    
26.         return -1;
    
27.         }
    
28.         tv.tv_sec = 10;  /* 1 second Timeout */
    
29.         tv.tv_usec = 0;  
    
30.         setsockopt(mysock, SOL_SOCKET, SO_RCVTIMEO, (char *)&timeout,sizeof(timeout));
    
31.         return mysock;
    
32. }

复制代码

   到这里其实移植工作就已经完成了，就是这么的简单，剩下就是怎么使用MQTT。直接上代码，这里用FreeRTOS新建一个MQTT的任务。

1. #define THREAD_STACK_SIZE 512
   
2. #define HOST_NAME "m2m.eclipse.org"
   
3. #define HOST_PORT 1883
   
4. 
   
5. void mqtt_thread( void *arg)
   
6. {
   
7.         MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
   
8.         MQTTString receivedTopic;
   
9.         int rc = 0;
   
10.         char buf[200];
    
11.         int buflen = sizeof(buf);
    
12.         int mysock = 0;
    
13.         MQTTString topicString = MQTTString_initializer;
    
14.         int payloadlen_in;
    
15.         unsigned char* payload_in;
    
16.         unsigned short msgid = 1;
    
17.         int subcount;
    
18.         int granted_qos =0;
    
19.         unsigned char sessionPresent, connack_rc;
    
20.         unsigned short submsgid;
    
21.         int len = 0;
    
22.         int req_qos = 1;
    
23.         unsigned char dup;
    
24.         int qos;
    
25.         unsigned char retained;
    
26. 
    
27.         char *host = "m2m.eclipse.org";
    
28.         int port = 1883;
    
29.         uint8_t  msgtypes = CONNECT;
    
30.         uint32_t curtick = xTaskGetTickCount();
    
31.         log_info("socket connect to server");
    
32.         mysock = transport_open(host,port);
    
33.         if(mysock < 0)
    
34.                 return mysock;
    
35.         log_notice("Sending to hostname %s port %d\n", host, port);
    
36.         data.clientID.cstring = "me";
    
37.         data.keepAliveInterval = 50;
    
38.         data.cleansession = 1;
    
39.         data.username.cstring = "";
    
40.         data.password.cstring = "";
    
41.         data.MQTTVersion = 4;
    
42.         while(1)
    
43.         {
    
44.                 if((xTaskGetTickCount() - curtick) >(data.keepAliveInterval/2*1000))
    
45.                 {
    
46.                         if(msgtypes == 0)
    
47.                         {
    
48.                                 curtick = xTaskGetTickCount();
    
49.                                 msgtypes = PINGREQ;
    
50.                         }
    
51.                 }
    
52. 
    
53.                 switch(msgtypes)
    
54.                 {
    
55.                         case CONNECT:        len = MQTTSerialize_connect(buf, buflen, &data);       
    
56.                                                         rc = transport_sendPacketBuffer(mysock, (unsigned char*)buf, len);
    
57.                                                         if (rc == len)
    
58.                                                                 log_info("send CONNECT Successfully");
    
59.                                                         else
    
60.                                                                 log_debug("send CONNECT failed");               
    
61.                                                         log_info("MQTT concet to server!");
    
62.                                                         msgtypes = 0;
    
63.                                                         break;
    
64.                         case CONNACK:         if (MQTTDeserialize_connack(&sessionPresent, &connack_rc, buf, buflen) != 1 || connack_rc != 0)
    
65.                                                         {
    
66.                                                                 log_info("Unable to connect, return code %d\n", connack_rc);
    
67.                                                         }
    
68.                                                         else log_info("MQTT is concet OK!");
    
69.                                                         msgtypes = SUBSCRIBE;
    
70.                                                         break;
    
71. 
    
72.                         case SUBSCRIBE:        topicString.cstring = "ledtest";
    
73.                                                         len = MQTTSerialize_subscribe(buf, buflen, 0, msgid, 1, &topicString, &req_qos);
    
74.                                                         rc = transport_sendPacketBuffer(mysock, (unsigned char*)buf, len);
    
75.                                                         if (rc == len)
    
76.                                                                 log_info("send SUBSCRIBE Successfully\n");
    
77.                                                         else
    
78.                                                                 log_debug("send SUBSCRIBE failed\n");       
    
79.                                                         log_info("client subscribe:[%s]",topicString.cstring);
    
80.                                                         msgtypes = 0;
    
81.                                                         break;
    
82.                         case SUBACK:        rc = MQTTDeserialize_suback(&submsgid, 1, &subcount, &granted_qos, buf, buflen);                                                       
    
83.                                                         log_info("granted qos is %d\n", granted_qos);                                                               
    
84.                                                         msgtypes = 0;
    
85.                                                         break;
    
86.                         case PUBLISH:        rc = MQTTDeserialize_publish(&dup, &qos, &retained, &msgid, &receivedTopic,        &payload_in, &payloadlen_in, buf, buflen);
    
87.                                                         log_info("message arrived : %s\n", payload_in);
    
88.                                                         if(strstr(payload_in,"on"))
    
89.                                                         {
    
90.                                                                 log_notice("LED on!!");
    
91.                                                         }
    
92.                                                         else if(strstr(payload_in,"off"))
    
93.                                                         {
    
94.                                                                 log_notice("LED off!!");
    
95.                                                         }
    
96.                                                         if(qos == 1)
    
97.                                                         {
    
98.                                                                 log_info("publish qos is 1,send publish ack.");
    
99.                                                                 memset(buf,0,buflen);
    
100.                                                                 len = MQTTSerialize_ack(buf,buflen,PUBACK,dup,msgid);   //publish ack                       
     
101.                                                                 rc = transport_sendPacketBuffer(mysock, (unsigned char*)buf, len);
     
102.                                                                 if (rc == len)
     
103.                                                                         log_info("send PUBACK Successfully");
     
104.                                                                 else
     
105.                                                                         log_debug("send PUBACK failed");                                       
     
106.                                                         }
     
107.                                                         msgtypes = 0;
     
108.                                                         break;
     
109. 
     
110.                         case PUBACK:        log_info("PUBACK!");
     
111.                                                         msgtypes = 0;
     
112.                                                         break;
     
113.                         case PUBREC:        log_info("PUBREC!");     //just for qos2
     
114.                                                         break;
     
115.                         case PUBREL:        log_info("PUBREL!");        //just for qos2
     
116.                                                         break;
     
117.                         case PUBCOMP:        log_info("PUBCOMP!");        //just for qos2
     
118.                                                         break;
     
119.                         case PINGREQ:        len = MQTTSerialize_pingreq(buf, buflen);
     
120.                                                         rc = transport_sendPacketBuffer(mysock, (unsigned char*)buf, len);
     
121.                                                         if (rc == len)
     
122.                                                                 log_info("send PINGREQ Successfully\n");
     
123.                                                         else
     
124.                                                                 log_debug("send PINGREQ failed\n");       
     
125.                                                         log_info("time to ping mqtt server to take alive!");
     
126.                                                         msgtypes = 0;
     
127.                                                         break;
     
128.                         case PINGRESP:        log_info("mqtt server Pong");                                                       
     
129.                                                         msgtypes = 0;
     
130.                                                         break;
     
131.                 }
     
132.                                 memset(buf,0,buflen);
     
133.                                 rc=MQTTPacket_read(buf, buflen, transport_getdata);       
     
134.                                 if(rc >0)
     
135.                                 {
     
136.                                         msgtypes = rc;
     
137.                                         log_info("MQTT is get recv:");
     
138.                                 }
     
139.                                 gpio_write(&gpio_led, !gpio_read(&gpio_led));
     
140.         }
     
141. 
     
142. exit:
     
143.         transport_close(mysock);
     
144.     log_info("mqtt thread exit.");
     
145.     vTaskDelete(NULL);
     
146. }

复制代码

       MQTT服务器仍然是上篇文章用的m2m.eclipse.org，一开始就是调用 前面移植的transport_open(host,port)去连接MQTT服务器，并返回套接字。然后就是前面说的登录、订阅、发布、心跳等操作，这里在While里用状态来实现整个连接。

首先就是CONNECT登录，MQTTPacket_connectData data = MQTTPacket_connectData_initializer初始化登录数据结构体，然后再对data进行初始化，data.clientID.cstring = "me";data.keepAliveInterval = 50;data.cleansession = 1;

data.username.cstring = "";data.password.cstring = "";data.MQTTVersion = 4;这里表示cilentID为 me,心跳时间为50，用户名跟密码都为空，结构初始化后，就是要对数据进去打包，调用MQTTSerialize_connect函数。打包后就是调用transport_sendPacketBuffer发送数据包。数据发送完后是调用MQTTPacket_read去接收服务器返回来的数据，并得到返回的数据包类型，根据数据包类型进行不同的逻辑处理。发送CONNECT后服务器对相应的返回CONNACK数据包表示登录成功。

登录成功后，开始订阅我们想要Topics，这里订阅一个”ledtest”的Tipics。数据包的初始化、打包、发送跟登录是相似的，就不详述，直接看代码就可以了，服务器会相应的返回SUBACK。

    接下来是心跳，通过获取系统的Tick来判断是否要发PINGREQ，如果的话，让状态机状态为PINGREQ后去发送心跳包，相应服务器会返回PINGRESP。

    最后是PUBLISH，这是MQTT的最主要的通信协议，这里我只实现的客户端接收PUBLISH，其它的使用，其实都可以在源码的MQTTPacket里面sample可以找到例程。PUBLISH实现也很简单，定阅了Topics之后，只要其它客户端向这个Topics推送数据，服务器就会转发到订阅者上。MQTTPacket_read接收到服务器的推送，解析成PUBLISH数据包，然后状态机改变状态到PUBLISH再调用MQTTDeserialize_publish进行解包，得到推送的内容，最后根据推送的内容执行相应的动作，就实现远程控制。

     整个代码的效果如下：

    可以看到MQTT的登录，订阅，还有在PC上通过MQTT.fx推送信息给ledtest，RT8711上收到推送的内容，并执行打开LED的动作。

    也许有人会问，如果STM32或者其它单片机是用WIFI模块或者GPRS模块，没有LWIP的怎么办。其实只要理解的MQTT的源码，就不难用GPRS或者WiFi模块去实现。MQTT的源码里都是对协议包进行打包解包，数据传输都是在tranport.c里面，我们完全不用transport，可以自己写通信接口，然后把打包的数据包通过模块发出去，写接收接口，把模块接收到服务器数据调用MQTT解包接口解析就可以了。

6lxuv4n4sshdb

好贴，值得分享,

h8jtd1r95w

goldmusic

值得一读，谢谢,