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

#define IDENTIFIER_FLAG "thing.service."
#define RESPONSE_EVENT  "ResponseEvent"

linkkit_gateway_var_t glinkkit_var;

void alink_message_arrive(void *pcontext, void *pclient, iotx_mqtt_event_msg_pt msg)
{
    iotx_mqtt_topic_info_t     *topic_info = (iotx_mqtt_topic_info_pt) msg->msg;

    switch (msg->event_type) {
        case IOTX_MQTT_EVENT_PUBLISH_RECEIVED:
            /* print topic name and topic message */
            dy_syslog(LOG_DEBUG,"Message Arrived:");
            dy_syslog(LOG_DEBUG,"Topic  : %.*s", topic_info->topic_len, topic_info->ptopic);
            dy_syslog(LOG_DEBUG,"Payload: %.*s", topic_info->payload_len, topic_info->payload);

			if (strstr(topic_info->ptopic,"_reply"))
			{
				dy_syslog(LOG_DEBUG,"reply message!!!\n");
				break;
			}
			
			mqtt_message_t *mqtt_message = NULL;
			char cmd_identifier[64] = {0};
			char topic[TOPIC_MAX_LEN] = {0};
			char method[256] = {0};
			char* identifier = NULL;
			
			cJSON* root=cJSON_Parse(topic_info->payload);
	        if (!root)
			{
				dy_syslog(LOG_WARNING, "cJSON_Parse payload failed");
				return -1;
	        }
	        cJSON* nodes = cJSON_GetObjectItem(root,"params");
	        if (!nodes)
			{
				dy_syslog(LOG_WARNING, "cJSON_GetObjectItem params failed");
				return -1;
	        }

			GET_JSON_VALUE_STRING(root,"method",method);
			identifier = strstr(method, IDENTIFIER_FLAG);
			if(identifier)
			{
				strncpy(cmd_identifier, identifier+strlen(IDENTIFIER_FLAG), sizeof(cmd_identifier));
				cJSON_AddStringToObject(nodes,"identifier",cmd_identifier);
			}
			cJSON_AddStringToObject(nodes,"sn",pcontext);

			char *payload = cJSON_PrintUnformatted(nodes);
			snprintf(topic, TOPIC_MAX_LEN, "ipc/%s/%s/device/%s/data/%s", glinkkit_var.sn_str, "linkkit_gateway", pcontext, TOPIC_EVT_SET_RGLT);
			dy_syslog(LOG_DEBUG, "topic:%s payload %d %s", topic,strlen(payload),payload);
			ipc_session_publish(glinkkit_var.session, topic, payload, strlen(payload));
            break;
        default:
            break;
    }
}

int alink_subscribe(void *handle, char* product_key, char* device_name, char* pcontext, const char *fmt)
{
    int res = 0;
    char *topic = NULL;
    int topic_len = 0;

    topic_len = strlen(fmt) + strlen(product_key) + strlen(device_name) + 10;
    topic = HAL_Malloc(topic_len);
    if (topic == NULL) {
        dy_syslog(LOG_WARNING,"memory not enough");
        return -1;
    }
    memset(topic, 0, topic_len);
    HAL_Snprintf(topic, topic_len, fmt, product_key, device_name);

	res = IOT_MQTT_Subscribe_Sync(handle, topic, IOTX_MQTT_QOS0, alink_message_arrive, pcontext, 3000);
    if (res < 0)
	{
        dy_syslog(LOG_WARNING,"subscribe failed");
        HAL_Free(topic);
        return -1;
    }
	//dy_syslog(LOG_DEBUG,"IOT_MQTT_Subscribe %s\n", topic);

    HAL_Free(topic);
    return 0;
}

int alink_publish(void *handle, char	*ptopic, char *method, char *params)
{
    int 			res = 0;
    char		   *topic = NULL;
    int 			topic_len = 0;
    char		   *payload = "{\"id\":\"%d\",\"version\":\"1.0\",\"params\":%s,\"method\":\"%s\"}";
    static int report_id = 1;
    char *msg = NULL;
    int  msg_len = 0;

	topic_len = strlen(ptopic) + 1;
    topic = HAL_Malloc(topic_len);
    if (topic == NULL)
	{
        dy_syslog(LOG_WARNING,"memory not enough");
        return -1;
    }
	
    memset(topic, 0, topic_len);
	strncpy(topic, ptopic, strlen(ptopic) + 1);
	
    msg_len = strlen(payload) + 10 + strlen(params) + strlen(method) + 1;
    msg = HAL_Malloc(msg_len);
    if (msg == NULL)
	{
        dy_syslog(LOG_WARNING,"memory not enough");
        return -1;
    }
    memset(msg, 0, msg_len);
    /* devinfo update message */
    HAL_Snprintf(msg,
            msg_len,
            payload,
            report_id++,params,method
            );
	
    res = IOT_MQTT_Publish_Simple(handle, topic, IOTX_MQTT_QOS0, msg, strlen(msg));
    if (res < 0)
	{
        dy_syslog(LOG_WARNING,"publish failed, handle %p topic %s res = %d", handle,topic,res);
        HAL_Free(topic);
		dy_syslog(LOG_ERR,"%s exit(1)\n",__FUNCTION__);
		exit(1);
    }
	else
		dy_syslog(LOG_DEBUG, "alink pubish topic %s msg %s", topic,msg);
	
    HAL_Free(topic);
    HAL_Free(msg);
	
    return 0;
}

void alink_event_handle(void *pcontext, void *pclient, iotx_mqtt_event_msg_pt msg)
{
    //dy_syslog(LOG_DEBUG, "msg->event_type : %d pcontext : %s", msg->event_type,pcontext);
}

int alink_mqtt_connect(iotx_dev_meta_info_t *pmeta)
{
    int res = 0;
    iotx_mqtt_param_t mqtt_params;

    memset(&mqtt_params, 0x0, sizeof(mqtt_params));

    mqtt_params.handle_event.h_fp = alink_event_handle;
    pmeta->pclient = DY_IOT_MQTT_Construct(&mqtt_params, pmeta);
    if (pmeta->pclient == NULL)
	{
        dy_syslog(LOG_ERR,"--%s MQTT construct failed",pmeta->device_name);
        return -1;
    }
    pmeta->construct = 1;

	//res |= alink_subscribe(pmeta->pclient, pmeta->product_key, pmeta->device_name, pmeta->alias_sn, "/%s/%s/user/get");
	res |= alink_subscribe(pmeta->pclient, pmeta->product_key, pmeta->device_name, pmeta->alias_sn, "/sys/%s/%s/thing/service/#");
    if (res < 0) {
        DY_IOT_MQTT_Destroy(&pmeta->pclient);
		dy_syslog(LOG_ERR,"%s exit(1)\n",__FUNCTION__);
		exit(1);
    }

    IOT_MQTT_Yield(pmeta->pclient, 200);

    return 0;
}

void alink_get_message(linkkit_gateway_var_t* var)
{
    int ret;
	static int null_cnt = 0;
	
	linkkit_info_list_t *linkkit = NULL;
    list_for_each_entry(linkkit, &var->node_list,list)
    {
    	if(!linkkit->meta_info.construct)
			continue;
		
    	if(!linkkit->meta_info.pclient)
    	{
    		null_cnt++;
    		dy_syslog(LOG_WARNING, "get_msg device_name %s pclient is %p null_cnt(%d %d) %d\n",linkkit->meta_info.device_name,linkkit->meta_info.pclient,null_cnt,var->link_node_cnt,var->nodes_cfg_table->node_cnt);
			if(null_cnt > var->link_node_cnt)
			{
				dy_syslog(LOG_ERR,"%s exit(1)\n",__FUNCTION__);
				exit(1);
			}
			continue;
    	}
        //ret = alink_subscribe_t(linkkit->meta_info.pclient, linkkit->meta_info.product_key, linkkit->meta_info.device_name, linkkit->meta_info.alias_sn, "/%s/%s/user/get");
		ret |= alink_subscribe(linkkit->meta_info.pclient, linkkit->meta_info.product_key, linkkit->meta_info.device_name, linkkit->meta_info.alias_sn, "/sys/%s/%s/thing/service/#");
		if (ret < 0) {
			linkkit->meta_info.communication_fail++;
			if(linkkit->meta_info.communication_fail > 2)
			{
				ret=DY_IOT_MQTT_Destroy(&linkkit->meta_info.pclient);
				dy_syslog(LOG_WARNING,"DY_IOT_MQTT_Destroy ret %d\n",ret);
				alink_mqtt_connect(&linkkit->meta_info);
			}
    	}
		else
			linkkit->meta_info.communication_fail = 0;
    }
}

static int linkkit_gateway_node_connect(linkkit_gateway_var_t *var)
{
	int i;

	var->link_node_cnt = 0;
	for(i=0; i<var->nodes_cfg_table->node_cnt; i++)
    {
    	if(strlen(var->nodes_cfg_table->node[i].device_secret))
		{
			linkkit_info_list_t* linkkit = calloc(sizeof(linkkit_info_list_t), 1);

			strncpy(linkkit->meta_info.alias_sn, var->nodes_cfg_table->node[i].sn, sizeof(linkkit->meta_info.alias_sn));
			strncpy(linkkit->meta_info.device_name, var->nodes_cfg_table->node[i].device_name, sizeof(linkkit->meta_info.alias_sn));
			strncpy(linkkit->meta_info.device_secret, var->nodes_cfg_table->node[i].device_secret, sizeof(linkkit->meta_info.alias_sn));
			strncpy(linkkit->meta_info.product_key, var->nodes_cfg_table->node[i].product_key, sizeof(linkkit->meta_info.alias_sn));
			//strncpy(linkkit->meta_info.product_secret, var->nodes_cfg_table->node[i].product_key, sizeof(linkkit->meta_info.product_key));
			
   			if(!strlen(linkkit->meta_info.device_secret))
				continue;

			if(!strlen(linkkit->meta_info.device_name))
				strncpy(linkkit->meta_info.device_name, linkkit->meta_info.alias_sn, sizeof(linkkit->meta_info.device_name));
			
			var->link_node_cnt++;
	        dy_syslog(LOG_DEBUG,"===%s %d sn %s (%s %s %s %s)===\n",__FUNCTION__,__LINE__,linkkit->meta_info.alias_sn,linkkit->meta_info.device_name,linkkit->meta_info.device_secret,linkkit->meta_info.product_key,linkkit->meta_info.product_secret);
	        alink_mqtt_connect(&linkkit->meta_info);
	        list_add_tail(&linkkit->list,&var->node_list);
		}
    }
}

static int mqtt_msg_process(linkkit_gateway_var_t *var, int type, ipc_msg_t *mqtt_msg)
{
	int ret;
	static int null_cnt = 0;
	tag_table_t tag;

	//先检查参数
	ret = get_tag_data_from_str(&tag, mqtt_msg->payload);
    if (ret < 0)
    {
        dy_syslog(LOG_ERR, "get data structure failed");
        return -1;
    }
	
	linkkit_info_list_t *linkkit = NULL;
	list_for_each_entry(linkkit, &var->node_list, list)
	{
		if(linkkit->meta_info.construct && !linkkit->meta_info.pclient)
    	{
    		null_cnt++;
    		dy_syslog(LOG_WARNING,"[WARNING]--msg_process device_name %s pclient is %p null_cnt(%d %d) %d\n",linkkit->meta_info.device_name,linkkit->meta_info.pclient,null_cnt,var->link_node_cnt,var->nodes_cfg_table->node_cnt);
			if(null_cnt > var->link_node_cnt)
			{
				dy_syslog(LOG_ERR,"%s exit(1)\n",__FUNCTION__);
				exit(1);
			}
			continue;
    	}
		if(strcmp(linkkit->meta_info.device_name, tag.sn) == 0 || strcmp(linkkit->meta_info.alias_sn, tag.sn) == 0)
		{
			char topic[256] = {0};
			char method[256] = {0};
			char *params = tag.tag_node;
			
			if (params == NULL)
			{
		        dy_syslog(LOG_WARNING,"type %d params is NULL!!!",type);
		        break;
		    }
			if(type == 0)
			{
				char* fmt = "/sys/%s/%s/thing/event/property/post";
				char* fmt_method = "thing.event.property.post";
				
				snprintf(topic, sizeof(topic), fmt, linkkit->meta_info.product_key, linkkit->meta_info.device_name);
				alink_publish(linkkit->meta_info.pclient, topic, fmt_method, params);
			}
			else if(type == 1)
			{
				char* fmt = "/sys/%s/%s/thing/event/%s/post";
				char* fmt_method = "thing.event.%s.post";
				
				snprintf(topic, sizeof(topic), fmt, linkkit->meta_info.product_key, linkkit->meta_info.device_name, tag.identifier);
				snprintf(method, sizeof(method), fmt_method, tag.identifier);
				dy_syslog(LOG_DEBUG,"topic %s method %s %d %d",topic,method,sizeof(topic),sizeof(method));
				alink_publish(linkkit->meta_info.pclient, topic, method, params);
			}
			else if(type == 2)
			{
				char* fmt = "/sys/%s/%s/thing/event/%s/post";
				char* fmt_method = "thing.event.%s.post";
				
				snprintf(topic, sizeof(topic), fmt, linkkit->meta_info.product_key, linkkit->meta_info.device_name, var->response_event);
				snprintf(method, sizeof(method), fmt_method, var->response_event);
				dy_syslog(LOG_DEBUG,"topic %s method %s %d %d",topic,method,sizeof(topic),sizeof(method));
				alink_publish(linkkit->meta_info.pclient, topic, method, params);
			}
			break;
		}
	}
}

void msg_mqtt_recv(linkkit_gateway_var_t *var, int type, ipc_msg_t *msg)
{
    int ret,i;

	data_list_t* data = calloc(1,sizeof(data_list_t));	
	if(data)
	{
        int size = strlen(msg->topic) + 1 + msg->payloadLen;
        data->mqtt.topic = malloc(size);
		strcpy(data->mqtt.topic, msg->topic);
		data->type = type;
		data->mqtt.payloadLen = msg->payloadLen;
        data->mqtt.payload = data->mqtt.topic + strlen(data->mqtt.topic) + 1;
		memcpy(data->mqtt.payload, msg->payload, data->mqtt.payloadLen);
		var->data_cnt++;
		dy_syslog(LOG_DEBUG, "++%s topic %s data_cnt %d\n",__FUNCTION__,data->mqtt.topic,var->data_cnt);
    	list_add_tail(&data->list,&var->data_list);
	}

	var->data_flag = 1;
}

static int mqtt_msg_process_loop(linkkit_gateway_var_t *var)
{
	data_list_t* data = NULL;	
	list_for_each_entry(data, &var->data_list, list)
	{
		dy_syslog(LOG_DEBUG, "var->data_cnt %d", var->data_cnt);
		var->data_cnt--;
		mqtt_msg_process(var, data->type, &data->mqtt);
		list_del(&data->list);
        free(data->mqtt.topic);
        free(data);
		break;
	}

	var->data_cnt = 0;
	var->data_flag = 0;
}

static void linkkit_gateway_loop(linkkit_gateway_var_t *var)
{
	int ret = -1, maxfd = 0, i;
    fd_set rset;
    struct timeval timeout;
	
    while (1)
    {
    	SELECT_INIT();
        SELECT_ADD_FD(var->get_msg_timer);

		if(var->data_flag)
        {
			timeout.tv_usec = 50000;
        	timeout.tv_sec = 0;
        }
		else
		{
			timeout.tv_usec = 0;
        	timeout.tv_sec = 2;
		}

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

            if (errno == EINTR)
            {
                continue;
            }
            else
            {
                break;
            }
        }
        else if (ret > 0)
        {
            if (var->get_msg_timer > 0 && FD_ISSET(var->get_msg_timer, &rset))
            {
                FD_CLR(var->get_msg_timer, &rset);
				TIMER_CONFIRM(var->get_msg_timer);
                if (!var->data_flag)
					alink_get_message(var);
            }
        }

		mqtt_msg_process_loop(var);
    }
}

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

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

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

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

static int linkkit_gateway_mqtt_handle_recv_msg(void *obj, ipc_msg_t *msg)
{
    linkkit_gateway_var_t *var = (linkkit_gateway_var_t *)obj;
    bool matched;
    int ret;
    dy_syslog(LOG_DEBUG, " MQTT client: received MQTT topic:%s payload length:%d",
            msg->topic, msg->payloadLen);

    ret = mosquitto_topic_matches_sub("ipc/+/+/device/+/data_filtered/property/+", msg->topic, &matched);
    if (ret == 0 && matched)
    {
        //实时数据
        msg_mqtt_recv(var, 0, msg);
    }
    else
    {
        //事件
        ret = mosquitto_topic_matches_sub("ipc/+/+/device/+/data_filtered/event/+", msg->topic, &matched);
        if (ret == 0 && matched)
        {
            msg_mqtt_recv(var, 1, msg);
        }
        else
        {
            ret = mosquitto_topic_matches_sub("ipc/+/+/device/+/data_filtered/service/+", msg->topic, &matched);
            if (ret == 0 && matched)
            {
                msg_mqtt_recv(var, 2, msg);
            }
        }
    }
}

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

    snprintf(clientId, MAX_CLIENT_ID_LEN, "INT_linkkit_gw_%s", var->sn_str);
    var->session = ipc_session_new(clientId, (void*)var, IPC_DEFAULT);
    if (var->session == NULL) return -1;

    ipc_session_set_callbacks(var->session, linkkit_gateway_mqtt_handle_recv_msg, NULL);
    linkkit_gateway_subscribe_all(var);
    ipc_session_start(var->session);

}

int linkkit_gateway_init(linkkit_gateway_var_t *var)
{
    int i, j;
    const char *board_name = NULL;

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

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

    if (load_nodes_cfg(&var->nodes_cfg_table, NODES_CFG_PATH) == -1)
    {
        dy_syslog(LOG_ERR, "load nodes cfg fail");
    }
	
    if (load_templates_cfg(&var->template_table, TEMPLATES_CFG_PATH) == -1)
    {
        dy_syslog(LOG_ERR, "load template cfg fail");
    }
	
    kv_array_init(&var->identifier_backup, 32);
    linkkit_gateway_load_config(var);
	
	if(strlen(var->response_event) == 0)
		strncpy(var->response_event, RESPONSE_EVENT, sizeof(RESPONSE_EVENT));
	
	dy_syslog(LOG_DEBUG,"%s response_event:%s \n", __FUNCTION__,var->response_event);
    linkkit_gateway_mqtt_client_init(var);
	linkkit_gateway_node_connect(var);
	
	var->get_msg_timer = my_timer_create();
    if (var->get_msg_timer > 0)
    {
        my_timer_set(var->get_msg_timer, 1, 1000);
    }
	
    return 0;
}


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

	dy_syslog(LOG_DEBUG, "\n\
            |********************************************|\n\
            |           linkkit_gateway start  X_X         |\n\
            |********************************************|\n");
    memset(var, 0, sizeof(linkkit_gateway_var_t));
    linkkit_gateway_init(var);
    linkkit_gateway_loop(var);

    return 0;
}