MQTT历程中pub_sock用途是什么？

发布于 2018-03-21 21:48:49 浏览：2100 订阅该版

[tocm]

* 本帖最后由 sun_shine 于 2018-3-21 21:51 编辑 *

mqtt例程建立了两个socket连接，一个是c->sock,另一个是c->pubsock，感觉作用是一样的的都可以向服务器publish，请问pub_sock还有其他什么作用？大神指点一下。下面附上代码 res = ;
 
 ``` 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 ", 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: %d ", rt_tick_get(), res);
 goto _mqtt_disconnect;
 }
 } /* res == 0: timeount for ping. */

if (res < 0)
 {
 debug_printf("select res: %d ", res);
 goto _mqtt_disconnect;
 }

if (FD_ISSET(c->sock, &readset))
 {
 //debug_printf("sock FD_ISSET ");
 rc_t = MQTT_cycle(c);
 //debug_printf("sock FD_ISSET rc_t : %d ", 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_ISSET ");

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! ", client_ip_str);
#endif
 continue;
 }

if (len < sizeof(MQTTMessage))
 {
 c->readbuf[len] = '\0';
 debug_printf("pub_sock recv %d byte: %s ", len, c->readbuf);

if (strcmp((const char *)c->readbuf, "DISCONNECT") == 0)
 {
 debug_printf("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;
 //debug_printf("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)
 {
 debug_printf("MQTTSerialize_publish len: %d ", len);
 goto _mqtt_disconnect;
 }

if ((rc = sendPacket(c, len)) != PAHO_SUCCESS) // send the subscribe packet
 {
 debug_printf("MQTTSerialize_publish sendPacket rc: %d ", rc);
 goto _mqtt_disconnect;
 }```