#include "../common.h"
#include "../ems/emsctrlProc.h"
#include "../log_module/log_module.h"
#include "lz4.h"
#include <math.h>
#include "north_mqtt.h"
#include "north_jsonct.h"
#include "../proto_forward.h"
#include "../business_log/business_log.h"

int north_g_reflush_all_plan = 1;
all_plan north_g_all_plan[NORTH_CHANNEL_CONRTOL_CNT][MAX_DAY_YEAR+1] = {0};//mqtt北向可最多存储一年计划，如果北向存在当天策略则执行北向策略，否则到emms2_plan.json里找

static mqt_north_var_t mqt_north_var = {0};
mqt_north_var_t *get_mqt_north_var(void)
{
    return &mqt_north_var;
}

static int north_mqtt_publish_lz4(mqt_north_channel_t *channel,char *pub_topic,char *data,int data_len)
{
	mqtt_session_t *session = channel->session;
    char no_str[32];
    snprintf(no_str, sizeof(no_str), "%d", channel->index);

    if (channel->lz4){ // 当前通道开启lz4压缩
    
	    int ret = 0;
	    int src_size = data_len + 1;
	    int max_dst_size = LZ4_compressBound(src_size);
	    char *compressed_data = calloc(1,max_dst_size);

	    const int compressed_data_size = LZ4_compress_default(data, compressed_data, src_size, max_dst_size); 
	    if (compressed_data_size <= 0)
	    {
	        ems_syslog(LOG_ERR, "LZ4_compress err");
	        free(compressed_data);
	        return -1;
	    }
	    // printf("We successfully compressed some data! Ratio: %.2f\n",
	    //     (float) compressed_data_size/src_size);
	    char topic_lz4[256] = {0};
	    sprintf(topic_lz4,"%s/lz4",pub_topic);
	    ems_syslog(LOG_INFO,"lz4 change topic [%s]",topic_lz4);
	    ret = mqtt_session_publish(session,topic_lz4,compressed_data,compressed_data_size);
        if(ret == 0)
        {
            BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0022, no_str, "[北向]:index[%s]:标准北向mqtt发布lz压缩消息成功", no_str);
        }
        else
        {
            BUSINESS_LOG(BLOG_ERR, NORTH_ID_0022, no_str, "[北向]:index[%s]:标准北向mqtt发布lz压缩消息失败", no_str);
        }
	    free(compressed_data);
	    return ret;
    }
    else
    {
        int ret = mqtt_session_publish(session,pub_topic,data,data_len);
        if(ret == 0)
        {
            BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0023, no_str, "[北向]:index[%s]:标准北向mqtt发布消息成功", no_str);
        }
        else
        {
            BUSINESS_LOG(BLOG_ERR, NORTH_ID_0023, no_str, "[北向]:index[%s]:标准北向mqtt发布消息失败", no_str);
        }

        return ret;
    }
}

static int report_data_poll(mqt_north_var_t *var)
{
	TIMER_CONFIRM(var->report_timer);

    int dev_cnt = 0;
	char *data = NULL;

	data = _report_AllDev_data_print(var, &dev_cnt, var->mi++);

	if (dev_cnt > 0){  // 
	    char pub_topic[512] = {0};
	    sprintf(pub_topic, "lnxall/"North_Post"/%s/%s", var->sn_str, FUNCTION_PeriodReport);
	    ems_syslog(LOG_INFO, "pub topic [%s] data:[%s]", pub_topic, data);
		for (int i = 0; i < var->channel_cnt; i++)
		{
		    if (var->channel[i].comm_status == COMM_STATUS_LOGIN){
				if(north_mqtt_publish_lz4(&var->channel[i], pub_topic, data, strlen(data)) != 0)
					ems_syslog(LOG_WARNING,"channel %d send data error!!", i);
		    }
		}
	}
	
    if (data != NULL)
	    free(data);

	return 0;
}

static void report_all_alarm(mqt_north_channel_t *channel, char* function, int mi)
{
    int alarm_cnt = 0;
	char *data = NULL;
	mqt_north_var_t *var = channel->var;

	data = _report_allalarm_print(channel, &alarm_cnt, function, mi);
 
    char pub_topic[512] = {0};
    sprintf(pub_topic, "lnxall/"North_GetRsp"/%s/%s", var->sn_str, function);
    ems_syslog(LOG_INFO, "pub topic [%s] data:[%s]", pub_topic, data);
    if(north_mqtt_publish_lz4(channel, pub_topic, data, strlen(data)) != 0)
        ems_syslog(LOG_WARNING, "send exit all alarm data error!!");
	
    if (data != NULL)
	    free(data);
}

static void report_alarm(mqt_north_var_t *var)
{
    int alarm_change_cnt = 0;
	char *data = NULL;

	data = _report_alarm_print(var, &alarm_change_cnt, FUNCTION_AlarmReport, var->mi++);

    if (alarm_change_cnt > 0){ // 
	    char pub_topic[512] = {0};
	    sprintf(pub_topic, "lnxall/"North_Post"/%s/%s", var->sn_str, FUNCTION_AlarmReport);
	    ems_syslog(LOG_INFO, "pub topic [%s] data:[%s]", pub_topic, data);
		for (int i = 0; i < var->channel_cnt; i++)
        {
            if (var->channel[i].comm_status == COMM_STATUS_LOGIN){
	            if(north_mqtt_publish_lz4(&var->channel[i], pub_topic, data, strlen(data)) != 0)
	                ems_syslog(LOG_WARNING,"channel %d send alarm error!!", i);
            }
        }
    }
	
    if (data != NULL)
	    free(data);
}

static void get_single_device_data(mqt_north_channel_t *channel, char *device_type, char *no, char* function_id, int mi)
{
    char *data = NULL;
	mqt_north_var_t *var = channel->var;

	data = _report_SingleDev_data_print(channel, device_type, no, function_id, mi);

    char pub_topic[512] = {0};
    sprintf(pub_topic, "lnxall/"North_GetRsp"/%s/%s", var->sn_str, function_id);
    ems_syslog(LOG_INFO, "pub topic [%s] data:[%s]", pub_topic, data);
    if(north_mqtt_publish_lz4(channel, pub_topic, data, strlen(data)) != 0)
        ems_syslog(LOG_WARNING, "send single device data error!!");
	
    if (data != NULL)
	    free(data);
}

static char *error_info[] = {"", "can find tag", "to many message write error", "have force set"};

static int north_set_ems(mqt_north_channel_t *channel, char *name, double value) // return 0: 写成功; -1: 找不到tag; -2: to many message write error; -3: have force set
{
    int ret = -1;
    proto_forward_t *var_proto = get_proto_forward_var();     
    tag_t *tag_find = find_tag_by_name(var_proto,name);
    char addtmp[256] = {0};
    if ((tag_find) && (tag_find->point_type == PROPERTY_REALDATA)) //> 没办法判断是不是读写数据 
    {
        tag_value_t tag_val = {0};
        tag_val.type = TYPE_TAG_FLOAT;
        tag_val.value.to_float = value;
        ret = write_data_to_tag_by_p(tag_find, &tag_val, 1.0);
        if (ret < 0)
        {
            goto ERR;
        }
        
        snprintf(addtmp, sizeof(addtmp), "%s/:%s/%f", channel->name, tag_find->name, value);
        server_add_new_log(LOG_CODE_NORTH_MQTT_WRITE_POINT, ((device_t*)tag_find->dev_ptr)->no, LOG_MODE_TRIG, 1, addtmp);
        return ret;
    }
ERR:
    snprintf(addtmp, sizeof(addtmp), "%s/:%s/%f", channel->name, name, value);
    server_add_new_log(LOG_CODE_NORTH_MQTT_WRITE_POINT, "EMS", LOG_MODE_TRIG, 2, addtmp);
    return ret;
}

static void north_write_response(mqt_north_channel_t *channel, int mi, N_RESPONSE_RESULT result, char* function_id, char *errormsg) 
{
    time_t timestamp = time(NULL);
    char data[1024] = {0};
	mqt_north_var_t *var = channel->var;

    snprintf(data, 1024, "{\"identifier\":\"%s\",\"stationno\":\"%s\",\"stationname\":\"%s\",\"result\":%d,\"mi\":%d,\"timestamp\":%lu,\"errormsg\":\"%s\"}",
		 function_id, var->cab_id, var->cab_station_name, result, mi, timestamp, errormsg);

    char pub_topic[512] = {0};
    sprintf(pub_topic, "lnxall/"North_SetRsp"/%s/%s", var->sn_str, function_id);
    ems_syslog(LOG_INFO, "pub topic [%s] data:[%s]", pub_topic, data);
    if(north_mqtt_publish_lz4(channel, pub_topic, data, strlen(data)) != 0)
        ems_syslog(LOG_WARNING, "send response error!!");
}

static int send_login_msg(mqt_north_var_t *var, mqt_north_channel_t *channel)
{
    login_info_t info;
    char *str = NULL;

    info.timestamp = time(NULL);
    info.serial_num = channel->login.mi;
    channel->login.mi++;
    str = _login_print(var, &info, var->sn_str);

	ems_syslog(LOG_WARNING, "north mqtt login...");
    char pub_topic[512] = {0};
    sprintf(pub_topic,"lnxall/Post/%s/Login",var->sn_str); 
    ems_syslog(LOG_WARNING, "pub topic [%s] str:[%s]", pub_topic, str);
    if (north_mqtt_publish_lz4(channel, pub_topic, str, strlen(str)) != 0)
		ems_syslog(LOG_ERR, "send login error!!");

    if (str != NULL)
        free(str);
    return 0;
}

static int send_heartbeat_msg(mqt_north_var_t *var, mqt_north_channel_t *channel)
{
    char *str = NULL;
    int mi = channel->heart_beat.mi++;
    str = _heartbeat_print(var, mi, var->sn_str);

    char pub_topic[512] = {0};
    sprintf(pub_topic,"lnxall/Post/%s/HeartBeat",var->sn_str); 
    ems_syslog(LOG_INFO, "pub topic [%s] str:[%s]", pub_topic, str);
    if (north_mqtt_publish_lz4(channel, pub_topic, str, strlen(str)) != 0)
		ems_syslog(LOG_ERR, "send heartbeat error!!");
	
    if (str != NULL)
        free(str);
    return 0;
}

static int status_poll(mqt_north_var_t *var)
{
    TIMER_CONFIRM(var->status_poll_timer);

    int i;
    time_t now = time(NULL);

    for (i = 0; i < var->channel_cnt; i++)
    {
        mqt_north_channel_t *channel = &var->channel[i];
        char no_str[32];
        snprintf(no_str, sizeof(no_str), "%d", channel->index);
        BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0020, no_str, "[北向]:index[%s]:北向mqtt通道 status:%d", no_str, channel->comm_status);
		ems_syslog(LOG_NOTICE, "north mqtt channel[%d] status:%d", i, channel->comm_status);
        if (channel->comm_status == COMM_STATUS_CONNECTED)
        {
            if (now > channel->login.last_ts + var->login)
            {
                send_login_msg(var, channel);
                channel->login.last_ts = now;
            }
        }
        else if (channel->comm_status == COMM_STATUS_LOGIN)
        {
            if (now >= (channel->heart_beat.last_ts + var->heart_beat))
            {
                channel->heart_beat.last_ts = now;
                send_heartbeat_msg(var, channel);
                channel->heart_beat_lost_cnt++;
            }
        }
    }

    return 0;
}

static int cloud_north_mqtt_state_change(void *obj, int state)
{
    mqt_north_channel_t *channel = (mqt_north_channel_t *)obj;

    char no_str[32];
    snprintf(no_str, sizeof(no_str), "%d", channel->index);

    switch (state)
    {
    case MQTT_CONNECTING:
        channel->comm_status = COMM_STATUS_CONNECTING;
        BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0008, no_str, "[北向]:index[%s]:北向mqtt通道连接中", no_str);
        break;
    case MQTT_CONNECTED:
        channel->comm_status = COMM_STATUS_CONNECTED;
        BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0008, no_str, "[北向]:index[%s]:北向mqtt通道已连接", no_str);
        break;
    case MQTT_DISCONNECTED:
        ems_syslog(LOG_WARNING, "MQTT Disconnect");
        BUSINESS_LOG(BLOG_ERR, NORTH_ID_0008, no_str, "[北向]:index[%s]:北向mqtt通道断开连接", no_str);
        channel->comm_status = COMM_STATUS_IDLE;
        break;
    default:
        ems_syslog(LOG_ERR, "wrong mqtt state:%d", state);
        BUSINESS_LOG(BLOG_ERR, NORTH_ID_0008, no_str, "[北向]:index[%s]:北向mqtt通道错误状态%d", no_str,state);
        break;
    }
    return 0;
}

static void msg_login_result(mqt_north_channel_t *channel, char *payload)
{
    login_info_t info = {0};
    int ret = login_prase(&info, payload);

    char no_str[32];
    snprintf(no_str, sizeof(no_str), "%d", channel->index);

    if (!ret)
    {
        if (info.error_code == LOGIN_PASS)
        {
            channel->comm_status = COMM_STATUS_LOGIN;
            ems_syslog(LOG_INFO, "gateway login success");
            BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0009, no_str, "[北向]:index[%s]:北向mqtt login success", no_str);
        }
        else if (info.error_code == LOGIN_REFUSE)
        {
            ems_syslog(LOG_WARNING, "gateway login failed");
            BUSINESS_LOG(BLOG_WARNING, NORTH_ID_0009, no_str, "[北向]:index[%s]:北向mqtt login failed", no_str);
        }
    }
}

static void msg_confirm_connect(mqt_north_channel_t *var)
{
    var->heart_beat_lost_cnt = 0;
}

extern int north_control;

void north_all_plan_set_zero(int channel_index)
{
    if(channel_index >= NORTH_CHANNEL_CONRTOL_CNT)
    {
        return;
    }
    for(int i = 0; i < MAX_DAY_YEAR; i++)
    {
        memset(&north_g_all_plan[channel_index][i],0,sizeof(north_g_all_plan[channel_index][i]));
    }
}

// lnxall/[动作码]/[SN]/[功能码]
static int recv_north_cloud_mqtt_msg(void *obj, mqtt_message_t *mqtt_msg)
{
	size_t GETDATALEN = (mqtt_msg->payloadLen) * 15;
	ems_syslog(LOG_INFO, "received MQTT topic:%s payload length:%d", mqtt_msg->topic, mqtt_msg->payloadLen);

    mqt_north_channel_t *channel = (mqt_north_channel_t *)obj;
	int ret_dec = 0;
	cJSON *root = NULL;

    char *get_data = calloc(GETDATALEN,1);
    char message_type[64] = {0}, tmp_c[64] = {0}, function_id[64] = {0};
    if (strstr(mqtt_msg->topic, "lz4") && channel->lz4 == 1)
    {
        ret_dec = LZ4_decompress_safe(mqtt_msg->payload, get_data, mqtt_msg->payloadLen, GETDATALEN);
        if(ret_dec < 0)
        {
            ems_syslog(LOG_ERR, "Not enough space!!!");
            if (get_data != NULL)
                free(get_data);
            BUSINESS_LOG(BLOG_ERR, NORTH_ID_0010, NULL, "[北向]:北向mqtt收到LZ4数据Not enough space");
            return -1;
        }
        else
        {
            BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0010, NULL, "[北向]:北向mqtt收到LZ4数据解析正常");
        }
        ems_syslog(LOG_DEBUG, "decompress  payload %s", get_data);
        sscanf(mqtt_msg->topic,"lnxall/%[^/]/%[^/]/%s/lz4", message_type, tmp_c, function_id);
    }
    else
    {
        strncpy(get_data, mqtt_msg->payload, mqtt_msg->payloadLen);
        sscanf(mqtt_msg->topic, "lnxall/%[^/]/%[^/]/%s", message_type, tmp_c, function_id);
    }

    ems_syslog(LOG_INFO, "External MQTT client received MQTT topic:%s payload:%s",
              mqtt_msg->topic, get_data);
	
	// lnxall/Set/[SN]/[FUNC_ID]
    ems_syslog(LOG_INFO, "received message_type:%s function_id:%s", message_type, function_id);

    if (strcmp(function_id, FUNCTION_HeartBeat) == 0)
    {
        msg_confirm_connect(channel);
    }
    else if (strcmp(function_id, FUNCTION_LOGIN) == 0)
    {
        msg_login_result(channel, get_data);
    }
    else
    {
        if (channel->comm_status == COMM_STATUS_LOGIN) //>登录成功，Set\Get才有效
		{
		    root = cJSON_Parse(get_data);
		    if (!root)
		    {
		        ems_syslog(LOG_ERR, "json parse error");
		        if (get_data != NULL)
		        {
		            free(get_data);
		        }
                BUSINESS_LOG(BLOG_ERR, NORTH_ID_0011, NULL, "[北向]:北向mqtt收到数据解析成json格式失败");
		        return -1;
		    }
            else
            {
                BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0011, NULL, "[北向]:北向mqtt收到数据解析成json格式成功");
            }
		    cJSON *item = cJSON_GetObjectItem(root, "mi");
		    if (!item)
		    {
		        ems_syslog(LOG_ERR, "no mi!");
		        goto END;
		    }
		    int mi = item->valueint;
		    ems_syslog(LOG_ERR, "mi %d!", mi);
			int ret = 0;
			int error_cnt = 0;
			char error_all[1024] = "";
		    if ((strcmp(message_type, North_Set) == 0) && (north_control == channel->index)) //设置参数
		    {
		        ems_syslog(LOG_INFO, "north mqtt "North_Set);
		        if (strcmp(function_id, FUNCTION_EmsSet) == 0)
		        {
                    BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0013, NULL, "[北向]:北向mqtt收到设置参数数据function id:[EmsSet]");
		            item = cJSON_GetObjectItem(root, "tags");
		            if (!item)
		            {
		                ems_syslog(LOG_ERR, "no tags!");
						sprintf(error_all, "%s", "no tags");
		                ret = -1;
		                goto SET_END;
		            }
					char errormsg[128] = "";
		            int tag_num = cJSON_GetArraySize(item);
		            for (size_t i = 0; i < tag_num; i++)
		            {
		                cJSON *tag = cJSON_GetArrayItem(item, i);
		                char name_ems[64] = {0};
		                sprintf(name_ems, DEV_NO_EMS".%s", tag->string);
		                ems_syslog(LOG_NOTICE, "set:%s -> %.2f", name_ems, tag->valuedouble);
		                ret = north_set_ems(channel, name_ems, tag->valuedouble);
						if (ret < 0){
							error_cnt++;
			                snprintf(errormsg, 128, "%s:%s  ", name_ems, error_info[abs(ret)]);
							strcat(error_all, errormsg);
						}
		            }
		        }
				else if(strcmp(function_id, FUNCTION_SCHEDULE) == 0)
		        {
                    BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0013, NULL, "[北向]:北向mqtt收到设置参数数据function id:[Schedule]");
					cJSON *messages = cJSON_GetObjectItem(root, "messages");
					if (!messages)
					{
						ems_syslog(LOG_ERR, "no messages!");
						sprintf(error_all, "%s", "no messages");
						ret = -1;
						goto SET_END;
					}
                    north_all_plan_set_zero(north_control);
					if (north_parse_all_plan(messages,north_control) < 0)
					{
					    sprintf(error_all, "%s", "format error");
						ret = -1;
						goto SET_END;
					}
					north_g_reflush_all_plan = 1;
					north_save_all_plan_to_file(messages,north_control);
					server_add_new_log(LOG_CODE_NORTH_MQTT_WRITE_PLAN, "EMS", LOG_MODE_TRIG, 1, channel->name);
			    }
				else{
					ret = -1;
					sprintf(error_all, "%s", "no support function id");
                    BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0013, NULL, "[北向]:北向mqtt收到设置参数数据 未支持的function id");
				}
	SET_END:   
		        if (ret == 0 && error_cnt == 0)
		        {
                    BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0012, NULL, "[北向]:北向mqtt收到设置参数数据解析成功");
		            north_write_response(channel, mi, SUCCESS, function_id, error_all);
		        }
		        else
		        {
                    BUSINESS_LOG(BLOG_ERR, NORTH_ID_0012, NULL, "[北向]:北向mqtt收到设置参数数据解析失败");
		            north_write_response(channel, mi, ERROR, function_id, error_all);
		        }
		    }
			else if (strcmp(message_type, North_Get) == 0) //获取数据
	        {       
		        ems_syslog(LOG_INFO, "north mqtt "North_Get);
		        if (strcmp(function_id, FUNCTION_DevData) == 0)
		        {
	                char deviceType[DEV_NO_LEN] = "";
	                char no[DEV_NO_LEN] = "";
					GET_JSON_VALUE_STRING(root, "deviceType", deviceType);
					GET_JSON_VALUE_STRING(root, "no", no);
		            ems_syslog(LOG_INFO, "deviceType: %s, no: %s", deviceType, no);
		            get_single_device_data(channel, deviceType, no, function_id, mi);
                    BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0014, NULL, "[北向]:北向mqtt收到获取数据 function id:[DevData]");
		        }
				else if (strcmp(function_id, FUNCTION_AllAlarm) == 0)
	            {
	                report_all_alarm(channel, function_id, mi);
                    BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0014, NULL, "[北向]:北向mqtt收到获取数据 function id:[AllAlarm]");
				}
			}
        }
    }

END:
	if (root != NULL)
	{
		cJSON_Delete(root);
	}
	if (get_data != NULL)
	{
		free(get_data);
	}
	return 0;
}

static void lc_cabno_report(mqt_north_var_t *var)
{
    time_t now = time(NULL);
    if(now - var->cabno_report > CABNO_INTERVAL)
    {
        var->cabno_report = now;
        char *str = NULL;
        int mi = 1;
        str = _lc_cabno_print(var, mi);
        if(str != NULL)
        {
            for (int i = 0; i < var->channel_cnt; i++)
            {
                mqt_north_channel_t *channel = &var->channel[i];
                char pub_topic[512] = {0};
                sprintf(pub_topic,"lnxall/Post/%s/CabInfo",var->sn_str); 
                ems_syslog(LOG_INFO, "pub topic [%s] str:[%s]", pub_topic, str);
                if (north_mqtt_publish_lz4(channel, pub_topic, str, strlen(str)) != 0)
                    ems_syslog(LOG_ERR, "send lc_cabno error!!");
            }
            free(str);
        }
    }
}

static void north_mqtt_subscribe(mqt_north_channel_t *channel)
{
    char sub_topic[MAX_BASE_LEN] = {0};
    mqt_north_var_t *var = channel->var;

    sprintf(sub_topic, "lnxall/%s/%s/+", North_PostRsp, var->sn_str);     //登陆、心跳回复      lnxall/Post/[SN]/+
    mqtt_session_subscribe(channel->session, sub_topic);
    ems_syslog(LOG_NOTICE, "sub topic:%s", sub_topic);

	if (channel->lz4){
	    sprintf(sub_topic, "lnxall/%s/%s/+/lz4", North_PostRsp, var->sn_str); //>登陆、心跳回复 压缩
	    mqtt_session_subscribe(channel->session, sub_topic);
	    ems_syslog(LOG_NOTICE, "sub topic:%s", sub_topic);
	}
 
    sprintf(sub_topic, "lnxall/%s/%s/+", North_Get, var->sn_str);     //获取数据      lnxall/Get/[SN]/+
    mqtt_session_subscribe(channel->session, sub_topic);
    ems_syslog(LOG_NOTICE, "sub topic:%s", sub_topic);

	if (channel->lz4){
	    sprintf(sub_topic, "lnxall/%s/%s/+/lz4", North_Get, var->sn_str); //>获取数据 压缩
	    mqtt_session_subscribe(channel->session, sub_topic);
	    ems_syslog(LOG_NOTICE, "sub topic:%s", sub_topic);
	}
	
    sprintf(sub_topic, "lnxall/%s/%s/+", North_Set, var->sn_str);     //参数设置      lnxall/Set/[SN]/EmsSet
    mqtt_session_subscribe(channel->session, sub_topic);
    ems_syslog(LOG_NOTICE, "sub topic:%s", sub_topic);

	if (channel->lz4){
	    sprintf(sub_topic, "lnxall/%s/%s/+/lz4", North_Set, var->sn_str); //>参数设置 压缩
	    mqtt_session_subscribe(channel->session, sub_topic);
	    ems_syslog(LOG_NOTICE, "sub topic:%s", sub_topic);
	}
}

static int north_mqtt_channel_create(mqt_north_var_t *var)
{
    int i = 0;
	// char sub_topic[MAX_BASE_LEN] = {0};

    for (i = 0; i < var->channel_cnt; i++)
    {
        mqt_north_channel_t *channel = &var->channel[i];
        char no_str[32];
        snprintf(no_str, sizeof(no_str), "%d", i);
        if(channel->enable == 1)
        {
            BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0005, no_str, "[北向]:channel[%s]:北向mqtt通道已创建", no_str);
            ems_syslog(LOG_ERR,"channel %d create",i);
            channel->connect_flag = true;
            channel->var = var;
            mqtt_session_set_callbacks(channel->session, recv_north_cloud_mqtt_msg, cloud_north_mqtt_state_change);
            channel->comm_status = COMM_STATUS_CONNECTING;
            mqtt_session_start(channel->session);
            north_mqtt_subscribe(channel);
        }
        else
        {
            BUSINESS_LOG(BLOG_ERR, NORTH_ID_0005, no_str, "[北向]:channel[%s]:北向mqtt通道未创建", no_str);
        }
    }
    return 0;
}

static int set_channel_obj_from_json(cJSON *obj, char *str, char *target_str, size_t target_size) 
{
    if (obj == NULL || str == NULL || target_str == NULL || target_size == 0) {
        ems_syslog(LOG_ERR, "Invalid input parameters in set_channel_obj_from_json");
        return -1;
    }

    cJSON *item = cJSON_GetObjectItem(obj, str);
    if (item == NULL) {
        ems_syslog(LOG_DEBUG, "JSON item with key '%s' not found", str);
        return -1;
    }

    if (item->type != cJSON_String) {
        ems_syslog(LOG_ERR, "JSON item '%s' is not a string type", str);
        return -1;
    }

    size_t value_len = strlen(item->valuestring);
    if (target_size <= value_len) {
        ems_syslog(LOG_ERR, "target_str len error, need %zu but only %zu available", value_len + 1, target_size);
        return -1;
    }

    snprintf(target_str, target_size, "%s", item->valuestring);
    return 0;
}


static int load_uplink_cfg(mqt_north_var_t *var, char *data)
{
    cJSON *root = NULL;
    cJSON *objs = NULL;
    cJSON *obj = NULL;
    int cnt = 0;
    int i = 0, j = 0;
    int ret = 0;

    root = cJSON_Parse(data);
    if (!root)
    {
        ems_syslog(LOG_ERR, "parse cfg file error");
        ret = -1;
        goto out;
    }

	if (cJSON_GetObjectItem(root, "interval") != NULL) //>获取每类设备的数据上报周期，默认：300s
	{
		cJSON *interval = cJSON_GetObjectItem(root, "interval");
		for (j = 0; j < var->num; j++)
		{
			GET_JSON_VALUE_INT(interval, var->mqt_north_class[j]->class_type, var->mqt_north_class[j]->interval);
		}
	}
    if (cJSON_GetObjectItem(root, "heart_beat") != NULL) 
	{
        cJSON *heart_beat = cJSON_GetObjectItem(root, "heart_beat");
        var->heart_beat = heart_beat->valueint;
    }
    else
    {
        var->heart_beat = 60;
    }
    if (cJSON_GetObjectItem(root, "login") != NULL) 
	{
        cJSON *login = cJSON_GetObjectItem(root, "login");
        var->login = login->valueint;
    }
    else
    {
        var->login = 5;
    }
    // MQTT
    objs = cJSON_GetObjectItem(root, "standard");
    if (!objs)
    {
        ems_syslog(LOG_ERR, "get standard failed");
        ret = -1;
        goto out;
    }

    cnt = cJSON_GetArraySize(objs);
    if (cnt == 0)
    {
        ret = 0;
        ems_syslog(LOG_ERR, "No UPLINK Channel");
        goto out;
    }

    // if (cnt > 8)
    // {
    //     ems_syslog(LOG_WARNING, "exceed the max support channel count:%d, max:%d", cnt, MAX_UPLINK_CHANNEL_CNT);
    //     cnt = MAX_UPLINK_CHANNEL_CNT;
    // }

    for (i = 0; i < cnt; i++)
    {
        char clientId[128] = {0}; ///< clientId
        char channel_name[128] = {0};
        obj = cJSON_GetArrayItem(objs, i);
        if (cJSON_GetObjectItem(obj, "client_id") != NULL)
        {
            GET_JSON_VALUE_STRING(obj, "client_id", clientId);
        }
        if (strlen(clientId) == 0)
        {
            snprintf(clientId, sizeof(clientId), "%s_UP_%d", var->sn_str, i);
        }
        if (cJSON_GetObjectItem(obj, "name") != NULL)
        {
            GET_JSON_VALUE_STRING(obj, "name", channel_name);
        }
        if((strcmp(channel_name,"iec104")!=0) && (strcmp(channel_name,"modbus")!=0))
        {
            mqt_north_channel_t *channel = NULL;

            channel = &var->channel[var->channel_cnt++];
            channel->type = LINK_TYPE_MQTT;
            channel->connect_flag = false;
            mqtt_session_t *session = mqtt_session_new(clientId, (void*)channel);
            if (session == NULL)
            {
                ems_syslog(LOG_ERR, "malloc error");
                continue;
            }
            channel->session = session;
            set_channel_obj_from_json(obj,"name",channel->name,sizeof(channel->name));
            if((set_channel_obj_from_json(obj,"host",channel->addr,sizeof(channel->addr))!=0) || (set_channel_obj_from_json(obj,"user",channel->user,sizeof(channel->user))!=0) || (set_channel_obj_from_json(obj,"pass",channel->pass,sizeof(channel->pass))!=0))
            {
                ems_syslog(LOG_ERR, "get mqtt param error");
                goto out;
            }

            GET_JSON_VALUE_INT(obj, "port", channel->port);
            GET_JSON_VALUE_INT(obj, "lz4", channel->lz4);
            GET_JSON_VALUE_INT(obj, "index", channel->index);
            GET_JSON_VALUE_INT(obj, "enable", channel->enable);
            mqtt_session_set_address(session, channel->addr, channel->port, channel->user, channel->pass);
            mqtt_session_set_opts(session, DEFAULT_QOS, KEEPALIVEINTERVAL);
            ems_syslog(LOG_NOTICE, "mqtt north channel[%d], host:%s  port:%d  user:%s  pass:%s", i, channel->addr, channel->port, channel->user, channel->pass);
        }
        else
        {
            continue;
        }
    }

out:
    cJSON_Delete(root);

    return ret;
}


static int load_north_mqtt_cfg(mqt_north_var_t *var, const char *cfg_file)
{
    char *data = NULL;
    int ret = 0;

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

    data = read_file_data(cfg_file);
    if (data == NULL)
    {
        ems_syslog(LOG_ERR, "read cfg file %s error", cfg_file);
        ret = -1;
        goto out;
    }
	ems_syslog(LOG_INFO, "raw_data:%s", data);
    // new_data = calloc(strlen(var->sn_str) / strlen("[SN]") + 1, strlen(data));
    // replace_sub_str(data, "[SN]", var->sn_str, new_data);
    // free(data);
    // data = new_data;
	// ems_syslog(LOG_INFO, "replaced data:%s", data);

    ret = load_uplink_cfg(var, data);
    if (ret != 0)
    {
        ems_syslog(LOG_WARNING, "uplink config load fail");
        ret = 0;
    }

out:
    if (data != NULL)
    {
        free(data);
    }

    return ret;
}


static int load_north_dev_info(mqt_north_var_t *var, const char *cfg_file)
{
    char *data = NULL;
    cJSON *root = NULL;
    cJSON *class_a = NULL;
    cJSON *class_s = NULL;
	cJSON *devs = NULL;
    cJSON *dev = NULL;
	cJSON *tags = NULL;
	cJSON *tag = NULL;
    int class_cnt = 0, dev_cnt = 0, tag_cnt = 0;
    int i = 0, j = 0, ret = 0;

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

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

    root = cJSON_Parse(data);
    if (!root)
    {
        ems_syslog(LOG_ERR, "parse cfg file %s error", cfg_file);
        ret = -1;
        goto out;
    }
    class_a = cJSON_GetObjectItem(root, "dev_class");
    if (!class_a)
    {
        ems_syslog(LOG_ERR, "get north dev info failed");
        ret = -1;
        goto out;
    }

    class_cnt = cJSON_GetArraySize(class_a);
    for (i = 0; i < class_cnt; i++)
    {
        class_s = cJSON_GetArrayItem(class_a, i);
        if (class_s)
        {
            if (var->mqt_north_class[var->num] == NULL)
		        var->mqt_north_class[var->num] = calloc(sizeof(mqt_north_class_t), 1);
		    mqt_north_class_t *mqt_north_class = var->mqt_north_class[var->num];
				
            //mqt_north_class->class_type = calloc(32, sizeof(char));
            GET_JSON_VALUE_DY_STRING(class_s, "class_type", mqt_north_class->class_type);
            GET_JSON_VALUE_STRING(class_s, "name", mqt_north_class->name);
			mqt_north_class->interval = REPORT_INTERVAL; // 默认：300s
			ems_syslog(LOG_INFO, "class[%d] class_type:%s, name:%s, default interval:%d", i, mqt_north_class->class_type, mqt_north_class->name, mqt_north_class->interval);

			devs = cJSON_GetObjectItem(class_s, "devs");
			if (!devs){
				var->num++;
                continue;
			}
			
            dev_cnt = cJSON_GetArraySize(devs);
			for (j = 0; j < dev_cnt; j++)
			{
                dev = cJSON_GetArrayItem(devs, j);
				if (dev)
				{
				    if (mqt_north_class->mqt_north_dev[mqt_north_class->num] == NULL)
				        mqt_north_class->mqt_north_dev[mqt_north_class->num] = calloc(sizeof(mqt_north_dev_t), 1);
				    mqt_north_dev_t *mqt_north_dev = mqt_north_class->mqt_north_dev[mqt_north_class->num];
								
					GET_JSON_VALUE_STRING(dev, "no", mqt_north_dev->no);
					GET_JSON_VALUE_STRING(dev, "name", mqt_north_dev->disp_name);
					ems_syslog(LOG_INFO, "class[%d][%d], no:%s, name:%s", i, j, mqt_north_dev->no, mqt_north_dev->disp_name);

					tags = cJSON_GetObjectItem(dev, "tag_info");
					if (!tags){
						mqt_north_class->num++;
						continue;
					}
					
                    tag_cnt = cJSON_GetArraySize(tags);
					for (int m = 0; m < tag_cnt; m++)
					{
                        tag = cJSON_GetArrayItem(tags, m);
						if (tag)
						{
							cJSON *p0 = cJSON_GetArrayItem(tag, 0);
							cJSON *p1 = cJSON_GetArrayItem(tag, 1);
							cJSON *p2 = cJSON_GetArrayItem(tag, 2);
							if (!p0 || !p1 || !p2){
								ems_syslog(LOG_NOTICE, "invalid north tag :%d", m);
								continue;
							}

						    int property = get_property_by_string(p1->valuestring); //["ActiveEnergy", "realdata", "float"]
                            if (property == 0){       //>> 0 realdata;	1 system; 2 alarm
                                if (mqt_north_dev->realdata_tags[mqt_north_dev->realdata_num] == NULL)
				                     mqt_north_dev->realdata_tags[mqt_north_dev->realdata_num] = calloc(sizeof(mqt_north_tag_t), 1);
                                mqt_north_tag_t *mqt_north_tag = mqt_north_dev->realdata_tags[mqt_north_dev->realdata_num];

								snprintf(mqt_north_tag->name, TAG_NAME_LEN, "%s", p0->valuestring);
                                mqt_north_tag->property = property;
								mqt_north_tag->data_type = get_tag_type_by_string(p2->valuestring);
								
                                mqt_north_dev->realdata_num++;
							}else if (property == 2){

								if (mqt_north_dev->alarm_tags[mqt_north_dev->alarm_num] == NULL)
									 mqt_north_dev->alarm_tags[mqt_north_dev->alarm_num] = calloc(sizeof(mqt_north_tag_t), 1);
								mqt_north_tag_t *mqt_north_tag = mqt_north_dev->alarm_tags[mqt_north_dev->alarm_num];
								
								snprintf(mqt_north_tag->name, TAG_NAME_LEN, "%s", p0->valuestring);
								mqt_north_tag->property = property;
								mqt_north_tag->data_type = get_tag_type_by_string(p2->valuestring);
								
								mqt_north_dev->alarm_num++;
							}else
								; // what to do?  system
						}
					}	
					mqt_north_class->num++;
				}
			}
			var->num++;
        }
    }
	
out:
    cJSON_Delete(root);
    if (data != NULL)
    {
        free(data);
    }

    return ret;
}

static int north_mqtt_init(mqt_north_var_t *var)
{
    get_board_sn(var->sn_str);
	snprintf(var->cab_id, 256, "%s", cabinet_info.cab_id); // 站点编号
	snprintf(var->cab_station_name, 128, "%s", cabinet_info.cab_station_name);
    ems_syslog(LOG_INFO, "board SN:%s, Cab id:%s, Cab name:%s", var->sn_str, var->cab_id, var->cab_station_name);

	snprintf(var->protocol, 32, "%s", "V2.6");
	
    if (load_north_dev_info(var, N_MQTT_NORTH_MAP_FILE) == -1)
    {
        ems_syslog(LOG_CRIT,"[MQTT标准北向]:加载北向设备信息失败,文件名:%s请生成mqtt北向点表后重新启动程序",N_MQTT_NORTH_MAP_FILE);
        BUSINESS_LOG(BLOG_ERR, NORTH_ID_0003, NULL, "[北向]:加载北向设备信息失败,文件名:%s请生成mqtt北向点表后重新启动程序", N_MQTT_NORTH_MAP_FILE);
		return -1;
    }
    else
    {
        BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0003, NULL, "[北向]:加载北向mqtt北向点表设备信息成功,文件名:%s", N_MQTT_NORTH_MAP_FILE);
    }

    if (load_north_mqtt_cfg(var, N_MQTT_NORTH_CFG_FILE) < 0)
    {
        ems_syslog(LOG_CRIT,"[MQTT标准北向]:加载北向MQTT配置文件失败,文件名:%s,请检查配置文件格式后重新启动程序", N_MQTT_NORTH_CFG_FILE);
        BUSINESS_LOG(BLOG_ERR, NORTH_ID_0004, NULL, "[北向]:加载北向MQTT配置文件失败,文件名:%s请检查配置文件格式后重新启动程序", N_MQTT_NORTH_CFG_FILE);
        return -1;
    }
    else
    {
        BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0004, NULL, "[北向]:加载北向MQTT配置文件成功,文件名:%s", N_MQTT_NORTH_CFG_FILE);
    }

    north_mqtt_channel_create(var);

    var->status_poll_timer = my_timer_create();
    if (var->status_poll_timer <= 0)
    {
        ems_syslog(LOG_WARNING, "[ERR] create status check timer fail, %d", var->status_poll_timer);
        BUSINESS_LOG(BLOG_ERR, NORTH_ID_0006, NULL, "[北向]:北向mqtt创建轮询定时器失败");
    }

    if (var->status_poll_timer > 0)
    {
        my_timer_set(var->status_poll_timer, 1, 1000);
        BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0006, NULL, "[北向]:北向mqtt创建轮询定时器成功");
    }

    var->report_timer = my_timer_create();
    if (var->report_timer <= 0)
    {
        ems_syslog(LOG_WARNING, "[ERR] create status check timer fail, %d", var->report_timer);
        BUSINESS_LOG(BLOG_ERR, NORTH_ID_0007, NULL, "[北向]:北向mqtt创建上报周期定时器失败");
    }

    if (var->report_timer > 0)
    {
        my_timer_set(var->report_timer, 5, 10 * 1000);
        BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0007, NULL, "[北向]:北向mqtt创建上报周期定时器成功");
    }

	ems_syslog(LOG_INFO, "init done");

    return 0;
}

void change_mqtt_session(mqt_north_var_t *var)
{
    char *data = NULL;
    cJSON *root = NULL;
    data = read_file_data(N_MQTT_NORTH_CFG_FILE);
    if (data == NULL)
    {
        BUSINESS_LOG(BLOG_ERR, NORTH_ID_0021, NULL, "[北向]:read file [%s] error", N_MQTT_NORTH_CFG_FILE);
        ems_syslog(LOG_ERR, "read cfg file %s error", N_MQTT_NORTH_CFG_FILE);
        goto out;
    }
    else
    {
        BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0021, NULL, "[北向]:read file [%s] success", N_MQTT_NORTH_CFG_FILE);
    }

    cJSON *objs = NULL;
    cJSON *obj = NULL;
    int cnt = 0;
    int std_cnt = 0;
    int i = 0;
    bool new_channel = false;
    root = cJSON_Parse(data);
    if (!root)
    {
        ems_syslog(LOG_ERR, "parse cfg file error");
        goto out;
    }

    // MQTT
    objs = cJSON_GetObjectItem(root, "standard");
    if (!objs)
    {
        ems_syslog(LOG_ERR, "get MQTT standard failed");
        goto out;
    }
    cnt = cJSON_GetArraySize(objs);
    if (cnt == 0)
    {
        ems_syslog(LOG_ERR, "No MQTT UPLINK Channel");
        goto out;
    }
    for (i = 0; i < cnt; i++)
    {
        char channel_name[128] = {0};
        obj = cJSON_GetArrayItem(objs, i);
        if (cJSON_GetObjectItem(obj, "name") != NULL)
        {
            GET_JSON_VALUE_STRING(obj, "name", channel_name);
        }
        if((strcmp(channel_name,"iec104")!=0) && (strcmp(channel_name,"modbus")!=0))
        {
            std_cnt++;
        }
    }
    if(std_cnt != var->channel_cnt)         //说明增加了mqtt通道
    {
        var->channel_cnt = std_cnt;
        new_channel = true;
    }
    if(new_channel)
    {
        char clientId[128] = {0}; ///< clientId
        mqt_north_channel_t *channel = NULL;
        channel = &var->channel[var->channel_cnt-1];
        channel->type = LINK_TYPE_MQTT;
        channel->connect_flag = false;
        obj = cJSON_GetArrayItem(objs, var->channel_cnt+1); //实际为var->channel_cnt-1+2，+2用于跳过iec104和modbus
        if (cJSON_GetObjectItem(obj, "client_id") != NULL)
        {
            GET_JSON_VALUE_STRING(obj, "client_id", clientId);
        }
        if (strlen(clientId) == 0)
        {
            snprintf(clientId, sizeof(clientId), "%s_UP_%d", var->sn_str, var->channel_cnt-1);
        }

        mqtt_session_t *session = mqtt_session_new(clientId, (void*)channel);
        if (session == NULL)
        {
            ems_syslog(LOG_ERR, "malloc error");
            goto out;
        }
        channel->session = session;

        GET_JSON_VALUE_STRING(obj, "host", channel->addr);
        GET_JSON_VALUE_INT(obj, "port", channel->port);
        GET_JSON_VALUE_STRING(obj, "user", channel->user);
        GET_JSON_VALUE_STRING(obj, "pass", channel->pass);
		GET_JSON_VALUE_INT(obj, "lz4", channel->lz4);
        GET_JSON_VALUE_INT(obj, "enable", channel->enable);
        mqtt_session_set_address(session, channel->addr, channel->port, channel->user, channel->pass);
        mqtt_session_set_opts(session, DEFAULT_QOS, KEEPALIVEINTERVAL);
        if(channel->enable == 1)
        {
            channel->connect_flag = true;
            channel->var = var;
            mqtt_session_set_callbacks(channel->session, recv_north_cloud_mqtt_msg, cloud_north_mqtt_state_change);
            channel->comm_status = COMM_STATUS_CONNECTING;
            mqtt_session_start(channel->session);
            north_mqtt_subscribe(channel);
        }
        else
        {
            channel->var = var;
            mqtt_session_set_callbacks(channel->session, recv_north_cloud_mqtt_msg, cloud_north_mqtt_state_change);
            channel->comm_status = COMM_STATUS_CONNECTING;
            north_mqtt_subscribe(channel);
        }
    }
    else
    {
        for (i = 0; i < var->channel_cnt; i++)
        {
            mqt_north_channel_t *channel = NULL;

            channel = &var->channel[i];
            obj = cJSON_GetArrayItem(objs, i);
            GET_JSON_VALUE_INT(obj, "enable", channel->enable);
            if((channel->enable == 0) && channel->connect_flag)         //在初始化时连接，后续不向该平台推送
            {
                ems_syslog(LOG_ERR, "channel[%d] mqtt disconnect", i);
                channel->connect_flag = false;
                mqtt_session_disconn(channel->session);
            }
            else if((channel->enable == 1) && !(channel->connect_flag))     //需要向该平台推送，但之前取消过或者未连接过
            {
                channel->connect_flag = true;
                channel->comm_status = COMM_STATUS_CONNECTING;
                mqtt_session_start(channel->session);
            }
        }
    }

out:
    cJSON_Delete(root);
    if (data != NULL)
    {
        free(data);
    }
}

extern int reflush_all_plan(void);
void *north_all_plan_task(void *arg)
{
    int reflush_today = 0;
    north_get_all_plan_from_file();
	
    while (1)
    {
        sleep(1);
        time_t t = time(NULL);
        struct tm target_time = {};  
        localtime_r(&t, &target_time);
        struct tm *local_time = &target_time;
        if ((local_time->tm_hour == 0)&&(local_time->tm_min == 0))
        {
            if (reflush_today == 0)
            {
                north_g_reflush_all_plan = 1;
                reflush_today = 1;
            }
        }
        else
        {
            reflush_today = 0;
        }
        if (north_g_reflush_all_plan)
        {
            reflush_all_plan();
            north_g_reflush_all_plan = 0;
            server_add_new_log(LOG_CODE_REFLUSH_PLAN,"EMS",LOG_MODE_TRIG,1,NULL);
            get_pcs_ctrl_var()->config_save = true;
        }
    }
}
void north_all_plan_init()
{
    pthread_t pid;
    pthread_create(&pid, NULL, north_all_plan_task, NULL);
}

static void *north_mqtt_loop(void *arg)
{
	mqt_north_var_t *var = (mqt_north_var_t *)arg;

	north_all_plan_init();

    struct epoll_event ev, events[N_MAXEPOLLSIZE];
    int epollfd = epoll_create(N_MAXEPOLLSIZE);
    if(epollfd <= 0)
    {
        ems_syslog(LOG_NOTICE, "Epoll create error");
        BUSINESS_LOG(BLOG_ERR, NORTH_ID_0016, NULL, "[北向]:北向mqtt loop线程 Epoll create error");
        exit(EXIT_FAILURE);
    }
    else
    {
        BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0016, NULL, "[北向]:北向mqtt loop线程 Epoll create success");
    }

    ev.events = EPOLLIN | EPOLLET;
    ev.data.fd = var->status_poll_timer;
    if(epoll_ctl(epollfd, EPOLL_CTL_ADD, var->status_poll_timer, &ev))
    {
        ems_syslog(LOG_NOTICE, "epoll ctl failed");
        BUSINESS_LOG(BLOG_ERR, NORTH_ID_0017, NULL, "[北向]:北向mqtt loop线程 epoll ctl failed");
        exit(EXIT_FAILURE);
    }
    else
    {
        BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0017, NULL, "[北向]:北向mqtt loop线程 epoll ctl success");
    }
    ev.data.fd = var->report_timer;
    if(epoll_ctl(epollfd, EPOLL_CTL_ADD, var->report_timer, &ev))
    {
        ems_syslog(LOG_NOTICE, "epoll ctl failed");
        BUSINESS_LOG(BLOG_ERR, NORTH_ID_0018, NULL, "[北向]:北向mqtt loop线程 epoll ctl add failed");
        exit(EXIT_FAILURE);
    }
    else
    {
        BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0018, NULL, "[北向]:北向mqtt loop线程 epoll ctl add success");
    }

    while(1)
    {
        int i;
        int nfds = epoll_wait(epollfd, events, N_MAXEPOLLSIZE, 3000);
        if(nfds < 0)
        {
            ems_syslog(LOG_ERR, "error on epoll wait");
            BUSINESS_LOG(BLOG_ERR, NORTH_ID_0019, NULL, "[北向]:北向mqtt loop线程 error on epoll wait");
        }
        else if(nfds > 0)
        {
            BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0019, NULL, "[北向]:北向mqtt loop线程 epoll wait 正常");
            for (i = 0; i < nfds; ++i)
            {
                if (events[i].data.fd == var->status_poll_timer)
                {
					status_poll(var);
                    if(northcfg_change)
                    {
                        change_mqtt_session(var);
                        northcfg_change = false;
                    }
                }
				else if (events[i].data.fd == var->report_timer){
					report_data_poll(var);
					report_alarm(var);
                    if(cabinet_info.ctrl_mode == EMS_MODE_ZC)
                        lc_cabno_report(var);
				}
            }
        }
    }
    ems_syslog(LOG_ERR, "thread broken");

/*
	int ret = -1;
    struct pollfd pfds[2];

	north_all_plan_init();
	mqt_north_var_t *var = get_mqt_north_var();
    while (1)
    {
        pfds[0].fd = var->status_poll_timer;
        pfds[0].events = POLLIN;
        pfds[0].revents = 0;
        pfds[1].fd = var->report_timer;
        pfds[1].events = POLLIN;
        pfds[1].revents = 0;
        ret = poll(pfds, 0x2, 10 * 1000);
        if (ret < 0)
        {
            int error;
            error = errno;
            printf("errno %d\n", error);
            if (error == EINTR)
            {
                continue;
            }
            else
            {
                break;
            }
        }
        else if (ret > 0)
        {
            if (var->status_poll_timer > 0 && pfds[0].revents)
            {
                status_poll(var);
            }
            if (var->report_timer > 0 && pfds[1].revents)
            {
                report_data_poll(var);
				report_alarm(var);
            }
        }
    }*/
}

void north_mqtt_loop_init(mqt_north_var_t *var) 
{
    pthread_t tid = {0};
    if (pthread_create(&tid, NULL, north_mqtt_loop, var) != 0)
    {
        ems_syslog(LOG_NOTICE,"Failed to create north_mqtt_loop!");
        BUSINESS_LOG(BLOG_ERR, NORTH_ID_0015, NULL, "[北向]:北向mqtt loop线程 创建失败");
    }
    else
    {
        pthread_setname_np(tid, "north_mqtt_loop");
        BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0015, NULL, "[北向]:北向mqtt loop线程 创建成功");
    }
}


int common_north_mqtt_Init(void)
{
    mqt_north_var_t *p = get_mqt_north_var();
    memset(p, 0, sizeof(mqt_north_var_t));

    if (north_mqtt_init(p) < 0)
    {
        BUSINESS_LOG(BLOG_ERR, NORTH_ID_0002, NULL, "[北向]:北向mqtt任务初始化失败");
        return -1;
    }
    else
    {
        BUSINESS_LOG(BLOG_NOTICE, NORTH_ID_0002, NULL, "[北向]:北向mqtt任务初始化成功");
    }
		 
    north_mqtt_loop_init(p);

    return 0;
}