#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 "zhongheng_common_socket.h"

char heartbeat[64];
common_socket_var_t gvar = {0};

static int get_rt_data_from_json_str(pp_real_time_data_t *real_data, char *json_str)
{
    int ret = 0;

    cJSON *root = cJSON_Parse(json_str);
    if (root)
    {
        char *payload = NULL;
        char *buf = NULL;
        int len;
        char *tmp;

        buf = (char *)malloc(MAXBUF);
        if (buf == NULL)
        {
            dy_syslog(LOG_ERR, "malloc failed");
            ret = -1;
            goto out;
        }

        GET_JSON_VALUE_STRING(root, "port", buf);
        real_data->port = port_char2enum(buf);
        GET_JSON_VALUE_INT(root, "instruction_code", real_data->instruction_code);
        GET_JSON_VALUE_INT(root, "mi", real_data->mi);
        GET_JSON_VALUE_INT(root, "ts", real_data->ts);
        GET_JSON_VALUE_INT(root, "len", real_data->len);
        GET_JSON_VALUE_INT(root, "sub_term_addr_len", real_data->sub_term_addr_len);
        GET_JSON_VALUE_INT(root, "start_addr", real_data->start_addr);
        GET_JSON_VALUE_STRING(root, "sub_template_id", real_data->sub_template_id);
        GET_JSON_VALUE_STRING(root, "src_identifier", real_data->src_identifier);
        GET_JSON_VALUE_STRING(root, "sn", real_data->sn);
        GET_JSON_VALUE_STRING(root, "dtu_sn", real_data->dtu_sn);

        GET_JSON_VALUE_DY_STRING(root, "term_addr_b64", tmp);
        len = b64_decode(tmp, buf, MAXBUF);
        ASSERT(len == sizeof(real_data->term_addr));
        memcpy(real_data->term_addr, buf, sizeof(real_data->term_addr));
        free(tmp);

        GET_JSON_VALUE_DY_STRING(root, "sub_term_addr_b64", tmp);
        len = b64_decode(tmp, buf, MAXBUF);
        ASSERT(len == sizeof(real_data->sub_term_addr));
        memcpy(real_data->sub_term_addr, buf, sizeof(real_data->sub_term_addr));
        free(tmp);

        GET_JSON_VALUE_DY_STRING(root, "data_b64", tmp);
		
		char *b64_buf = (char *)malloc(real_data->len * 2);
        if (b64_buf == NULL)
        {
            dy_syslog(LOG_ERR, "malloc failed");
            ret = -1;
			free(buf);
            goto out;
        }
        len = b64_decode(tmp, b64_buf, real_data->len * 2 + 1024);
        free(tmp);
        //dy_syslog(LOG_DEBUG, "len:%d, real_data->len:%d", len, real_data->len);
        ASSERT(len == real_data->len);
        real_data->data = malloc(len);
        if (real_data->data == NULL)
        {
            dy_syslog(LOG_ERR, "malloc fail len:%d", len);
            free(buf);
			free(b64_buf);
            goto out;
        }
        memcpy(real_data->data, b64_buf, len);
        free(buf);
		free(b64_buf);
    }
    else
    {
        dy_syslog(LOG_ERR, "json parse error");
        ret = -1;
    }

out:
    cJSON_Delete(root);

    return ret;
}


static int common_socket_heart_beat(common_socket_var_t *var)
{
    int ret;
	time_t now = time(NULL);
    socket_state_e state;

	TIMER_CONFIRM(var->heartbeat_timer);
	
	connect_config_t *connect_cfg = NULL;	
	list_for_each_entry(connect_cfg, &var->connect_list, list)
	{
        state = dy_socket_session_get_state(connect_cfg->socket_client);
        if (state != SOCKET_CONNECTED)
			continue;
		if(connect_cfg->last_send == 0 && connect_cfg->common_cfg.login_packet_len != 0)
		{
			connect_cfg->last_send = now;
            dy_syslog_hex(LOG_DEBUG, connect_cfg->common_cfg.login_packet, connect_cfg->common_cfg.login_packet_len, "len %d			网关->[%s]		", connect_cfg->common_cfg.login_packet_len, connect_cfg->addr);
			dy_socket_session_snd(connect_cfg->socket_client, connect_cfg->common_cfg.login_packet, connect_cfg->common_cfg.login_packet_len);
		}
		if(connect_cfg->common_cfg.heartbeat_interval)
		{
			ret = check_timeout_second(now, &connect_cfg->last_send, connect_cfg->common_cfg.heartbeat_interval);
			//dy_syslog(LOG_DEBUG, "ret %d", ret);
			if(ret)
			{
                dy_syslog_hex(LOG_DEBUG, connect_cfg->common_cfg.heartbeat_packet, connect_cfg->common_cfg.heartbeat_packet_len, "len %d			网关->[%s]		", connect_cfg->common_cfg.heartbeat_packet_len, connect_cfg->addr);
				dy_socket_session_snd(connect_cfg->socket_client, connect_cfg->common_cfg.heartbeat_packet, connect_cfg->common_cfg.heartbeat_packet_len);
			}
		}
	}

    return 0;
}

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

    while (1)
    {
        SELECT_INIT();
        SELECT_ADD_FD(var->heartbeat_timer);
        timeout.tv_usec = 0;
        timeout.tv_sec = 10;
        ret = select(maxfd + 1, &rset, 0, 0, &timeout);
        if (ret < 0)
        {
            printf("errno %d\n", errno);
            if (errno == EINTR)
            {
                continue;
            }
            else
            {
                break;
            }
        }
        else if (ret == 0)
        {
        }
        else
        {
            if (var->heartbeat_timer > 0 && FD_ISSET(var->heartbeat_timer, &rset))
            {
                FD_CLR(var->heartbeat_timer, &rset);
                common_socket_heart_beat(var);
            }
        }
    }
}

static void common_socket_subscribe_all(common_socket_var_t *var)
{
    ipc_session_t *session = var->session;
    char topic[TOPIC_MAX_LEN] = {0};

    snprintf(topic, TOPIC_MAX_LEN, "ipc/%s/+/device/+/data/%s", var->sn_str, TOPIC_UP_RAW_DATA);
    ipc_session_subscribe(session, topic);
}

static int common_socket_mqtt_handle_recv_msg(void *obj, ipc_msg_t *mqtt_msg)
{
    common_socket_var_t *var = (common_socket_var_t *)obj;

    dy_syslog(LOG_DEBUG, " MQTT client: received MQTT topic:%s payload length:%d",
              mqtt_msg->topic, mqtt_msg->payloadLen);
    if (strstr(mqtt_msg->topic, TOPIC_UP_RAW_DATA))
    {
    	int ret = 0;
    	pp_real_time_data_t real_data = {0};

	    ret = get_rt_data_from_json_str(&real_data, mqtt_msg->payload);
	    if (ret < 0)
	    {
	        dy_syslog(LOG_ERR, "parse real data structure failed");
	    }

		connect_config_t *connect_cfg = NULL;	
		list_for_each_entry(connect_cfg, &var->connect_list, list)
		{
			if(real_data.port == connect_cfg->common_cfg.binding_port)
			{
				//dy_syslog_hex(LOG_DEBUG, real_data.data, real_data.len, "send tcp data len:%d", real_data.len);
				if(connect_cfg->last_send == 0 && connect_cfg->common_cfg.login_packet_len != 0)
				{
                    dy_syslog_hex(LOG_DEBUG, connect_cfg->common_cfg.login_packet, connect_cfg->common_cfg.login_packet_len, "len %d			网关->[%s]		", connect_cfg->common_cfg.login_packet_len, connect_cfg->addr);
                    dy_socket_session_snd(connect_cfg->socket_client, connect_cfg->common_cfg.login_packet, connect_cfg->common_cfg.login_packet_len);
				}
				char *data = calloc(1, real_data.len+connect_cfg->common_cfg.frame_header_len+connect_cfg->common_cfg.frame_tail_len);
				if(data)
				{
					memcpy(data, connect_cfg->common_cfg.frame_header, connect_cfg->common_cfg.frame_header_len);
					memcpy(data+connect_cfg->common_cfg.frame_header_len, real_data.data, real_data.len);
					memcpy(data+connect_cfg->common_cfg.frame_header_len+real_data.len, connect_cfg->common_cfg.frame_tail, connect_cfg->common_cfg.frame_tail_len);
                    dy_syslog_hex(LOG_DEBUG, data, real_data.len + connect_cfg->common_cfg.frame_header_len + connect_cfg->common_cfg.frame_tail_len, "len %d			网关->[%s]		", real_data.len + connect_cfg->common_cfg.frame_header_len + connect_cfg->common_cfg.frame_tail_len, connect_cfg->addr);
                    dy_socket_session_snd(connect_cfg->socket_client, data, real_data.len + connect_cfg->common_cfg.frame_header_len + connect_cfg->common_cfg.frame_tail_len);
					free(data);
				}
			}
            if(real_data.port == connect_cfg->mqtt_cfg.binding_port)
            {
                mqtt_session_publish(connect_cfg->session, connect_cfg->mqtt_cfg.pub_topic, real_data.data, real_data.len);
            }
		}

		if (real_data.data)
	    {
	        free(real_data.data);
	    }
    }
}

static int send_data_to_port(common_socket_var_t *var, gw_port_e port, unsigned char *buf, int len)
{
    cJSON *rs_data = cJSON_CreateObject();
    int mi = time(NULL);
    char topic[TOPIC_MAX_LEN] = {0};
    static char *data = NULL;

    if (data == NULL)
    {
        data = malloc(4096);
    }
    b64_encode(buf, len, data, 4096);
    cJSON_AddStringToObject(rs_data, "raw_data", data);

    cJSON_AddStringToObject(rs_data, "data_b64", data);
    cJSON_AddNumberToObject(rs_data, "len", len);
    cJSON_AddNumberToObject(rs_data, "period", 0);
    cJSON_AddStringToObject(rs_data, "port", port_enum2char(port));
    cJSON_AddNumberToObject(rs_data, "mi", mi);
    cJSON_AddStringToObject(rs_data, "src_identifier", "transparent");
    cJSON_AddNumberToObject(rs_data, "communication_timeout", 0);

    char *data_tmp = cJSON_Print(rs_data);
    dy_syslog(LOG_DEBUG, "data_tmp:%s", data_tmp);

    snprintf(topic, TOPIC_MAX_LEN, "ipc/%s/%s/device/noneed/data/%s", var->sn_str, port_enum2char(port),
             TOPIC_SEND_RGLT_SIGNAL_RAW_DATA);
    ipc_session_publish(var->session, topic, data_tmp, strlen(data_tmp));
    // 回收资源
    free(data_tmp);
    cJSON_Delete(rs_data);
}

static int external_mqtt_handle_recv_msg(void *obj, mqtt_message_t *mqtt_msg)
{
    connect_config_t *connect_cfg = (connect_config_t *)obj;
    common_socket_var_t *var = connect_cfg->var;

    // send to port
    send_data_to_port(var, connect_cfg->mqtt_cfg.binding_port, mqtt_msg->payload, mqtt_msg->payloadLen);
}

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

    snprintf(clientId, MAX_CLIENT_ID_LEN, "%s", var->proc_name);
    var->session = ipc_session_new(clientId, (void*)var, IPC_DEFAULT);
    if (var->session == NULL) return -1;
    ipc_session_set_callbacks(var->session, common_socket_mqtt_handle_recv_msg, NULL);
    common_socket_subscribe_all(var);
    ipc_session_start(var->session);
}



static int socket_handle_rcv_msg(void *obj, char *buf, int len)
{
    connect_config_t *connect_cfg = (connect_config_t *)obj;
    common_socket_var_t *var = connect_cfg->var;

    dy_syslog_hex(LOG_DEBUG, buf, len, "len %d			网关<-[%s]		", len, connect_cfg->addr);

	if(len == connect_cfg->common_cfg.login_packet_len && connect_cfg->common_cfg.login_packet && memcmp(connect_cfg->common_cfg.login_packet, buf, len) == 0)
	{
		dy_syslog(LOG_DEBUG, "login packet...");
		return 0;
	}

	if(len == connect_cfg->common_cfg.heartbeat_packet_len && connect_cfg->common_cfg.heartbeat_packet && memcmp(connect_cfg->common_cfg.heartbeat_packet, buf, len) == 0)
	{
		dy_syslog(LOG_DEBUG, "heartbeat packet...");
		return 0;
	}

	connect_cfg->last_send = time(NULL);

    send_data_to_port(var, connect_cfg->common_cfg.binding_port, buf, len);
}

static int connect_state_change(void *obj, socket_state_e state)
{
    connect_config_t *connect_cfg = (connect_config_t *)obj;
    common_socket_var_t *var = connect_cfg->var;

	dy_syslog(LOG_DEBUG, "connect_state_change state %d", state);
	if(state == SOCKET_DISCONNECTED)
		connect_cfg->last_send = 0;
	
	return 0;
}

int del_space(char *src)
{
	char *pTmp = src;
	unsigned int iSpace = 0;
	
	while (*src != '\0')
	{
		if (*src != ' ')
		{
			*pTmp++ = *src;
		}
		else
		{
			iSpace++;
		}
		
		src++;
	}
	
	*pTmp = '\0';
	return iSpace;
}

static int common_socket_load_conf(common_socket_var_t *var, char *file)
{
	char buff[128] = {0};
	int ret 	   = -1;
	int i		   = 0;

	char *json_str = NULL;
    json_str = read_file_data(file);
    if (!json_str)
    {
        return -1;
    }

    cJSON *root = cJSON_Parse(json_str);
    if (!root)
    {
        return -1;
    }

    cJSON *nodes = cJSON_GetObjectItem(root, "common_socket_cfg");
    if (!nodes)
    {
        ret = -1;
        goto __cleanup;
    }

    int size = cJSON_GetArraySize(nodes);

    for (i = 0; i < size ; i++)
    {
        cJSON *node = cJSON_GetArrayItem(nodes, i);
        if (!node)
        {
            ret = i + 1;
            goto __cleanup;
        }

		connect_config_t *connect_cfg = calloc(1, sizeof(connect_config_t));

        connect_cfg->var = var;
		GET_JSON_VALUE_STRING(node, "protocol_type", buff);
        connect_cfg->type = socket_type_char2enum(buff);
        if (connect_cfg->type == SOCKET_MQTT)
        {
            char addr[MAX_HOST_LEN]; ///<MQTT地址
            unsigned short port;     ///<MQTT端口
            char user[MAX_USER_LEN]; ///<MQTT
            char pass[MAX_PASS_LEN];
            char clientId[MAX_CLIENT_ID_LEN] = {0};

            snprintf(clientId, MAX_CLIENT_ID_LEN, "%s_TRS_%d", var->sn_str, i);
            connect_cfg->session = mqtt_session_new(clientId, (void*)connect_cfg);
            if (connect_cfg->session == NULL)
            {
                continue;
            }
            GET_JSON_VALUE_STRING(node, "binding_port", buff);
            connect_cfg->mqtt_cfg.binding_port = port_char2enum(buff);

            GET_JSON_VALUE_STRING(node, "host", addr);
            GET_JSON_VALUE_INT(node, "port", port);
            GET_JSON_VALUE_STRING(node, "user", user);
            GET_JSON_VALUE_STRING(node, "pass", pass);
            char cafile_url[128] = {0}, certfile_url[128] = {0}, keyfile_url[128] = {0};
            GET_JSON_VALUE_STRING(node, "cafile", cafile_url);
            GET_JSON_VALUE_STRING(node, "certfile", certfile_url);
            GET_JSON_VALUE_STRING(node, "keyfile", keyfile_url);
            if (strlen(cafile_url) && strlen(certfile_url) && strlen(keyfile_url))
            {
                char cafile[128] = {0};
                char certfile[128] = {0};
                char keyfile[128] = {0};
                char path[128] = {0};
                sprintf(path, "%s/%d", COMMON_SOCKET_TLS_FILE_PATH, i);
                if (strlen(cafile_url))
                {
                    mqtt_tls_file_overlow(cafile_url, path, cafile, 128);
                }
                if (strlen(certfile_url))
                {
                    mqtt_tls_file_overlow(certfile_url, path, certfile, 128);
                }
                if (strlen(keyfile_url))
                {
                    mqtt_tls_file_overlow(keyfile_url, path, keyfile, 128);
                }
                dy_syslog(LOG_DEBUG, "cafile:%s,certfile:%s,keyfile:%s", cafile, certfile, keyfile);
                mqtt_session_set_tls(connect_cfg->session, cafile, certfile, keyfile);
            }
            mqtt_session_set_address(connect_cfg->session, addr, port, user, pass);
            mqtt_session_set_opts(connect_cfg->session, DEFAULT_QOS, KEEPALIVEINTERVAL);
            GET_JSON_VALUE_DY_STRING(node, "pub_topic", connect_cfg->mqtt_cfg.pub_topic);
            GET_JSON_VALUE_DY_STRING(node, "sub_topic", connect_cfg->mqtt_cfg.sub_topic);
        }
        else
        {
            GET_JSON_VALUE_DY_STRING(node, "socket_addr", connect_cfg->addr);
            GET_JSON_VALUE_INT(node, "socket_port", connect_cfg->port);
            GET_JSON_VALUE_STRING(node, "binding_port", buff);
            connect_cfg->common_cfg.binding_port = port_char2enum(buff);

            GET_JSON_VALUE_STRING(node, "frame_header", buff);
            del_space(buff);
            connect_cfg->common_cfg.frame_header_len = strlen(buff) / 2;
            if (connect_cfg->common_cfg.frame_header_len > 0)
            {
                connect_cfg->common_cfg.frame_header = calloc(1, connect_cfg->common_cfg.frame_header_len);
                str2hex(connect_cfg->common_cfg.frame_header, buff, connect_cfg->common_cfg.frame_header_len);
            }

            GET_JSON_VALUE_STRING(node, "frame_tail", buff);
            del_space(buff);
            connect_cfg->common_cfg.frame_tail_len = strlen(buff) / 2;
            if (connect_cfg->common_cfg.frame_tail_len > 0)
            {
                connect_cfg->common_cfg.frame_tail = calloc(1, connect_cfg->common_cfg.frame_tail_len);
                str2hex(connect_cfg->common_cfg.frame_tail, buff, connect_cfg->common_cfg.frame_tail_len);
            }

            GET_JSON_VALUE_STRING(node, "login_packet", buff);
            del_space(buff);
            connect_cfg->common_cfg.login_packet_len = strlen(buff) / 2;
            if (connect_cfg->common_cfg.login_packet_len > 0)
            {
                connect_cfg->common_cfg.login_packet = calloc(1, connect_cfg->common_cfg.login_packet_len);
                str2hex(connect_cfg->common_cfg.login_packet, buff, connect_cfg->common_cfg.login_packet_len);
            }

            GET_JSON_VALUE_STRING(node, "heartbeat_packet", buff);
            del_space(buff);
            connect_cfg->common_cfg.heartbeat_packet_len = strlen(buff) / 2;
            if (connect_cfg->common_cfg.heartbeat_packet_len > 0)
            {
                connect_cfg->common_cfg.heartbeat_packet = calloc(1, connect_cfg->common_cfg.heartbeat_packet_len);
                str2hex(connect_cfg->common_cfg.heartbeat_packet, buff, connect_cfg->common_cfg.heartbeat_packet_len);
            }

            GET_JSON_VALUE_INT(node, "heartbeat_interval", connect_cfg->common_cfg.heartbeat_interval);
        }

        dy_syslog(LOG_INFO, "==%d %d socket_addr:%s port:%d protocol_type:%d==", size, i, connect_cfg->addr, connect_cfg->port, connect_cfg->type);
		list_add_tail(&connect_cfg->list, &var->connect_list);
    }
    ret = 0;

__cleanup:
    cJSON_Delete(root);
    return ret;
}


static int common_socket_init(common_socket_var_t *var)
{	
	INIT_LIST_HEAD(&var->connect_list);
	
    get_board_sn(var->sn_str);
    get_process_name(var->proc_name);

    if (load_nodes_cfg(&var->nodes_cfg_table, NODES_CFG_PATH) == -1)
    {
        dy_syslog(LOG_ERR, "load nodes cfg fail");
    }

    common_socket_mqtt_client_init(var);
	
    if (access(COMMON_SOCKET_CFG, F_OK) != -1)
    {
        common_socket_load_conf(var, COMMON_SOCKET_CFG);
    }

	connect_config_t *connect_cfg = NULL;	
	list_for_each_entry(connect_cfg, &var->connect_list, list)
	{
		var->node_cnt++;
        if (connect_cfg->type == SOCKET_MQTT)
        {
            char topic[TOPIC_MAX_LEN] = {0};

            mqtt_session_set_callbacks(connect_cfg->session, external_mqtt_handle_recv_msg, NULL);
            mqtt_session_start(connect_cfg->session);
            mqtt_session_subscribe(connect_cfg->session, connect_cfg->mqtt_cfg.sub_topic);
        }
        else
        {
            connect_cfg->socket_client = dy_socket_session_init(connect_cfg->type, connect_cfg->addr,
                                                                connect_cfg->port,
                                                                socket_handle_rcv_msg,
                                                                connect_state_change,
                                                                (void *)connect_cfg);
        }
    }

    var->heartbeat_timer = my_timer_create();
    if (var->heartbeat_timer > 0)
    {
    	if(var->node_cnt)
			my_timer_set(var->heartbeat_timer, 3, 1000);
		else
        	my_timer_set(var->heartbeat_timer, 3, 10000);
    }

    dy_syslog(LOG_INFO, "init done, board SN:%s", var->sn_str);

    return 0;
}

static int get_date_times_s()
{
    int ret = -1;
    char buff[255] = {0};
    char *string = NULL;

	sprintf(buff, "uptime");
    ret = system_reply(buff, &string);
	dy_syslog(LOG_DEBUG, "ret %d string %p", ret, string);
    if (ret == 0)
    {
        if (!string)return 0;
		int m_v = 0;
		char *m = strstr(string, "up");
		dy_syslog(LOG_DEBUG, "string:%s %d", string, strlen(string));
		if(m)
		{
			sscanf(m+3, "%d", &m_v);
			dy_syslog(LOG_DEBUG, "m_v %d ", m_v);
		}
        free(string);
		if(m_v < 5) 		return 0;
		else	return 1;
    }
    return 1;
}

int main(int argc, char *argv[])
{
    common_socket_var_t *var = &gvar;

    memset(var, 0, sizeof(common_socket_var_t));
	if(get_date_times_s() == 0)
	{
		dy_syslog(LOG_INFO, "zhonheng sleep 210...");		
		sleep(210);
	}
	else
	{
		dy_syslog(LOG_INFO, "zhonheng sleep 300...");
		sleep(300);
	}
    common_socket_init(var);
    common_socket_loop(var);

    return 0;
}