通過(guò)paho-mqtt軟件包入門rt-thread的sal

一、paho-mqtt軟件包程序流程

1.1 paho_mqtt_start
在rt_wlan_register_event_handler函數(shù)注冊(cè)好RT_WLAN_EVT_READY的回調(diào)函數(shù)paho_mqtt_start，當(dāng)wifi準(zhǔn)備好后調(diào)用mq_start啟動(dòng)mqtt。在mq_start中，初始化MQTTClient結(jié)構(gòu)體，設(shè)置mqtt連接的參數(shù)：mqtt的uri、mqtt的用戶名（username）和密碼（password）、mqtt發(fā)布和訂閱的主題Topic、消息質(zhì)量等級(jí)QoS，最后調(diào)用paho_mqtt_start創(chuàng)建處理mqtt的線程paho_mqtt_thread。

static void mq_start(void)
{
/* init condata param by using MQTTPacket_connectData_initializer /
MQTTPacket_connectData condata = MQTTPacket_connectData_initializer;
static char cid[20] = { 0 };
static int is_started = 0;
if (is_started)
{
return;
}
/ config MQTT context param /
{
client.isconnected = 0;
client.uri = MQTT_URI;
/ generate the random client ID /
rt_snprintf(cid, sizeof(cid), "rtthread%d", rt_tick_get());
/ config connect param /
memcpy(&client.condata, &condata, sizeof(condata));
client.condata.clientID.cstring = cid;
client.condata.keepAliveInterval = 60;
client.condata.cleansession = 1;
client.condata.username.cstring = MQTT_USERNAME;
client.condata.password.cstring = MQTT_PASSWORD;
/ config MQTT will param. /
client.condata.willFlag = 1;
client.condata.will.qos = 1;
client.condata.will.retained = 0;
client.condata.will.topicName.cstring = MQTT_PUBTOPIC;
client.condata.will.message.cstring = MQTT_WILLMSG;
/ malloc buffer. /
client.buf_size = client.readbuf_size = 1024;
client.buf = malloc(client.buf_size);
client.readbuf = malloc(client.readbuf_size);
if (!(client.buf && client.readbuf))
{
LOG_E("no memory for MQTT client buffer!");
goto _exit;
}
/ set event callback function /
client.connect_callback = mqtt_connect_callback;
client.online_callback = mqtt_online_callback;
client.offline_callback = mqtt_offline_callback;
/ set subscribe table and event callback /
client.messageHandlers[0].topicFilter = MQTT_SUBTOPIC;
client.messageHandlers[0].callback = mqtt_sub_callback;
client.messageHandlers[0].qos = QOS1;
/ set default subscribe event callback /
client.defaultMessageHandler = mqtt_sub_default_callback;
}
/ run mqtt client /
paho_mqtt_start(&client);
is_started = 1;
_exit:
return;
}
rt_wlan_register_event_handler(RT_WLAN_EVT_READY, (void ( )(int, struct rt_wlan_buff *, void *))mq_start, RT_NULL);

1.2 paho_mqtt_thread
在paho_mqtt_thread中調(diào)用paho-mqtt提供的接口和rt-thread的sal的接口完成與mqtt服務(wù)器的交互，包括以下幾個(gè)方面：與服務(wù)器的連接、訂閱主題、向服務(wù)器發(fā)送心跳包、處理服務(wù)器發(fā)送下來(lái)的消息（CONNACK、PUBACK、SUBACK、PUBLISH、PUBREC、PUBCOMP、PINGRESP）、回環(huán)服務(wù)器通過(guò)topic發(fā)送下來(lái)的消息。

static void paho_mqtt_thread(void *param)
{
MQTTClient *c = (MQTTClient )param;
int i, rc, len;
int rc_t = 0;
c->pub_sock = socket(AF_INET, SOCK_DGRAM, 0);
if (c->pub_sock == -1)
{
debug_printf("create pub_sock error!n");
goto _mqtt_exit;
}
/ bind publish socket. */
{
struct sockaddr_in pub_server_addr;
c->pub_port = pub_port;
pub_port ++;
pub_server_addr.sin_family = AF_INET;
pub_server_addr.sin_port = htons((c->pub_port));
pub_server_addr.sin_addr.s_addr = INADDR_ANY;
memset(&(pub_server_addr.sin_zero), 0, sizeof(pub_server_addr.sin_zero));
rc = bind(c->pub_sock, (struct sockaddr *)&pub_server_addr, sizeof(struct sockaddr));
if (rc == -1)
{
debug_printf("pub_sock bind error!n");
goto _mqtt_exit;
}
}
_mqtt_start:
if (c->connect_callback)
{
c->connect_callback(c);
}
rc = net_connect(c);
if (rc != 0)
{
goto _mqtt_restart;
}
rc = MQTTConnect(c);
if (rc != 0)
{
goto _mqtt_restart;
}
for (i = 0; i < MAX_MESSAGE_HANDLERS; i++)
{
const char topic = c->messageHandlers[i].topicFilter;
if(topic == RT_NULL)
continue;
rc = MQTTSubscribe(c, topic, QOS2);
debug_printf("Subscribe #%d %s %s!n", i, topic, (rc < 0) ? ("fail") : ("OK"));
if (rc != 0)
{
goto _mqtt_disconnect;
}
}
if (c->online_callback)
{
c->online_callback(c);
}
c->tick_ping = rt_tick_get();
while (1)
{
int res;
rt_tick_t tick_now;
fd_set readset;
struct timeval timeout;
tick_now = rt_tick_get();
if (((tick_now - c->tick_ping) / RT_TICK_PER_SECOND) > (c->keepAliveInterval - 5))
{
timeout.tv_sec = 1;
//debug_printf("tick close to ping.n");
}
else
{
timeout.tv_sec = c->keepAliveInterval - 10 - (tick_now - c->tick_ping) / RT_TICK_PER_SECOND;
//debug_printf("timeount for ping: %dn", timeout.tv_sec);
}
timeout.tv_usec = 0;
FD_ZERO(&readset);
FD_SET(c->sock, &readset);
FD_SET(c->pub_sock, &readset);
/ int select(maxfdp1, readset, writeset, exceptset, timeout); /
res = select(((c->pub_sock > c->sock) ? c->pub_sock : c->sock) + 1,
&readset, RT_NULL, RT_NULL, &timeout);
if (res == 0)
{
len = MQTTSerialize_pingreq(c->buf, c->buf_size);
rc = sendPacket(c, len);
if (rc != 0)
{
debug_printf("[%d] send ping rc: %d n", rt_tick_get(), rc);
goto _mqtt_disconnect;
}
/ wait Ping Response. /
timeout.tv_sec = 5;
timeout.tv_usec = 0;
FD_ZERO(&readset);
FD_SET(c->sock, &readset);
res = select(c->sock + 1, &readset, RT_NULL, RT_NULL, &timeout);
if (res <= 0)
{
debug_printf("[%d] wait Ping Response res: %dn", rt_tick_get(), res);
goto _mqtt_disconnect;
}
} / res == 0: timeount for ping. */
if (res < 0)
{
debug_printf("select res: %dn", res);
goto _mqtt_disconnect;
}
if (FD_ISSET(c->sock, &readset))
{
//debug_printf("sock FD_ISSETn");
rc_t = MQTT_cycle(c);
//debug_printf("sock FD_ISSET rc_t : %dn", rc_t);
if (rc_t < 0) goto _mqtt_disconnect;
continue;
}
if (FD_ISSET(c->pub_sock, &readset))
{
struct sockaddr_in pub_client_addr;
uint32_t addr_len = sizeof(struct sockaddr);
MQTTMessage *message;
MQTTString topic = MQTTString_initializer;
//debug_printf("pub_sock FD_ISSETn");
len = recvfrom(c->pub_sock, c->readbuf, c->readbuf_size, MSG_DONTWAIT,
(struct sockaddr *)&pub_client_addr, &addr_len);
if (pub_client_addr.sin_addr.s_addr != *((uint32_t )(&netif_default->ip_addr)))
{
#if 1
char client_ip_str[16]; / ###.###.###.### */
strcpy(client_ip_str,
inet_ntoa(*((struct in_addr *) & (pub_client_addr.sin_addr))));
debug_printf("pub_sock recvfrom len: %s, skip!n", client_ip_str);
#endif
continue;
}
if (len < sizeof(MQTTMessage))
{
c->readbuf[len] = '?';
debug_printf("pub_sock recv %d byte: %sn", len, c->readbuf);
if (strcmp((const char *)c->readbuf, "DISCONNECT") == 0)
{
debug_printf("DISCONNECTn");
goto _mqtt_disconnect_exit;
}
continue;
}
message = (MQTTMessage *)c->readbuf;
message->payload = c->readbuf + sizeof(MQTTMessage);
topic.cstring = (char *)c->readbuf + sizeof(MQTTMessage) + message->payloadlen;
//debug_printf("pub_sock topic:%s, payloadlen:%dn", topic.cstring, message->payloadlen);
len = MQTTSerialize_publish(c->buf, c->buf_size, 0, message->qos, message->retained, message->id,
topic, (unsigned char *)message->payload, message->payloadlen);
if (len <= 0)
{
debug_printf("MQTTSerialize_publish len: %dn", len);
goto _mqtt_disconnect;
}
if ((rc = sendPacket(c, len)) != PAHO_SUCCESS) // send the subscribe packet
{
debug_printf("MQTTSerialize_publish sendPacket rc: %dn", rc);
goto _mqtt_disconnect;
}
} /* pbulish sock handler. */
} /* while (1) */
_mqtt_disconnect:
MQTTDisconnect(c);
_mqtt_restart:
if (c->offline_callback)
{
c->offline_callback(c);
}
net_disconnect(c);
rt_thread_delay(RT_TICK_PER_SECOND * 5);
debug_printf("restart!n");
goto _mqtt_start;
_mqtt_disconnect_exit:
MQTTDisconnect(c);
net_disconnect(c);
_mqtt_exit:
debug_printf("thread exitn");
return;
}

二、與mqtt broker的交互
paho-mqtt軟件包提供了兩種發(fā)布消息到mqtt broker的方式：udp和管道。在MQTTClient結(jié)構(gòu)體中有三個(gè)成員與通信有關(guān)：sock、pub_sock、pub_pipe，其中sock是與mqtt broker通信的套接字，pub_sock和pub_pipe是兩種不同的發(fā)布方式：pub_sock是通過(guò)udp的方式發(fā)布消息；pub_pipe是通過(guò)管道，最終由sock發(fā)布消息。如下面的代碼所示，使用哪種方式可以通過(guò)宏來(lái)配置。下面展開(kāi)描述這兩種方式如何與mqtt broker交互的。

/* publish interface */

#if defined(RT_USING_POSIX) && (defined(RT_USING_DFS_NET) || defined(SAL_USING_POSIX))
int pub_pipe[2];
#else
int pub_sock;
int pub_port;
#endif

2.1 管道（pipe）方式
在paho_mqtt_pipe.c中的paho_mqtt_thread，下面的代碼完成了發(fā)布消息、接收訂閱消息、處理心跳包的工作。下面以三個(gè)點(diǎn)細(xì)說(shuō)。

當(dāng)需要發(fā)布消息時(shí)，應(yīng)用層需要調(diào)用MQTTPublish，這個(gè)函數(shù)會(huì)調(diào)用write向管道的寫端pub_pipe[1]寫入待發(fā)送的數(shù)據(jù)。而管道的讀端pub_pipe[0]在select中被監(jiān)聽(tīng)，當(dāng)MQTTPublish被調(diào)用時(shí)，select可以往下執(zhí)行，首先調(diào)用read從管道中讀取數(shù)據(jù)，接著調(diào)用MQTTSerialize_publish將數(shù)據(jù)封包，最后調(diào)用sendPacket將數(shù)據(jù)發(fā)送出去。

當(dāng)接收到訂閱的消息時(shí)，select會(huì)往下執(zhí)行，接著調(diào)用MQTT_cycle讀取并解析出數(shù)據(jù)。
select的超時(shí)時(shí)間是50s，如果50s沒(méi)有消息處理，則向broker發(fā)送心跳包。

FD_ZERO(&readset);
FD_SET(c->sock, &readset);
FD_SET(c->pub_pipe[0], &readset);
/* int select(maxfdp1, readset, writeset, exceptset, timeout); /
res = select(((c->pub_pipe[0] > c->sock) ? c->pub_pipe[0] : c->sock) + 1,
&readset, RT_NULL, RT_NULL, &timeout);
if (res == 0)
{
len = MQTTSerialize_pingreq(c->buf, c->buf_size);
rc = sendPacket(c, len);
if (rc != 0)
{
LOG_E("[%d] send ping rc: %d ", rt_tick_get(), rc);
goto _mqtt_disconnect;
}
/ wait Ping Response. /
timeout.tv_sec = 5;
timeout.tv_usec = 0;
FD_ZERO(&readset);
FD_SET(c->sock, &readset);
res = select(c->sock + 1, &readset, RT_NULL, RT_NULL, &timeout);
if (res <= 0)
{
LOG_E("[%d] wait Ping Response res: %d", rt_tick_get(), res);
goto _mqtt_disconnect;
}
} / res == 0: timeount for ping. */
if (res < 0)
{
LOG_E("select res: %d", res);
goto _mqtt_disconnect;
}
if (FD_ISSET(c->sock, &readset))
{
//LOG_D("sock FD_ISSET");
rc_t = MQTT_cycle(c);
//LOG_D("sock FD_ISSET rc_t : %d", rc_t);
if (rc_t < 0) goto _mqtt_disconnect;
continue;
}
if (FD_ISSET(c->pub_pipe[0], &readset))
{
MQTTMessage *message;
MQTTString topic = MQTTString_initializer;
//LOG_D("pub_sock FD_ISSET");
len = read(c->pub_pipe[0], c->readbuf, c->readbuf_size);
if (len < sizeof(MQTTMessage))
{
c->readbuf[len] = '?';
LOG_D("pub_sock recv %d byte: %s", len, c->readbuf);
if (strcmp((const char *)c->readbuf, "DISCONNECT") == 0)
{
LOG_D("DISCONNECT");
goto _mqtt_disconnect_exit;
}
continue;
}
message = (MQTTMessage *)c->readbuf;
message->payload = c->readbuf + sizeof(MQTTMessage);
topic.cstring = (char *)c->readbuf + sizeof(MQTTMessage) + message->payloadlen;
//LOG_D("pub_sock topic:%s, payloadlen:%d", topic.cstring, message->payloadlen);
len = MQTTSerialize_publish(c->buf, c->buf_size, 0, message->qos, message->retained, message->id,
topic, (unsigned char *)message->payload, message->payloadlen);
if (len <= 0)
{
LOG_D("MQTTSerialize_publish len: %d", len);
goto _mqtt_disconnect;
}
if ((rc = sendPacket(c, len)) != PAHO_SUCCESS) // send the subscribe packet
{
LOG_D("MQTTSerialize_publish sendPacket rc: %d", rc);
goto _mqtt_disconnect;
}
}

2.2 udp方式
udp方式中，處理流程與管道方式基本相似。下面說(shuō)明一下這種方式兩個(gè)套接字的工作流程。
MQTTClient結(jié)構(gòu)體中有兩個(gè)socket，一個(gè)是基于tcp的負(fù)責(zé)控制與服務(wù)器連接的sock，另一個(gè)是基于udp協(xié)議的負(fù)責(zé)消息發(fā)布的pub_sock。

2.2.1 sock
連接：在net_connect調(diào)用socket、connet函數(shù)建立與服務(wù)器的tcp連接。
處理：sock接收到服務(wù)器的數(shù)據(jù)后，在MQTT_cycle中處理來(lái)自服務(wù)器的CONNACK、PUBACK、SUBACK、PUBLISH、PUBREC、PUBCOMP、PINGRESP消息。
斷開(kāi)連接：在net_disconnect函數(shù)中調(diào)用closesocket關(guān)閉與服務(wù)器的tcp連接。

2.2.2 pub_sock
連接：分為pub_sock的綁定和mqtt連接的建立
1、調(diào)用socket創(chuàng)建pub_sock，之后調(diào)用bind綁定pub_sock到udp端口。
2、在MQTTConnect函數(shù)中，通過(guò)sock發(fā)送connect消息給服務(wù)器，建立mqtt連接。
處理：先recvfrom將接受的數(shù)據(jù)拷貝到MQTTClient的readbuf，再將數(shù)據(jù)回環(huán)發(fā)布到服務(wù)器。
斷開(kāi)連接：通過(guò)sock向服務(wù)器發(fā)送DISCONNECT消息，斷開(kāi)mqtt連接。