#include <errno.h>
#include <fcntl.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/resource.h>
#include <sys/stat.h>
#include <sys/timerfd.h>
#include <sys/types.h>
#include <syslog.h>
#include <sys/syscall.h>
#include <unistd.h>
#include <sys/time.h>

#include "lnxall_list.h"
#include "mqtt_session.h"

extern void *mqtt_recv_loop(void *param);
void my_message_callback(struct mosquitto *mosq, void *obj, const struct mosquitto_message *message);
void my_connect_callback(struct mosquitto *mosq, void *obj, int result);
void my_disconnect_callback(struct mosquitto *mosq, void *obj, int rc);
void my_publish_callback(struct mosquitto *mosq, void *obj, int mid);
static int lnxall_condvar_init(void * _cond)
{
    int ret;
    pthread_cond_t * condp;
    pthread_condattr_t attr;

    ret = 0;
    condp = (pthread_cond_t *) _cond;
    if (condp == NULL)
        return -1;

    memset(&attr, 0, sizeof(attr));
    ret = pthread_condattr_init(&attr);
    if (ret != 0)
        goto err0;
    ret = pthread_condattr_setclock(&attr, CLOCK_MONOTONIC);
    if (ret != 0)
        goto err0;

    memset(condp, 0, sizeof(*condp));
    ret = pthread_cond_init(condp, &attr);
    pthread_condattr_destroy(&attr);
    if (ret == 0)
        return 0;

err0:
    syslog(LOG_ERR | LOG_USER, "Error in [%s], failed to initialize condvar: %d",
        __func__, ret);
    return -1;
}

static int lnxall_condvar_timedwait(void * _condp, void * _mutexp, unsigned int inval)
{
    int ret;
    struct timespec spec;
    pthread_cond_t * condp;
    pthread_mutex_t * mutexp;

    condp = (pthread_cond_t *) _condp;
    mutexp = (pthread_mutex_t *) _mutexp;
    if (condp == NULL || mutexp == NULL)
        return -1;

    spec.tv_sec = 0;
    spec.tv_nsec = 0;
    ret = clock_gettime(CLOCK_MONOTONIC, &spec);
    if (ret == -1) {
        /* nearly impossible, just emit a message */
        syslog(LOG_ERR | LOG_USER, "Error, failed to system boot time for condition variable!");
        return -1;
    }

    spec.tv_sec += (time_t) (inval / 1000);
    spec.tv_nsec += (long) ((inval % 1000) * 1000000);
    if (spec.tv_nsec >= 1000000000) {
        spec.tv_sec += (time_t) (spec.tv_nsec / 1000000000);
        spec.tv_nsec %= 1000000000;
    }

    ret = pthread_cond_timedwait(condp, mutexp, &spec);
    if (ret != 0 && ret != ETIMEDOUT) {
        syslog(LOG_ERR | LOG_USER, "Error, failed to wait condvar: %d (%s)",
            ret, strerror(ret) ? : "");
    }
    return ret;
}

void *mqtt_run_loop(void *param) 
{ 
    mqtt_session_t *session = (mqtt_session_t *)param; 
    mqtt_client_t *client = &session->client; 
    struct mosquitto *mosq = client->mosq; 
    int rc; 
    int reconnect_delay = 2; 
    const int max_reconnect_delay = 60; 
    int connection_attempts = 0; 

    syslog(LOG_USER | LOG_ERR, "Starting MQTT run loop for client: %s, broker: %s:%d", 
           client->clientId, client->addr, client->port); 

    while (1) { 
        connection_attempts++; 
        syslog(LOG_USER | LOG_ERR, "Connection attempt #%d to %s:%d", 
               connection_attempts, client->addr, client->port); 

        if (connection_attempts == 1 || !mosq) { 
            if (session->client.tls_cafile) 
            { 
                syslog(LOG_USER | LOG_ERR, "Configuring TLS with cafile:%s, certfile:%s, keyfile:%s", 
                       session->client.tls_cafile, session->client.tls_certfile, session->client.tls_keyfile); 
                rc = mosquitto_tls_set(mosq, session->client.tls_cafile, NULL, 
                                      session->client.tls_certfile, session->client.tls_keyfile, NULL); 
                if (rc != MOSQ_ERR_SUCCESS) { 
                    syslog(LOG_USER | LOG_ERR, "Failed to set TLS parameters: %s", mosquitto_strerror(rc)); 
                    sleep(reconnect_delay); 
                    reconnect_delay = (reconnect_delay * 2) > max_reconnect_delay ? max_reconnect_delay : (reconnect_delay * 2); 
                    continue; 
                } 
                mosquitto_tls_opts_set(mosq, 0, "tlsv1.2", NULL); // 0: 不验证服务器证书 
            } 
        } 

        syslog(LOG_USER | LOG_ERR, "MQTT connect parameters - clientId: %s, keepalive: %d", 
               client->clientId, client->keepalive); 

        rc = mosquitto_connect(mosq, client->addr, client->port, client->keepalive); 
        if (rc != MOSQ_ERR_SUCCESS) { 
            syslog(LOG_USER | LOG_ERR, "Failed to connect to broker: %s (rc=%d)", 
                   mosquitto_strerror(rc), rc); 
            sleep(reconnect_delay); 
            reconnect_delay = (reconnect_delay * 2) > max_reconnect_delay ? max_reconnect_delay : (reconnect_delay * 2); 
            continue; 
        } 

        syslog(LOG_USER | LOG_ERR, "Successfully connected to broker %s:%d", client->addr, client->port); 
        reconnect_delay = 2; 
        connection_attempts = 0; 

        // 连接成功后添加状态监控 
        syslog(LOG_USER | LOG_ERR, "Starting MQTT event loop..."); 
        time_t loop_start_time = time(NULL); 

        rc = mosquitto_loop_forever(mosq, -1, 1); 

        time_t loop_end_time = time(NULL); 
        syslog(LOG_USER | LOG_ERR, "MQTT event loop exited after %ld seconds with error: %d - %s", 
               loop_end_time - loop_start_time, rc, mosquitto_strerror(rc)); 

        switch(rc) { 
            case MOSQ_ERR_CONN_LOST: 
                syslog(LOG_USER | LOG_ERR, "Connection lost with broker"); 
                break; 
            case MOSQ_ERR_PROTOCOL: 
                syslog(LOG_USER | LOG_ERR, "Protocol error with broker - check MQTT version and message format"); 
                break; 
            case MOSQ_ERR_AUTH: 
                syslog(LOG_USER | LOG_ERR, "Authentication failed - check username/password"); 
                break; 
            default: 
                syslog(LOG_USER | LOG_ERR, "Unexpected error: %d", rc); 
        } 

    	session->state = MQTT_DISCONNECTED;
    	mosquitto_disconnect(mosq);
        syslog(LOG_USER | LOG_ERR, "Reinitializing MQTT client for reconnect..."); 

        mosquitto_destroy(mosq); 
        mosq = mosquitto_new(client->clientId, true, session); 
        if (!mosq) { 
            syslog(LOG_USER | LOG_ERR, "Failed to create new MQTT client"); 
            sleep(reconnect_delay); 
            reconnect_delay = (reconnect_delay * 2) > max_reconnect_delay ? max_reconnect_delay : (reconnect_delay * 2); 
            continue; 
        } 

        client->mosq = mosq; 
        session->client.mosq = mosq; 

        mosquitto_message_callback_set(mosq, my_message_callback); 
        mosquitto_connect_callback_set(mosq, my_connect_callback); 
        mosquitto_disconnect_callback_set(mosq, my_disconnect_callback); 
        if (client->qos > 0) { 
            mosquitto_publish_callback_set(mosq, my_publish_callback); 
        } 
        mqtt_session_set_callbacks(session, session->handle_recv_msg, session->connect_state_change); 

        if (strlen(client->user) > 0 || strlen(client->pass) > 0) { 
            rc = mosquitto_username_pw_set(mosq, 
                                          strlen(client->user) > 0 ? client->user : NULL, 
                                          strlen(client->pass) > 0 ? client->pass : NULL); 
            if (rc != MOSQ_ERR_SUCCESS) { 
                syslog(LOG_USER | LOG_ERR, "Failed to set username/password: %s", mosquitto_strerror(rc)); 
            } 
        } 

        syslog(LOG_USER | LOG_ERR, "Waiting %d seconds before reconnect attempt", reconnect_delay); 
        sleep(reconnect_delay); 
        reconnect_delay = (reconnect_delay * 2) > max_reconnect_delay ? max_reconnect_delay : (reconnect_delay * 2); 
    } 

    return NULL; 
}
/**
*******************************************************************************
*
* @brief  订阅topic_list中所有的topic，通常用于断线重连时进行调用
* @param session MQTT 会话结构体的指针
* @return 0 for success, -1 for error
*
*******************************************************************************
*/
int mqtt_session_subscribe_all(mqtt_session_t *session)
{
    topic_node_t *node = NULL;
    topic_node_t *tmp = NULL;
    mqtt_client_t *client = &session->client;
    struct mosquitto *mosq = client->mosq;

    if (session->state != MQTT_CONNECTED)
    {
        syslog(LOG_USER | LOG_WARNING, "MQTT client %s state is not connected", client->clientId);
        return -1;
    }

    pthread_rwlock_rdlock(&session->topic_lock);
    list_for_each_entry_safe(node, tmp, &session->topic_list, list)
    {
        mosquitto_subscribe(mosq, &node->tid, node->topic, client->qos);
    }
    pthread_rwlock_unlock(&session->topic_lock);

    return 0;
}

static bool isHighPrioirty(mqtt_message_t *msg)
{
    if (strstr(msg->topic, "_HPRI"))
    {
        return true;
    }
    return false;
}

void my_message_callback(struct mosquitto *mosq, void *obj, const struct mosquitto_message *message)
{
    mqtt_session_t *session = (mqtt_session_t *)obj;
    mqtt_message_t *mqtt_msg = NULL;
    int size;
    bool hp = false;

    if (session->received > MAX_RX_QUEUE_LENGTH)
    {
        syslog(LOG_USER | LOG_ERR, "queue is full: %d", session->received);
        return;
    }

    mqtt_msg = malloc(sizeof(mqtt_message_t));
    if (mqtt_msg == NULL)
    {
        syslog(LOG_USER | LOG_ERR, "malloc failed");
        return;
    }

    size = strlen(message->topic) + 1 + message->payloadlen;

    mqtt_msg->topic = calloc(1, size + 1);
    strcpy(mqtt_msg->topic, message->topic);
    mqtt_msg->payloadLen = message->payloadlen;
    mqtt_msg->payload = mqtt_msg->topic + strlen(message->topic) + 1;
    memcpy(mqtt_msg->payload, message->payload, message->payloadlen);

    hp = isHighPrioirty(mqtt_msg);

    pthread_mutex_lock(&session->recv_lock);
    if (hp)
    {
        list_add_head(&mqtt_msg->list, &session->rcv_list);
    }
    else
    {
        list_add_tail(&mqtt_msg->list, &session->rcv_list);
    }

    session->received++;
    pthread_mutex_unlock(&session->recv_lock);
    pthread_cond_signal(&session->recv_cond);
}

void my_connect_callback(struct mosquitto *mosq, void *obj, int result)
{
    mqtt_session_t *session = (mqtt_session_t *)obj;

    syslog(LOG_USER | LOG_DEBUG, "%s %sconnected",
        session->client.clientId, result ? "not " : "");

    if (session == NULL)
    {
        return;
    }

    if (result != 0) {
        session->state = MQTT_CONNECTING;
        return;
    }

    session->state = MQTT_CONNECTED;
    session->send_id = -1;
    mqtt_session_subscribe_all(session);
    if (session->connect_state_change)
    {
        session->connect_state_change(session->user_data, session->state);
    }
}

void my_subscribe_callback(struct mosquitto *mosq, void *obj, int mid, int qos_count, const int *granted_qos)
{
    mqtt_session_t *session = (mqtt_session_t *)obj;

    if (session == NULL)
    {
        return;
    }
    syslog(LOG_USER | LOG_DEBUG, "Subscribed mid:%d", mid);
}

void my_publish_callback(struct mosquitto *mosq, void *obj, int mid)
{
    mqtt_session_t *session = (mqtt_session_t *)obj;
    if (session->client.qos > 0 && mid == session->send_id)
    {
        pthread_mutex_lock(&session->pub_lock);
        pthread_cond_signal(&session->pub_cond);
        pthread_mutex_unlock(&session->pub_lock);
    }
}

void my_disconnect_callback(struct mosquitto *mosq, void *obj, int rc)
{
    mqtt_session_t *session = (mqtt_session_t *)obj;

    if (session == NULL)
    {
        return;
    }
    syslog(LOG_USER | LOG_DEBUG, "%s disconnect", session->client.clientId);

    if (session->state == MQTT_CONNECTED)
    {
        session->state = MQTT_CONNECTING;
    }
    if (session->connect_state_change)
    {
        session->connect_state_change(session->user_data, session->state);
    }
}

int mqtt_session_set_address(mqtt_session_t *session, char *addr, unsigned short port, char *user, char *pass)
{
    mqtt_client_t *client = NULL; // MQTT client

    client = &session->client;

    strncpy(client->addr, addr, sizeof(client->addr));
    client->port = port;
    if (user != NULL)
    {
        strncpy(client->user, user, sizeof(client->user));
    }
    else
    {
        client->user[0] = 0;
    }

    if (pass != NULL)
    {
        strncpy(client->pass, pass, sizeof(client->pass));
    }
    else
    {
        client->pass[0] = 0;
    }

    return 0;
}

int mqtt_session_set_tls(mqtt_session_t *session, char *tls_cafile, char *tls_certfile, char *tls_keyfile)
{
    mqtt_client_t *client = NULL; // MQTT client

    client = &session->client;

    if (strlen(tls_cafile) == 0)
        return -1;

    client->tls_cafile = strdup(tls_cafile);
    if (strlen(tls_certfile))
    {
        client->tls_certfile = strdup(tls_certfile);
    }
    else
    {
        client->tls_certfile = NULL;
    }

    if (strlen(tls_keyfile))
    {
        client->tls_keyfile = strdup(tls_keyfile);
    }
    else
    {
        client->tls_keyfile = NULL;
    }

    return 0;
}

int mqtt_session_set_opts(mqtt_session_t *session, int qos, int keepalive)
{
    mqtt_client_t *client = NULL; // MQTT client

    client = &session->client;
    client->qos = qos;
    client->keepalive = keepalive;

    return 0;
}

int mqtt_session_set_callbacks(mqtt_session_t *session, int (*on_message)(void *obj, mqtt_message_t *mqtt_msg),
                               int (*on_state_change)(void *obj, int state))
{
    session->handle_recv_msg = on_message;
    session->connect_state_change = on_state_change;

    return 0;
}

mqtt_session_t *mqtt_session_new(char *client_id, void *user_data)
{
    mqtt_session_t *session = calloc(1, sizeof(mqtt_session_t));

    if (session == NULL)
    {
        syslog(LOG_USER | LOG_ERR, "malloc error");
        return NULL;
    }

    if (client_id != NULL && strlen(client_id) > 0)
    {
        //strncpy(session->client.clientId, client_id, MAX_CLIENT_ID_LEN);
	snprintf(session->client.clientId, sizeof(session->client.clientId), "%s", client_id);
    }
    else
    {
        snprintf(session->client.clientId, MAX_CLIENT_ID_LEN, "TID_%ld", syscall(SYS_gettid));
    }

    if (user_data == NULL)
    {
        session->user_data = session;
    }
    else
    {
        session->user_data = user_data;
    }

    INIT_LIST_HEAD(&session->topic_list);
    INIT_LIST_HEAD(&session->rcv_list);
    pthread_mutex_init(&session->recv_lock, NULL);
    lnxall_condvar_init(&session->recv_cond);
    pthread_mutex_init(&session->pub_lock, NULL);
    lnxall_condvar_init(&session->pub_cond);
    pthread_rwlock_init(&session->topic_lock, NULL);
    return session;
}

/**
*******************************************************************************
*
* @brief  获取MQTT会话的client ID
* @param session MQTT 会话结构体的指针
* @return client_id for success; NULL for fail
*
*******************************************************************************
*/
char *mqtt_session_get_client_id(mqtt_session_t *session)
{
    if (session == NULL)
    {
        return NULL;
    }

    return session->client.clientId;
}

int mqtt_session_start(mqtt_session_t *session)
{
    struct mosquitto *mosq = NULL;
    mqtt_client_t *client = &session->client;
    pthread_t thread_mqtt_session;
    pthread_t thread_mqtt;
    char *user = NULL;
    char *pass = NULL;

    if (client->keepalive == 0)
    {
        client->keepalive = 60;
    }

    mosquitto_lib_init();
    mosq = mosquitto_new(client->clientId, true, session);
    if (!mosq)
    {
        switch (errno)
        {
        case ENOMEM:
            syslog(LOG_USER | LOG_ERR, "Error: Out of memory.");
            break;
        case EINVAL:
            syslog(LOG_USER | LOG_ERR, "Error: Invalid id.");
            break;
        }
        mosquitto_lib_cleanup();
        return -1;
    }
    if (session->client.tls_cafile)
    {
        mosquitto_tls_set(mosq, session->client.tls_cafile, NULL, session->client.tls_certfile, session->client.tls_keyfile, NULL);
        mosquitto_tls_opts_set(mosq, 0, "tlsv1.2", NULL); // 0: 不验证服务器证书
        syslog(LOG_USER | LOG_DEBUG, "enable mosquitto tls cafile:%s certfile:%s keyfile:%s ...", session->client.tls_cafile, session->client.tls_certfile, session->client.tls_keyfile);
    }

    if (strlen(client->user))
    {
        user = client->user;
    }

    if (strlen(client->pass))
    {
        pass = client->pass;
    }

    if (mosquitto_username_pw_set(mosq, user, pass))
    {
        syslog(LOG_USER | LOG_ERR, "Error: Problem setting username and password.\n");
        mosquitto_lib_cleanup();
        return -1;
    }
    client->mosq = mosq;

    mosquitto_message_callback_set(mosq, my_message_callback);
    mosquitto_connect_callback_set(mosq, my_connect_callback);
    mosquitto_disconnect_callback_set(mosq, my_disconnect_callback);
    if (client->qos > 0)
        mosquitto_publish_callback_set(mosq, my_publish_callback);
    // mosquitto_subscribe_callback_set(mosq, my_subscribe_callback);

    pthread_create(&thread_mqtt, NULL, mqtt_run_loop, (void *)session);
    pthread_create(&thread_mqtt_session, NULL, mqtt_recv_loop, (void *)session);
    return 0;
}

/**
*******************************************************************************
*
* @brief  订阅topic
* @param session MQTT 会话结构体的指针
* @param topic 要订阅的topic
* @return 0 for success; -1 for fail
*
*******************************************************************************
*/
int mqtt_session_subscribe(mqtt_session_t *session, const char *topic)
{
    topic_node_t *node = NULL;
    mqtt_client_t *client = &session->client;
    struct mosquitto *mosq = client->mosq;

    if (topic == NULL)
        return -1;
    node = malloc(sizeof(topic_node_t) + strlen(topic) + 1);
    if (node == NULL)
    {
        syslog(LOG_USER | LOG_ERR, "malloc failed, strlen(topic):%u", (unsigned int) strlen(topic));
        return -1;
    }

    syslog(LOG_USER | LOG_DEBUG, "client:%s sub topic:%s", session->client.clientId, topic);

    strcpy(node->topic, topic);
    pthread_rwlock_wrlock(&session->topic_lock);
    list_add_tail(&node->list, &session->topic_list);
    pthread_rwlock_unlock(&session->topic_lock);

    if (session->state == MQTT_CONNECTED)
    {
        mosquitto_subscribe(mosq, &node->tid, topic, client->qos);
    }
    return 0;
}

int mqtt_session_subscribe_qos(mqtt_session_t *session, const char *topic, int qos)
{
    topic_node_t *node = NULL;
    mqtt_client_t *client = &session->client;
    struct mosquitto *mosq = client->mosq;

    if (topic == NULL)
        return -1;
    node = malloc(sizeof(topic_node_t) + strlen(topic) + 1);
    if (node == NULL)
    {
        syslog(LOG_USER | LOG_ERR, "malloc failed, strlen(topic):%u", (unsigned int) strlen(topic));
        return -1;
    }

    syslog(LOG_USER | LOG_DEBUG, "client:%s sub topic:%s", session->client.clientId, topic);

    strcpy(node->topic, topic);
    pthread_rwlock_wrlock(&session->topic_lock);
    list_add_tail(&node->list, &session->topic_list);
    pthread_rwlock_unlock(&session->topic_lock);

    if (session->state == MQTT_CONNECTED)
    {
        mosquitto_subscribe(mosq, &node->tid, topic, qos);
    }
    return 0;
}

/**
*******************************************************************************
*
* @brief  取消订阅topic
* @param session MQTT 会话结构体的指针
* @param topic 要取消订阅的topic
* @return 0 for success; -1 for fail
*
*******************************************************************************
*/
int mqtt_session_unsubscribe(mqtt_session_t *session, char *topic)
{
    topic_node_t *node = NULL;
    topic_node_t *tmp = NULL;
    mqtt_client_t *client = &session->client;
    struct mosquitto *mosq = client->mosq;

    pthread_rwlock_wrlock(&session->topic_lock);
    list_for_each_entry_safe(node, tmp, &session->topic_list, list)
    {
        if (strcmp(topic, node->topic) == 0)
        {
            list_del(&node->list);
            free(node);
            break;
        }
    }
    pthread_rwlock_unlock(&session->topic_lock);
    mosquitto_unsubscribe(mosq, NULL, topic);
    return 0;
}

/**
*******************************************************************************
*
* @brief
     对于调用者来说，只要加入publish list了，就可以认为是发送成功了，
     因为MQTT会话层会保证已经加入list的msg会被发送出去，
     如果消息发送成功了，则从list移除该消息，
     如果发送失败将不会从list删除，等待下次发送时继续发送
     因此，对于上层调用者来说，只要插入list成功，就可以认为发送成功了
* @param session MQTT 会话结构体的指针
* @param topic 消息topic
* @param payload 消息体
* @param payloadlen 消息长度
* @return 0 for success
*
*******************************************************************************
*/
int mqtt_session_publish(mqtt_session_t *session, char *topic, const void *payload, int payloadlen)
{
    mqtt_client_t *client = &session->client;
    struct mosquitto *mosq = client->mosq;

    if (session == NULL || session->state != MQTT_CONNECTED)
    {
        // syslog(LOG_USER | LOG_ERR, "MQTT client %s is disconnected", session->client.clientId);
        return -1;
    }

    if (topic == NULL || strlen(topic) == 0)
    {
        syslog(LOG_USER | LOG_WARNING, "topic is empty");
        return -1;
    }

    if (payload == NULL || payloadlen == 0)
    {
        syslog(LOG_USER | LOG_WARNING, "payload is empty, topic:%s", topic);
    }

    return mosquitto_publish(mosq, NULL, topic, payloadlen, payload, client->qos, false);
}

int mqtt_session_publish_qos(mqtt_session_t *session, char *topic, const void *payload, int payloadlen, int qos)
{
    mqtt_client_t *client = &session->client;
    struct mosquitto *mosq = client->mosq;

    if (session == NULL || session->state != MQTT_CONNECTED)
    {
        // syslog(LOG_USER | LOG_ERR, "MQTT client %s is disconnected", session->client.clientId);
        return -1;
    }

    if (topic == NULL || strlen(topic) == 0)
    {
        syslog(LOG_USER | LOG_WARNING, "topic is empty");
        return -1;
    }

    if (payload == NULL || payloadlen == 0)
    {
        syslog(LOG_USER | LOG_WARNING, "payload is empty, topic:%s", topic);
    }

    return mosquitto_publish(mosq, NULL, topic, payloadlen, payload, qos, false);
}

/**
*******************************************************************************
*
* @brief 阻塞调用publish，当QoS > 0时只有调用到on_publish的回调才返回，默认超时时间为5秒
* @param session MQTT 会话结构体的指针
* @param topic 消息topic
* @param payload 消息体
* @param payloadlen 消息长度
* @return 0 for success
*
*******************************************************************************
*/
int mqtt_session_publish_sync(mqtt_session_t *session, char *topic, const void *payload, int payloadlen)
{
    mqtt_client_t *client = &session->client;
    struct mosquitto *mosq = client->mosq;

    if (session == NULL || session->state != MQTT_CONNECTED)
    {
        // syslog(LOG_USER | LOG_ERR, "MQTT client %s is disconnected", session->client.clientId);
        return -1;
    }

    if (topic == NULL || strlen(topic) == 0)
    {
        syslog(LOG_USER | LOG_WARNING, "topic is empty");
        return -1;
    }

    if (payload == NULL || payloadlen == 0)
    {
        syslog(LOG_USER | LOG_WARNING, "payload is empty, topic:%s", topic);
    }

    if (client->qos > 0)
    {
        int retry = 0;

        while (true)
        {
            if (session->send_id != -1)
            {
                // 其他线程正在发送
                usleep(100 * 1000);
                retry++;
                if (retry > PUB_TIMEOUT * 10)
                {
                    syslog(LOG_USER | LOG_WARNING, "publish wait timeout");
                    return -1;
                }
            }
            else
            {
                break;
            }
        }
        pthread_mutex_lock(&session->pub_lock);
        int rc = mosquitto_publish(mosq, &session->send_id, topic, payloadlen, payload, client->qos, false);
        if (rc == MOSQ_ERR_SUCCESS)
        {
            int ret;

            ret = lnxall_condvar_timedwait(&session->pub_cond, &session->pub_lock, PUB_TIMEOUT * 1000);
            if (ret == ETIMEDOUT)
            {
                // reconnect
                if (session->state == MQTT_CONNECTED)
                {
                    session->state = MQTT_CONNECTING;
                    mosquitto_reconnect_async(session->client.mosq);
                }
            }
            pthread_mutex_unlock(&session->pub_lock);
            session->send_id = -1;
            return ret;
        }
        else
        {
            pthread_mutex_unlock(&session->pub_lock);
            session->send_id = -1;
            return rc;
        }
    }
    else
    {
        int rc = mosquitto_publish(mosq, &session->send_id, topic, payloadlen, payload, client->qos, false);

        return rc;
    }
}

/**
*******************************************************************************
*
* @brief 自定义QoS和Retain的publish
* @param session MQTT 会话结构体的指针
* @param topic 消息topic
* @param payload 消息体
* @param payloadlen 消息长度
* @param qos  Qos
* @param retain retain
* @return 0 for success
*
*******************************************************************************
*/
int mqtt_session_publish_with_flag(mqtt_session_t *session, char *topic, char *payload, int payloadlen, int qos, bool retain)
{
    mqtt_client_t *client = &session->client;
    struct mosquitto *mosq = client->mosq;

    if (session->state != MQTT_CONNECTED)
    {
        syslog(LOG_USER | LOG_ERR, "MQTT client %s is disconnected", session->client.clientId);
        return -1;
    }

    if (topic == NULL || strlen(topic) == 0)
    {
        syslog(LOG_USER | LOG_WARNING, "topic is empty");
        return -1;
    }

    if (payload == NULL || strlen(payload) == 0)
    {
        syslog(LOG_USER | LOG_WARNING, "payload is empty, topic:%s", topic);
        return -1;
    }

    return mosquitto_publish(mosq, NULL, topic, payloadlen, payload, qos, retain);
}

int mqtt_session_conn(mqtt_session_t *session)
{
    mqtt_client_t *client = &session->client;
    struct mosquitto *mosq = client->mosq;

    mosquitto_connect(mosq, client->addr, client->port, client->keepalive);
    return 0;
}

/**
*******************************************************************************
*
* @brief 断开MQTT连接
* @param session MQTT 会话结构体的指针
* @return 0 for success; -1 for fail
*
*******************************************************************************
*/
int mqtt_session_disconn(mqtt_session_t *session)
{
    mqtt_client_t *client = &session->client;
    struct mosquitto *mosq = client->mosq;
    topic_node_t *node = NULL;
    topic_node_t *tmp = NULL;

    session->state = MQTT_DISCONNECTED;
    mosquitto_disconnect(mosq);

    pthread_rwlock_wrlock(&session->topic_lock);
    list_for_each_entry_safe(node, tmp, &session->topic_list, list)
    {
        list_del(&node->list);
        free(node);
    }
    pthread_rwlock_unlock(&session->topic_lock);
    return 0;
}

/**
*******************************************************************************
*
* @brief  Does a topic match a subscription?
* @param sub[in] 订阅的topic
* @param topic[in] 收到的topic
* @param result[out] 返回是否匹配
* @return 0 for success; -1 for fail
*
*******************************************************************************
*/
int mqtt_topic_matches_sub(const char *sub, const char *topic, unsigned char *result)
{
    return mosquitto_topic_matches_sub(sub, topic, (bool *)result);
}

/**
*******************************************************************************
*
* @brief  获取MQTT状态
* @param session MQTT 会话结构体的指针
* @return
    MQTT_CONNECTING, // 正在连接
    MQTT_CONNECTED, // 已连接
    MQTT_DISCONNECTED, //主动断开连接
*
*******************************************************************************
*/
mqtt_state_e mqtt_session_get_state(mqtt_session_t *session)
{
    return session->state;
}

int mqtt_session_destroy_and_free(mqtt_session_t *session)
{
    return 0;
}