5 MQTT 在“ read 0:1, break “后断开重连

static int net_read(MQTTClient *c, unsigned char *buf,  int len, int timeout)
{
    int bytes = 0;
    int rc;
    while (bytes < len)
    {
#ifdef MQTT_USING_TLS
        if (c->tls_session)
        {
            rc = mbedtls_client_read(c->tls_session, &buf[bytes], (size_t)(len - bytes));
            if (rc <= 0)
            {
                bytes = -1;
                break;
            }
            else
            {
                bytes += rc;
            }
            goto _continue;
        }
#endif
        rc = recv(c->sock, &buf[bytes], (size_t)(len - bytes), MSG_DONTWAIT);
        if (rc == -1)
        {
            if (errno != ENOTCONN && errno != ECONNRESET)
            {
                bytes = -1;
                break;
            }
        }
        else
            bytes += rc;
#ifdef MQTT_USING_TLS
_continue:
#endif
        if (bytes >= len)
        {
            break;
        }
        if (timeout > 0)
        {
            fd_set readset;
            struct timeval interval;
            LOG_D("net_read %d:%d, timeout:%d", bytes, len, timeout);
            timeout  = 0;
            interval.tv_sec = 1;
            interval.tv_usec = 0;
            FD_ZERO(&readset);
            FD_SET(c->sock, &readset);
            select(c->sock + 1, &readset, RT_NULL, RT_NULL, &interval);
        }
        else
        {
            LOG_D("net_read %d:%d, break!", bytes, len);
            break;
        }
    }
    return bytes;
}

static int MQTTPacket_readPacket(MQTTClient *c)
{
    int rc = PAHO_FAILURE;
    MQTTHeader header = {0};
    int len = 0;
    int rem_len = 0;
    /* 1. read the header byte.  This has the packet type in it */
    if (net_read(c, c->readbuf, 1, 0) != 1)
        goto exit;

typedef struct MQTTMessage
{
    enum QoS qos;
    unsigned char retained;
    unsigned char dup;
    unsigned short id;
    void *payload;
    size_t payloadlen;
} MQTTMessage;

int paho_mqtt_publish(MQTTClient *client, enum QoS qos, const char *topic, const char *msg_str)
{
    static unsigned short id = 0;
    MQTTMessage message = {0};
    if (qos != QOS1)
    {
        LOG_E("Not support Qos(%d) config, only support Qos(d).", qos, QOS1);
        return PAHO_FAILURE;
    }
    if (++id == 0)
    {
        ++id;
    }
    message.qos = qos;
    message.retained = 0;
    message.id = id;
    message.payload = (void *)msg_str;
    message.payloadlen = rt_strlen(message.payload);
    if (client->isblocking && client->pub_sem)
    {
        if(rt_sem_take(client->pub_sem, 5 * RT_TICK_PER_SECOND) < 0)
        {
            return PAHO_FAILURE;
        }
    }
    return MQTTPublish(client, topic, &message);
}