#include <stdio.h>
#include <stdlib.h>
#include <sys/stat.h>

#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>

#include "common.h"
#include "dy_utils/cJSON.h"
#include "dy_utils/dy_common.h"

socket_mqtt_var_t socket_mqtt_var = {0};

static int front_cfg_parse(socket_mqtt_var_t *var, const char *cfg_file)
{
    char *data = NULL;
    cJSON *root = NULL;
    cJSON *mqtts = NULL;
    cJSON *mqtt = NULL;
    cJSON *tcp = NULL;
    cJSON *udp = NULL;
    int mqtt_cnt = 0;
    int i = 0;
    int ret = 0;
    char addr[MAX_HOST_LEN];          ///<MQTT地址
    unsigned short port;              ///<MQTT端口
    char user[MAX_USER_LEN];          ///<MQTT
    char pass[MAX_PASS_LEN];          ///<MQTT地址
    char clientId[MAX_CLIENT_ID_LEN]; ///<clientId

    if (var == NULL || cfg_file == NULL)
    {
        dy_syslog(LOG_ERR, "socket var:%p cfg_file:%p", var, cfg_file);
        return -1;
    }

    data = read_file_data(cfg_file);
    if (data == NULL)
    {
        dy_syslog(LOG_ERR, "read cfg file %s error", cfg_file);
        ret = -1;
        goto out;
    }

    root = cJSON_Parse(data);
    if (!root)
    {
        dy_syslog(LOG_ERR, "parse cfg file %s error", cfg_file);
        ret = -1;
        goto out;
    }
    tcp = cJSON_GetObjectItem(root, "TCP");
    if (!tcp)
    {
        dy_syslog(LOG_ERR, "get TCP failed");
    }
    else
    {
        GET_JSON_VALUE_INT(tcp, "port", var->tcp_port);
    }

    udp = cJSON_GetObjectItem(root, "UDP");
    if (!udp)
    {
        dy_syslog(LOG_ERR, "get UDP failed");
    }
    else
    {
        GET_JSON_VALUE_INT(udp, "port", var->udp_port);
    }

    mqtts = cJSON_GetObjectItem(root, "MQTT");
    if (!mqtts)
    {
        dy_syslog(LOG_ERR, "get MQTT failed");
        ret = -1;
        goto out;
    }

    mqtt_cnt = cJSON_GetArraySize(mqtts);
    if (mqtt_cnt == 0)
    {
        ret = 0;
        dy_syslog(LOG_INFO, "No MQTT Channel");
        var->mqtt_cnt = mqtt_cnt;
        goto out;
    }

    if (mqtt_cnt > MAX_MQTT_CHANNEL_CNT)
    {
        dy_syslog(LOG_WARNING, "exceed the max support channel count:%d, max:%d", mqtt_cnt, MAX_MQTT_CHANNEL_CNT);
        mqtt_cnt = MAX_MQTT_CHANNEL_CNT;
    }
    var->mqtt_cnt = mqtt_cnt;

    for (i = 0; i < mqtt_cnt; i++)
    {

        mqtt = cJSON_GetArrayItem(mqtts, i);

        GET_JSON_VALUE_STRING(mqtt, "host", addr);
        GET_JSON_VALUE_INT(mqtt, "port", port);
        GET_JSON_VALUE_STRING(mqtt, "user", user);
        GET_JSON_VALUE_STRING(mqtt, "pass", pass);

        if (cJSON_GetObjectItem(mqtt, "client_id") != NULL)
        {
            GET_JSON_VALUE_STRING(mqtt, "client_id", clientId);
        }
        if (strlen(clientId) == 0)
        {
            snprintf(clientId, MAX_CLIENT_ID_LEN, "%s_%d_FRT", var->sn_str, i);
        }

        mqtt_session_t *session = mqtt_session_new(clientId, NULL);
        if (session == NULL)
        {
            dy_syslog(LOG_ERR, "malloc error");
            return -1;
        }

        var->mqtt[i] = session;
        mqtt_session_set_address(session, addr, port, user, pass);
    }
out:
    cJSON_Delete(root);
    if (data != NULL)
    {
        free(data);
    }
    return ret;
}

static void main_loop(socket_mqtt_var_t *var)
{
    int ret = -1, maxfd;
    fd_set rset;
    struct timeval timeout;

    while (1)
    {
        SELECT_INIT();
        timeout.tv_usec = 0;
        timeout.tv_sec = 10;
        ret = select(maxfd + 1, &rset, 0, 0, &timeout);
        if (ret < 0)
        {
            dy_syslog(LOG_ERR, "errno %d", errno);
            if (errno == EINTR)
            {
                continue;
            }
            else
            {
                break;
            }
        }
        else if (ret > 0)
        {
        }
    }
}

int main_init(socket_mqtt_var_t *var)
{
    int i;

    get_board_sn(var->sn_str);
    front_cfg_parse(var, FRONT_CFG);

    // Initialize the socket receive side data
    INIT_LIST_HEAD(&var->socket_rcv_head);

    pthread_mutex_init(&var->socket_data_lock, NULL);
    pthread_cond_init(&var->socket_recv_cond, NULL);

    pthread_rwlock_init(&var->sock_sess_lock, NULL);

    mqtt_init(var);

    return 0;
}

int main(int argc, char *argv[])
{
    int port = -1;

    pthread_t thread_tcp_server;
    pthread_t thread_pkt_forward;
    memset(&socket_mqtt_var, 0, sizeof(socket_mqtt_var));

    main_init(&socket_mqtt_var);

    if (socket_mqtt_var.tcp_port != 0)
    {
        port = socket_mqtt_var.tcp_port;
        pthread_create(&thread_tcp_server, 0, start_tcp_server, (void *)&port);
    }

    pthread_create(&thread_pkt_forward, 0, socket_forward_loop, NULL);

    main_loop(&socket_mqtt_var);

    pthread_join(thread_tcp_server, NULL);
    pthread_join(thread_pkt_forward, NULL);

    return 0;
}