#include <stdio.h>
#include <stdlib.h>

#include "ipc_session.h"
#include "dy_utils/dy_common.h"
#include "dy_utils/dy_ipc.h"
#include "cloud_emmsv2_common.h"
#include "dy_utils/protocol.h"
#include "dy_utils/dy_pp.h"


static void cloud_emmsv2_internal_subscribe_all(cloud_emmsv2_var_t *var)
{
    //ipc_session_t *ipc_session = var->session;
    char topic[TOPIC_MAX_LEN] = {0};

    snprintf(topic, TOPIC_MAX_LEN, "ipc/%s/lnxall/device/alarmReport", var->sn_str);
    ipc_session_subscribe(var->session, topic);

    snprintf(topic, TOPIC_MAX_LEN, "ipc/%s/lnxall/device/sysparam", var->sn_str);
    ipc_session_subscribe(var->session, topic);

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

static int cloud_emmsv2_internal_mqtt_handle_recv_msg(void *obj, ipc_msg_t *mqtt_msg)
{
    cloud_emmsv2_var_t *var = (cloud_emmsv2_var_t *)obj;

    if (strstr(mqtt_msg->topic, TOPIC_ALARM_REPORT))
    {
        dbg_syslog(LOG_ERR, "alarm received, topic %s, payload %s", mqtt_msg->topic, mqtt_msg->payload);
        cJSON *root = cJSON_Parse(mqtt_msg->payload);
        if (!root)
        {
            dbg_syslog(LOG_ERR, "parse payload fialed");
            return -1;
        }

        cJSON *alarmmsg = cJSON_GetObjectItem(root, "alarmmsg");
        cloud_emmsv2_pubalarm(var, alarmmsg);
        cJSON_Delete(root);
    }
    else if (strstr(mqtt_msg->topic, TOPIC_SYAPARAM_REPORT))
    {
        dbg_syslog(LOG_INFO, "sysparam changed, topic %s payload %s", mqtt_msg->topic, mqtt_msg->payload);
        cJSON *root = cJSON_Parse(mqtt_msg->payload);
        if (!root)
        {
            dbg_syslog(LOG_ERR, "parse payload failed");
            return -1;
        }

        cJSON *param = cJSON_GetObjectItem(root, "param");
        int paracnt = cJSON_GetArraySize(param);
        for(int i = 0; i < paracnt; i++)
        {
            cJSON *para = cJSON_GetArrayItem(param, i);
            if (para)
            {
                char paraname[MAX_ID_LEN] = {0};
                GET_JSON_VALUE_STRING(para, "name", paraname);
                if (strncmp(paraname, SYSPARA_ID, strlen(SYSPARA_ID)))
                {
                    char paraid[MAX_ID_LEN] = {0};
                    GET_JSON_VALUE_STRING(para, "value", paraid);
                    strncpy(var->lcid, paraid, strlen(paraid));
                }
                else if (strncmp(paraname, SYSPARA_CAP, strlen(SYSPARA_CAP)))
                {
                    int capacity = 0;
                    GET_JSON_VALUE_INT(para, "value", capacity);
                    var->capacity = capacity;
                }
                else if (strncmp(paraname, SYSPARA_POWER, strlen(SYSPARA_POWER)))
                {
                    int power = 0; 
                    GET_JSON_VALUE_INT(para, "value", power);
                    var->power = power;
                }
                else if (strncmp(paraname, SYSPARA_STAN, strlen(SYSPARA_STAN)))
                {
                    char parasta[MAX_STAN_LEN] = {0};
                    GET_JSON_VALUE_STRING(para, "value", parasta);
                    strncpy(var->staname, parasta, strlen(parasta));
                }
            }
        }

        cJSON_Delete(root);
    }
    else if (strstr(mqtt_msg->topic, TOPIC_PARAM_REPORT))
    {
        char cmd[CMD_MAX_LENGTH] = {0};
        cJSON *root = cJSON_Parse(mqtt_msg->payload);
        if (!root)
        {
            dbg_syslog(LOG_ERR, "parse payload failed");
            return -1;
        }

        strcat(cmd, "hset "REDIS_REPOPARAM_BASE" ");
        cJSON *tags = cJSON_GetObjectItem(root, "tags");
        cJSON *curtag = cJSON_GetArrayItem(tags, 0);
        do
        {
            char repdata[12] = {0};
            strcat(cmd, curtag->string);
            strcat(cmd, " ");
            if (curtag->valuedouble)
                snprintf(repdata, 12, "%lf", curtag->valuedouble);
            else
                snprintf(repdata, 12, "%d", curtag->valueint);
            strcat(cmd, repdata);
            strcat(cmd, " ");
            if (strstr(curtag->string, SET_CTRLMODE))
                var->controlMode = curtag->valueint;
            curtag = curtag->next;
        } while (curtag != NULL);
        
        redisReply *reply = (redisReply *)redisCommand(var->context, cmd);
        if (NULL == reply)
        {
            dbg_syslog(LOG_ERR, "write report param failed");
        }
        freeReplyObject(reply);
        cJSON_Delete(root);
    }

    return 0;
}

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

    snprintf(clientId, sizeof(clientId), "INT_DY_CLD_EMMSV2_%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, cloud_emmsv2_internal_mqtt_handle_recv_msg, NULL);
    cloud_emmsv2_internal_subscribe_all(var);
    ipc_session_start(var->session);
	return 0;
}