#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/time.h>
#include <errno.h>
#include <string.h>

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

extern int dy_mqtt_session_subscribe_all(mqtt_session_t *session);
static int _do_recv(mqtt_session_t *session, struct list_head *recv_list)
{
    ipc_msg_t *msg = NULL;
    ipc_msg_t *tmp = NULL;

    if (list_empty(recv_list))
    {
        return 0;
    }
    list_for_each_entry_safe(msg, tmp, recv_list, list)
    {
        session->handle_recv_msg(session->user_data, msg);
        list_del(&msg->list);
        free(msg->topic);
        free(msg);
    }
    return 0;
}

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 */
        fprintf(stderr, "Error, failed to system uptime for condition variable!\n");
        fflush(stderr);
        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) {
        fprintf(stderr, "Error, failed to wait cond-var: %d (%s)\n",
            ret, strerror(ret) ? : "");
        fflush(stderr);
    }
    return ret;
}

static int mqtt_recv_do_loop(mqtt_session_t *session)
{
    struct list_head tmp_recv_list = LIST_HEAD_INIT(tmp_recv_list);

    pthread_mutex_lock(&session->recv_lock);
    while (session->received == 0)
    {
        lnxall_condvar_timedwait(&session->recv_cond, &session->recv_lock, 5000);
    }

    if (session->received)
    {
        // real received packets
        if (!list_empty(&session->rcv_list))
        {
            list_move_tail_list(&session->rcv_list, &tmp_recv_list);
        }
        session->received = 0;
        pthread_mutex_unlock(&session->recv_lock);

        _do_recv(session, &tmp_recv_list);
    }
    else
    {
        pthread_mutex_unlock(&session->recv_lock);
    }

    return 0;
}

void *mqtt_recv_loop(void *param)
{
    mqtt_session_t *session = (mqtt_session_t *)param;

    sleep(5); // sleep a while wait system ready

    while (1)
    {
        mqtt_recv_do_loop(session);
    }
}