#include <sys/select.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/syscall.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <errno.h>
#include "modbus_tcp_common.h"
#include "dy_utils/hash_intptr.h"

struct modbus_shared_ctx {
	char maddr[32];
	modbus_t * mctx;
	pthread_mutex_t mutex;
};

static struct hash_intptr * g_hash = NULL;
static pthread_mutex_t g_mutex = PTHREAD_MUTEX_INITIALIZER;
gw_port_e ports[] = {MODBUS_TCP};


template_cfg_t *modbus_tcp_find_template(template_table_t *template_table, char *ptemplate_id)
{
    int i;
    for (i = 0; i < template_table->template_cnt; i++) {
        if (strcmp(ptemplate_id, template_table->template[i].template_id) == 0) {
            return &(template_table->template[i]);
        }
    }

    return NULL;
}

static modbus_t * modbus_newtcp(const char * ipaddr, int portno, pthread_mutex_t * * mutpp, int * isold)
{
    char * maddr;
    int mlen, ret;
    modbus_t * rval;
    struct modbus_shared_ctx * sctx;

    *isold = 0;
    rval = NULL;
    sctx = NULL;
    maddr = NULL;

    mlen = asprintf(&maddr, "%s:%d", ipaddr, portno);
    if (mlen <= 0) {
        dbg_syslog(LOG_ERR, "Error, system out of memory!");
        return NULL;
    }

    ret = pthread_mutex_lock(&g_mutex);
    if (ret) {
        dbg_syslog(LOG_ERR, "Error, mutex lock has failed: %s\n", ret);
        goto err0;
    }

    hash_intptr_findptr(g_hash, maddr, (unsigned int) mlen, (void * *) &sctx);
    if (sctx != NULL) {
        *isold = 1;
        *mutpp = &sctx->mutex;
        rval = sctx->mctx;
        goto err1;
    }

    sctx = (struct modbus_shared_ctx *) calloc(0x1, sizeof(*sctx));
    if (sctx == NULL) {
        dbg_syslog(LOG_ERR, "Error, System out of memory!");
        goto err1;
    }

    strncpy(sctx->maddr, maddr, sizeof(sctx->maddr) - 1);
    sctx->mctx = modbus_new_tcp(ipaddr, portno);
    if (sctx->mctx == NULL) {
        dbg_syslog(LOG_ERR, "Error, modbus_new_tcp(%s, %d) has failed.",
            ipaddr, portno);
        free(sctx);
        goto err1;
	}

    ret = pthread_mutex_init(&sctx->mutex, NULL);
    if (ret != 0) {
        modbus_close(sctx->mctx);
        modbus_free(sctx->mctx);
        sctx->mctx = NULL;
        free(sctx);
        dbg_syslog(LOG_ERR, "Error, failed to initialize mutex: %d", ret);
        goto err1;
    }

    ret = hash_intptr_addptr(&g_hash, maddr, (unsigned int) mlen, sctx);
    if (ret < 0) {
        modbus_close(sctx->mctx);
        modbus_free(sctx->mctx);
        sctx->mctx = NULL;
        free(sctx);
        dbg_syslog(LOG_ERR, "Error, failed to push shared ctx: %s", maddr);
        goto err1;
    }

    *mutpp = &sctx->mutex;
    rval = sctx->mctx;

err1:
    ret = pthread_mutex_unlock(&g_mutex);
    if (ret != 0) {
        dbg_syslog(LOG_ERR, "Error, failed to unlock mutex: %d", ret);
    }
err0:
    if (maddr != NULL) {
        free(maddr);
    }

    return rval;
}

void modbus_freetcp(const char * ipaddr, int portno, modbus_t * mctx)
{
    char * maddr;
    int mlen, ret;
    struct modbus_shared_ctx * sctx;

    sctx = NULL;
    maddr = NULL;

    mlen = asprintf(&maddr, "%s:%d", ipaddr, portno);
    if (mlen <= 0) {
        dbg_syslog(LOG_ERR, "Error, system out of memory!");
        return;
    }

    ret = pthread_mutex_lock(&g_mutex);
    if (ret) {
        dbg_syslog(LOG_ERR, "Error, mutex lock has failed: %s\n", ret);
        goto err0;
    }

    hash_intptr_findptr(g_hash, maddr, (unsigned int) mlen, (void * *) &sctx);
    if (sctx != NULL) {
        if (sctx->mctx != mctx) {
            dbg_syslog(LOG_ERR, "Error, modbus context mismatch: %p, %p!",
                sctx->mctx, mctx);
        }

        modbus_close(sctx->mctx);
        modbus_free(sctx->mctx);
        sctx->mctx = NULL;
        pthread_mutex_destroy(&sctx->mutex);
    }

    hash_intptr_remove(&g_hash, maddr, (unsigned int) mlen);

    ret = pthread_mutex_unlock(&g_mutex);
    if (ret != 0) {
        dbg_syslog(LOG_ERR, "Error, failed to unlock mutex: %d", ret);
    }
err0:
    if (maddr != NULL) {
        free(maddr);
    }
}

int modbus_tcp_client_init(modbus_tcp_thread_var_t *tcp_thread_var, pthread_mutex_t * * mutexpp)
{
    int isold = 0, ret;
    uint32_t old_response_to_sec;
    uint32_t old_response_to_usec;
    uint32_t new_response_to_sec;
    uint32_t new_response_to_usec;
    pthread_mutex_t * mutexp = NULL;

    tcp_thread_var->ctx = modbus_newtcp(tcp_thread_var->tcp_ip_addr, tcp_thread_var->tcp_port, &mutexp, &isold);
    if (tcp_thread_var->ctx == NULL) {
        dbg_syslog(LOG_ERR, "Unable to allocate libmodbus context\n");
        return -1;
    }

	/* for old modbus-TCP connection, just skip following steps */
    if (isold) {
        if (mutexpp != NULL) {
            *mutexpp = mutexp;
        }
        return 0;
    }

    ret = (mutexp != NULL) ? pthread_mutex_lock(mutexp) : -1;
    if (ret) {
        dbg_syslog(LOG_ERR, "Error, failed to lock mutex: %d", ret);
        return -1;
    }

    modbus_set_debug(tcp_thread_var->ctx, FALSE); // TRUE
    modbus_set_error_recovery(tcp_thread_var->ctx, MODBUS_ERROR_RECOVERY_LINK | MODBUS_ERROR_RECOVERY_PROTOCOL);
    modbus_get_response_timeout(tcp_thread_var->ctx, &old_response_to_sec, &old_response_to_usec);

    if (tcp_thread_var->communication_timeout >= 1500) {
        modbus_set_response_timeout(tcp_thread_var->ctx, tcp_thread_var->communication_timeout / 1000, (tcp_thread_var->communication_timeout % 1000) * 1000);
    }
    else {
        modbus_set_response_timeout(tcp_thread_var->ctx, 1, 500000);
    }

    if (modbus_connect(tcp_thread_var->ctx) == -1) {
        dbg_syslog(LOG_ERR, "Connection failed: %s\n", modbus_strerror(errno));
		pthread_mutex_unlock(mutexp);
		modbus_freetcp(tcp_thread_var->tcp_ip_addr, tcp_thread_var->tcp_port, tcp_thread_var->ctx);
        tcp_thread_var->ctx = NULL;
        return -1;
    }
    if (tcp_thread_var->communication_timeout < 1500) {
        modbus_set_response_timeout(tcp_thread_var->ctx, tcp_thread_var->communication_timeout / 1000, (tcp_thread_var->communication_timeout % 1000) * 1000);
    }
    modbus_get_response_timeout(tcp_thread_var->ctx, &new_response_to_sec, &new_response_to_usec);
    dbg_syslog(LOG_DEBUG, "ip_add %s:%d Connection succeed ctx %p..., timeouts: (%u, %u), (%u, %u)",
        tcp_thread_var->tcp_ip_addr, tcp_thread_var->tcp_port, tcp_thread_var->ctx,
        old_response_to_sec, old_response_to_usec, new_response_to_sec, new_response_to_usec);

	pthread_mutex_unlock(mutexp);
	if (mutexpp != NULL) {
        *mutexpp = mutexp;
    }
    return 0;
}

static int modbus_tcp_load_tcp_connections(modbus_tcp_var_t *var)
{
    int i;
    int find = 0;
    modbus_tcp_thread_var_t *tcp_thread_var = NULL;
    template_cfg_t *ptemplate = NULL;

    for (i = 0; i < var->nodes_cfg_table->node_cnt; i++) {
        if (strlen(var->nodes_cfg_table->node[i].app_key) == 0 && var->nodes_cfg_table->node[i].port == MODBUS_TCP) {
            ptemplate = modbus_tcp_find_template(var->template_table, var->nodes_cfg_table->node[i].template_id);
            if (ptemplate == NULL) {
                dbg_syslog(LOG_ERR, "cannot find template id:%s", var->nodes_cfg_table->node[i].template_id);
                exit(-1);
            }
            find = 0;
            list_for_each_entry(tcp_thread_var, &var->tcp_thread_list, list) {
                if (strcmp(var->nodes_cfg_table->node[i].tcp_ip_addr, tcp_thread_var->tcp_ip_addr) == 0 &&
                    var->nodes_cfg_table->node[i].tcp_port == tcp_thread_var->tcp_port)
                {
                    find = 1;
                    if (ptemplate->communication_timeout > tcp_thread_var->communication_timeout) {
                        tcp_thread_var->communication_timeout = ptemplate->communication_timeout;
                    }
                    break;
                }
            }
            if (find == 0) {
                tcp_thread_var = calloc(1, sizeof(modbus_tcp_thread_var_t));
                INIT_LIST_HEAD(&tcp_thread_var->data_list);
                INIT_LIST_HEAD(&tcp_thread_var->node_list);
                INIT_LIST_HEAD(&tcp_thread_var->list);

                strncpy(tcp_thread_var->tcp_ip_addr, var->nodes_cfg_table->node[i].tcp_ip_addr, sizeof(tcp_thread_var->tcp_ip_addr));
                tcp_thread_var->tcp_port = var->nodes_cfg_table->node[i].tcp_port;
                tcp_thread_var->communication_timeout = ptemplate->communication_timeout;
                list_add_tail(&tcp_thread_var->list, &var->tcp_thread_list);
                tcp_thread_var->var = var;
                pthread_mutex_init(&tcp_thread_var->data_lock, NULL);
            }

            tcp_node_t *node = calloc(1, sizeof(tcp_node_t));
            node->tcp_thread = tcp_thread_var;
            node->node_cfg = &var->nodes_cfg_table->node[i];
            list_add_tail(&node->list, &tcp_thread_var->node_list);
        }
    }

    list_for_each_entry(tcp_thread_var, &var->tcp_thread_list, list) {
        if (tcp_thread_var->communication_timeout == 0) {
            tcp_thread_var->communication_timeout = MODBUS_TCP_COMMUNICATION_TIMEOUT;
        }
        modbus_tcp_client_init(tcp_thread_var, NULL);
    }

    return 0;
}

// 建立与内部broker之间的MQTT连接
static int modbus_tcp_mqtt_client_init(modbus_tcp_var_t *var)
{
    char clientId[MAX_CLIENT_ID_LEN] = {0};

    snprintf(clientId, MAX_CLIENT_ID_LEN, "INT_modbus_tcp_%s", var->sn_str);
    dbg_syslog(LOG_NOTICE, "clientId: %s", clientId);
    var->session = ipc_session_new(clientId, (void *)var, IPC_DEFAULT);
    if (var->session == NULL) {
        return -1;
    }
        
    ipc_session_start(var->session);

    return 0;
}

static int modbus_tcp_init(modbus_tcp_var_t *var)
{
    int i;
    const char *board_name = NULL;
    modbus_tcp_thread_var_t *tcp_thread_var = NULL;

    check_make_dir(NODES_CACHE);
    check_make_dir(NODES_CFG);
    get_board_sn(var->sn_str);
    get_process_name(var->proc_name);
    board_name = get_board_name();

    dbg_syslog(LOG_NOTICE, "board SN: %s", var->sn_str);
    dbg_syslog(LOG_NOTICE, "board name: %s", board_name);

    if (load_nodes_cfg(&var->nodes_cfg_table, NODES_CFG_PATH) == -1) {
        dbg_syslog(LOG_ERR, "load nodes cfg fail");
    }
    if (load_templates_cfg(&var->template_table, TEMPLATES_CFG_PATH) == -1) {
        dbg_syslog(LOG_ERR, "load template cfg fail");
    }

    dev_status_init(&var->ds, var->nodes_cfg_table, var->template_table, ports, ARRAY_SIZE(ports));
    node_sta_recovery(var->ds, MODBUS_TCP_NODE_STATUS_BAK_FILE);

    INIT_LIST_HEAD(&var->tcp_thread_list);
    modbus_tcp_load_tcp_connections(var);
    list_for_each_entry(tcp_thread_var, &var->tcp_thread_list, list) {
        modbus_tcp_start_thread(var, tcp_thread_var);
    }

    var->node_status_timer = my_timer_create();
    if (var->node_status_timer > 0) {
        my_timer_set(var->node_status_timer, 1, 5000);
    }

    modbus_tcp_mqtt_client_init(var);

    return 0;
}

static int ds_flush_timer(modbus_tcp_var_t *var)
{
    time_t now = time(NULL);

    TIMER_CONFIRM(var->node_status_timer);

    if (now >= var->nodes_status_sync_date + 30) {
        static time_t lasttime = 0;
        var->nodes_status_sync_date = now;
        char *node_status = ds_print(var->ds);

        dbg_syslog(LOG_INFO, "online_status_changed:%d strlen(node_status):%d, (now-lasttime):%d",
                  var->ds->online_status_changed, strlen(node_status), now - lasttime);
        if (node_status && strlen(node_status) > 20 && (var->ds->online_status_changed || now > lasttime + 3600)) {    //在线状态变化上报，1小时也会上报
            char topic[TOPIC_MAX_LEN];
            int ret = 0;

            snprintf(topic, TOPIC_MAX_LEN, "ipc/%s/%s/%s", var->sn_str, var->proc_name, TOPIC_EVT_NODESSTATUS);
            dbg_syslog(LOG_INFO, "report node_status topic  : %s", topic);
            dbg_syslog(LOG_INFO, "report node_status payload: %s", node_status);
            ret = ipc_session_publish(var->session, topic, node_status, strlen(node_status));
            if (ret == 0) {
                lasttime = now;
                var->ds->online_status_changed = false;
            }
            node_sta_backup(var->ds, MODBUS_TCP_NODE_STATUS_BAK_FILE);
        }
        if (node_status) {
            write_file_data(NODES_CACHE "/modbus_tcp_status.json", node_status, strlen(node_status));
            free(node_status);
        }
    }

    return 0;
}

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

    while (1) {
        SELECT_INIT();
        SELECT_ADD_FD(var->node_status_timer);

        timeout.tv_usec = 0;
        timeout.tv_sec = 5;

        ret = select(maxfd + 1, &rset, 0, 0, &timeout);
        if (ret < 0) {
            int error = errno;
            dbg_syslog(LOG_INFO, "errno %d\n", error);

            if (error == EINTR) {
                continue;
            }
            else {
                break;
            }
        }
        else if (ret > 0) {
            if (var->node_status_timer > 0 && FD_ISSET(var->node_status_timer, &rset)) {
                FD_CLR(var->node_status_timer, &rset);
                ds_flush_timer(var);
            }
        }
    }
}

int main(int argc, char *argv[])
{
    int sdelay = 0;
    modbus_tcp_var_t var;
    memset(&var, 0, sizeof(var));

    lnxall_loglevel_set(LNXALL_LOGNOTICE, 1);

    if (argc > 1 && argv[1] != NULL) {
        sdelay = (int) strtol(argv[1], NULL, 0);
    }

    /* delay 10 seconds on start */
    if (sdelay > 0 && sdelay < 1800) {
        int ret;
        struct timespec ts = {
            .tv_sec = (time_t) sdelay,
            .tv_nsec = 0
        };
        ret = nanosleep(&ts, NULL);
        dbg_syslog(LOG_ERR, "delay(%d) has returned: %d", sdelay, ret);
    }

    modbus_tcp_init(&var);
    modbus_tcp_loop(&var);

    return 0;
}