#include <stdio.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>

#include "cJSON.h"
#include "platmacro.h"
#include "mosquitto.h"
#include "bmser_mqtt.h"
#include "bmser_log.h"
#include "bmser_error_code.h"
#include "bmser_mqtt_upgrade.h"
#include "exchange_with_mqtt.h"
#include "bmser_dev.h"
#include "project_userdef.h"
#include "bmser_db_control_db.h"

#define CAN_DATA_HEAD_LEN               5
#define MAX_RACKID_NUM                  64
#define MAX_RACKID_ID_LEN               16
#define MAX_BRO_SHOW_VERSION_LEN        1024*10
#define BRO_SHOW_VER_WAIT_TIME_US       1 * 1000 * 1000
#define BAU_DEV_LEVEL                   3
#define BCU_DEV_LEVEL                   2
#define BMU_DEV_LEVEL                   1

#define HTTP_UPDATE_TOPIC_IDENTIFIER      "httpUpdateConfig"
#define HTTP_UPDATE_TOPIC                 "ipc/+/emg/+/alert/Set/httpUpdateConfig"                      /* 订阅http升级配置文件topic */


static const char *g_upgrade_topics[] = {
    DEV_RESET_TOPIC,
    DEV_VERSION_SHOW_TOPIC,
    DEV_UPGRADE_TOPIC,
    DEV_BINARY_CMD_TOPIC,
    DEV_BINARY_FLOW_TOPIC,
    DEV_UPGRADE_CFG_FILE_TOPIC,
    DEV_UPDATE_APP_TOPIC,
    DEV_UPDATE_TOPIC,
    DEV_RACKID_SET_MAUALLY_TOPIC,
    TOPIC_NAME_GET_UPGRADE_VERSION,
    DEV_BRO_GET_RACKID_TOPIC,
    HTTP_UPDATE_TOPIC,
    DEV_UPGRADE_CFG_FILE_TOPIC_V20,
    DEV_UPDATE_APP_TOPIC_V20,
    DEV_UPDATE_TOPIC_V20,
};

static const char *g_dev_layer_map[] = {"BMU", "BCU", "BAU"};
static exchange_with_mqtt_callback_s g_exchange_with_mqtt_cbs = {0x0};
int32_t g_mqttConnState = 0;

typedef struct
{
    bool bro_show_version[BAU_DEV_LEVEL];                                       //是否为广播读版本号
    int  bro_show_version_level[BAU_DEV_LEVEL];                                 //广播读版本号层级
    struct mosquitto* mosq;                                                     //mosq连接句柄
    char target_rackid[MAX_RACKID_ID_LEN + 1];                                  //目标簇id
    char bro_show_version_buf[BAU_DEV_LEVEL][MAX_BRO_SHOW_VERSION_LEN + 1];     //保存广播读版本号的信息
    char active_rackid[MAX_RACKID_NUM + 1];                                     //当前活跃的单簇id
} EXCHANGE_WITH_MQTT_STRU;

EXCHANGE_WITH_MQTT_STRU g_exchange_with_mqtt_mgr = {0};

int exchange_with_mqtt_session_publish(struct mosquitto* mosq, const char* cmd, const void* payload, int len);

/**
 * @description: add_rackid
 * @param {uint8_t} bcu_id
 * @return {*}
 */
static int add_rackid(uint8_t bcu_id)
{
    if (0 == bcu_id) {
        return -1;
    }

    EXCHANGE_WITH_MQTT_STRU *var = &g_exchange_with_mqtt_mgr;
    //at first check bcu_id exist
    int i = 0;
    for (i = 0; i < MAX_RACKID_NUM; i++) {
        if (var->active_rackid[i] == bcu_id) {
            return 0;
        }
    }

    //not find add to tail
    for (i = 0; i < MAX_RACKID_NUM; i++) {
        if (0 == var->active_rackid[i]) {
            var->active_rackid[i] = bcu_id;
            return 0;
        }
    }

    return -1;
}

/**
 * @description: add_version
 * @param {int} layer
 * @param {char} *ver
 * @return {*}
 */
static int add_version(int layer, char *ver)
{
    if (layer < 0 || layer > BAU_DEV_LEVEL || layer < BMU_DEV_LEVEL || NULL == ver) {
        return -1;
    }
    
    int len = 0;
    EXCHANGE_WITH_MQTT_STRU *var = &g_exchange_with_mqtt_mgr;
    layer--;
    //版本号去重
    if (strstr(var->bro_show_version_buf[layer], ver)) {
        return 0;
    }

    len = strlen(var->bro_show_version_buf[layer]);
    snprintf(var->bro_show_version_buf[layer] + len, sizeof(var->bro_show_version_buf[layer]) - len, "%s,", ver);

    return 0;
}

/**
 * @description: 广播读版本号线程函数
 * @param {void*} arg
 * @return {*}
 */
static void* brocast_show_version_proc(void* arg)
{
    int ret = 0, len = 0;
    static int cnt = 0;
    char in_buf[1024 + 128] = {0}, out_buf[1024 + 256] = {0};
    uint8_t level = 0;
    uint8_t* layer = (uint8_t*)arg;
    
    if (NULL == layer) {
        pthread_exit(NULL);   
    }
    
    level = *layer;
    if (level > BAU_DEV_LEVEL || level < BMU_DEV_LEVEL) {
        pthread_exit(NULL);
    }
    level--;
    
    while (1) {
        usleep(100);
        cnt++;
        if (cnt < BRO_SHOW_VER_WAIT_TIME_US / 100) {
            continue;
        } else {
            cnt = 0;
            break;
        }
    }

    EXCHANGE_WITH_MQTT_STRU *var = &g_exchange_with_mqtt_mgr;
    len = strlen(var->bro_show_version_buf[level]);
    if (len <= 0) {
        goto ERR_EXIT;
    }
    var->bro_show_version_buf[level][len - 1] = '\0';

    dy_syslog(LOG_DEBUG, "bro show version, layer:%s, buf:%s.\n", g_dev_layer_map[level], var->bro_show_version_buf[level]);
    snprintf(in_buf, sizeof(in_buf), "{\"frames\":{\"devLevel\":%d,\"version\":[%s]}}", level + 1, var->bro_show_version_buf[level]);
    ret = bmser_upgrade_status_pack(SHOW_VERSION, in_buf, out_buf, sizeof(out_buf));
    if (0 != ret) {
        goto ERR_EXIT;
    }

    ret = exchange_with_mqtt_session_publish(var->mosq, SHOW_VERSION, out_buf, strlen(out_buf));
    
ERR_EXIT:
    var->bro_show_version[level] = false;
    var->bro_show_version_level[level] = 0;
    memset(var->bro_show_version_buf[level], 0, sizeof(var->bro_show_version_buf[level]));
    pthread_exit(NULL);
}

/**
 * @description: 手动设置rackid线程函数
 * @param {void*} arg
 * @return {*}
 */
static void* set_rackid_proc(void* arg)
{
    int ret = 0;
    static int cnt = 0;
    char *p = (char *)arg;
    uint16_t target_id = LITTLE_TO_BIG_ENDIAN_16(atoi(p));
    EXCHANGE_WITH_MQTT_STRU *var = &g_exchange_with_mqtt_mgr;
    while (1) {
        if (var->active_rackid[0] > 0) {
            break;
        }

        usleep(100);
        cnt++;
        if (cnt < BRO_SHOW_VER_WAIT_TIME_US / 100) {
            continue;
        } else {
            cnt = 0;
            break;
        }
    }

    if (0 == var->active_rackid[0]) {
        return (void *)-1;
    }
    
    #if 0
    uint8_t send_buf[MAX_CMD_LINE*2] = {0};
    canrbus_xncan_header_t *p_payload = (canrbus_xncan_header_t *)send_buf;
    uint8_t len = sizeof(target_id);

    p_payload->dev_id = var->active_rackid[0];
    p_payload->func = 1;
    p_payload->layer = 2;
    p_payload->start_addr = LITTLE_TO_BIG_ENDIAN_16(0x0012);

    memcpy(p_payload->data, &target_id, len);
    len += sizeof(canrbus_xncan_header_t);

    ret = exchange_with_mqtt_session_publish(var->mosq, BINARY_FLOW, send_buf, len);
    #else
    BMS_PROTOCOL_INTF_REG_DATA_S regData = {0};
    regData.devId = var->active_rackid[0];
    regData.startAddr = 0x0012;
    regData.regCount = 1;
    regData.len = 2;
    regData.data = (uint8_t *)&target_id;
    ret = BmserProtocol_WriteRegisterDb(&regData);
    #endif
    dy_syslog(LOG_DEBUG, "set target id result = %d.", ret);
    //先认为mqtt成功就ok
    if (0 == ret) {
        /*
        ipc/21012C001969/CANBUS_0/device/21012C00196901/data_filtered/service/RackIdSetManually 
        {"identifier":"RackIdSetManually","time":1662887451,"tags":{"status":""},"mi":0,"sn":"21012C00196901"}
        */
        char out_buf[256] = {0};
        char tmp_buf[64] = {0};
        snprintf(tmp_buf, sizeof(tmp_buf), "%s", "{\"status\":\"\"}");
        ret = bmser_upgrade_status_pack(RACKID_SET_MANUALLY, tmp_buf, out_buf, sizeof(out_buf));
        if (0 != ret) {
            return (void *) -1;
        }
        ret = exchange_with_mqtt_session_publish(var->mosq, RACKID_SET_MANUALLY, out_buf, strlen(out_buf));
    }
    pthread_exit(NULL);
}

/**
 * @description: 广播获取rackid线程函数
 * @param {void*} arg
 * @return {*}
 */
static void* brocast_get_rackid_proc(void* arg)
{
    int i = 0, len = 0, ret = 0;
    char buf[128] = {0}, tmp_buf[128 + 16] = {0}, out_buf[256] = {0};
    static int cnt = 0;
    EXCHANGE_WITH_MQTT_STRU *var = &g_exchange_with_mqtt_mgr;
    
    while (1) {
        usleep(100);
        cnt++;
        if (cnt < BRO_SHOW_VER_WAIT_TIME_US / 100) {
            continue;
        } else {
            cnt = 0;
            break;
        }
    }
    buf[0] = '[';
    len = strlen(buf);
    for (i = 0; i < MAX_RACKID_NUM; i++) {
        if (var->active_rackid[i] != 0) {
            len += snprintf(buf + len, sizeof(buf) - len, "%d,", var->active_rackid[i]);
        } else {
            break;
        }
    }

    if (len <= 2) {
        dy_syslog(LOG_ERR, "less len, buf is %s", buf);
        return (void *) -1;
    }
    buf[len - 1] = ']';
    dy_syslog(LOG_DEBUG, "get active rackid, buf=%s.\n", buf);

    //{"identifier":"RackIdGetBroadcast","time":1662603066,"tags":{"status":[1,2]},"mi":0,"sn":"21012C00196901"}
    snprintf(tmp_buf, sizeof(tmp_buf), "{\"status\": %s}", buf);
    ret = bmser_upgrade_status_pack(BRO_GET_RACKIDS, tmp_buf, out_buf, sizeof(out_buf));
    if (0 != ret) {
        return (void *)-1;
    }
    ret = exchange_with_mqtt_session_publish(var->mosq, BRO_GET_RACKIDS, out_buf, strlen(out_buf));
    pthread_exit(NULL);
}

/**
 * @description: 通过mqtt消息发送CAN包 
 * @param {uint8_t*} buf 消息内容
 * @param {uint32_t} len 消息长度
 * @param {void*} param
 * @return {*}
 */
void exchange_with_mqtt_send_package(uint8_t* buf, uint32_t len, void* param)
{
    int result = 0;
    EXCHANGE_WITH_MQTT_STRU *var = &g_exchange_with_mqtt_mgr;
    result = exchange_with_mqtt_session_publish(var->mosq, BINARY_FLOW, buf, len);
    if (0 != result) {
        dy_syslog(LOG_ERR, "bmser mqtt publish fail, result = %d.", result);
    }
    return;
}

/**
 * @description: 通过MQTT消息上报升级状态
 * @param {char *} status 升级状态
 * @param {uint32_t} status_len 升级状态长度
 * @param {int} percent 升级百分百
 * @return {*}
 */
void exchange_with_mqtt_send_upgrade_status(const char * status, uint32_t status_len, int percent)
{
    if (NULL == status) {
        return;
    }

    int result = 0;
    char in_buf[128] = {0}, buf[512] = {0};
    EXCHANGE_WITH_MQTT_STRU *var = &g_exchange_with_mqtt_mgr;

    snprintf(in_buf, sizeof(in_buf), "{\"status\":\"%s\",\"percent\":%d}", status, percent);
    bmser_upgrade_status_pack(UPGRADE_PROCESS, in_buf, buf, sizeof(buf));
    
    result = exchange_with_mqtt_session_publish(var->mosq, UPGRADE_PROCESS, buf, strlen(buf));
    if (0 != result) {
        dy_syslog(LOG_ERR, "bmser mqtt publish upgrade status fail, result = %d.", result);
        return;
    }
    
    //dy_syslog(LOG_DEBUG, "bmser mqtt publish upgrade status, percent = %d.\n", percent);
    return;
}

/**
 * @description: 响应开始/停止升级请求
 * @param {char *} status 升级状态
 * @return {*}
 */
static void dev_upgrade_resp_mqtt_with_status(const char * status, uint32_t status_len)
{
    if (NULL == status || status_len > 64) {
        return;
    }

    int result = 0;
    char in_buf[128] = {0}, buf[512] = {0};
    EXCHANGE_WITH_MQTT_STRU *var = &g_exchange_with_mqtt_mgr;

    snprintf(in_buf, sizeof(in_buf), "{\"status\":\"%s\"}", status);
    bmser_upgrade_status_pack(DEV_UPGRADE, in_buf, buf, sizeof(buf));
    
    result = exchange_with_mqtt_session_publish(var->mosq, DEV_UPGRADE, buf, strlen(buf));
    if (0 != result) {
        dy_syslog(LOG_ERR, "bmser mqtt publish upgrade status fail, result = %d.", result);
        return;
    }

    return;
}

/**
 * @description: 通过MQTT消息解析ota信息
 * @param {OTA_Exchange_Stru} *ota ota升级信息
 * @param {bmser_mqtt_upgrade_message_s} *msg mqtt消息
 * @return {*}  0:success, others:fail.
 */
static int pack_ota_info_from_mqtt_msg(OTA_Exchange_Stru *ota, bmser_mqtt_upgrade_message_s *msg)
{
    if (NULL == ota || NULL == msg) {
        return -1;
    }
    
    bool brocast = false;   
    brocast = msg->is_broadcast;
    if (BAU_DEV_LEVEL == msg->devLevel) {
        //升级总控，广播暂不支持, 先按单升处理
        ota->bauId = msg->lay3;
        ota->bcuId = 0;
        ota->bmuId = 0;
        ota->dstLevel = 3;
    } else if (BCU_DEV_LEVEL == msg->devLevel) {
        //升级主控
        ota->bauId = msg->lay3;
        ota->bcuId = true == brocast ? 63 : msg->lay2;
        ota->bmuId = 0;
        ota->dstLevel = 2;
    } else if (BMU_DEV_LEVEL == msg->devLevel) {
        //升级从控,分两种情况: a,升级某一簇的从控, b,升级所有簇的从控
        ota->bauId = msg->lay3;
        ota->bcuId = msg->lay2 == 255 ? 63: msg->lay2; //这里如果升级所有的簇的从控, 需要设置值为63
        ota->bmuId = true == brocast ? 127 : msg->lay1;
        ota->dstLevel = 1;
    } else {
        return -1;
    }

    ota->rackNum = msg->rack_num;
    if (ota->dstLevel == 1 && ota->bcuId < 63) { //升级某一簇的从控
        ota->rackNum = 1;
    }
    if (strlen(msg->url) > 0) {
        snprintf((char *)ota->fileName, sizeof(ota->fileName), "%s", msg->url);
    }

    return 0;
}

static int bmser_upgrade_httpUpdateConfigProcess(const char *str)
{
    if (str == NULL)
    {
        return -1;
    }
    int ret = 0;
    cJSON *jsonObj = cJSON_Parse(str);
    cJSON *subJson = NULL;
    if (jsonObj == NULL)
    {
        return -1;
    }

    subJson = cJSON_GetObjectItem(jsonObj, "updateFile");
    if (subJson != NULL)
    {
        int array_size = cJSON_GetArraySize(subJson);
        for (int i = 0; i < array_size; i++) {
            cJSON *item = cJSON_GetArrayItem(subJson, i);
            cJSON *filePath = cJSON_GetObjectItem(item, "filePath");
            cJSON *dstPath = cJSON_GetObjectItem(item, "dstPath");
            char cmd[512] = {0};
            if (filePath != NULL && cJSON_IsString(filePath) && dstPath != NULL && cJSON_IsString(dstPath)) 
            {
                if(strstr(filePath->valuestring, ".ipk"))
                {
                    sprintf(cmd, "opkg install http://%s", filePath->valuestring);
                }
                else
                {
                    sprintf(cmd, "URL=%s && FILE_NAME=$(basename $URL) && wget -O %s/$FILE_NAME http://%s", filePath->valuestring,dstPath->valuestring,filePath->valuestring);
                }
                ret |= system(cmd);
            }
        }
    }
    cJSON_Delete(jsonObj);
    return ret;
}

/**
 * @description: 消息回调函数，收到订阅的消息后调用
 * @param {void*} userdata
 * @param {mosquitto_message*} message
 * @return {*} 0:success, others:fail.
 */
static int bmser_mqtt_message_callback(void* userdata, const struct mosquitto_message* message)
{
    if (NULL == message) {
        return -1;
    }
    
    long user_arg = 0;
    char version_info[128] = {0}, mount_dir[32] = {0};
    int pv = 0, mv = 0, sv = 0, cv = 0;
    char ver[64] = {0}, version_str[128] = {0}, buf[256] = {0};
    pthread_t p_bro_show_version[BAU_DEV_LEVEL];
    pthread_t p_bro_get_rack;
    pthread_t p_set_rack_id;
    int ret = 0, tid = 0;
    bmser_mqtt_upgrade_message_s msg = {0};
    OTA_Exchange_Stru ota = {0};
    EXCHANGE_WITH_MQTT_STRU *var = &g_exchange_with_mqtt_mgr;

    if (strstr(message->topic, DEV_UPGRADE)) {
        ret = bmser_upgrade_message_parse(message->payload, message->payloadlen, &msg);
        if (0 != ret) {
            return -1; 
        }
    
        if (0 == strcmp(msg.cmd, "stop")) {
            //停止升级
            if (g_exchange_with_mqtt_cbs.stop_upgrade != NULL) {
                g_exchange_with_mqtt_cbs.stop_upgrade();
                dev_upgrade_resp_mqtt_with_status(UPGRADE_STOP, strlen(UPGRADE_STOP));
            }

        } else if (0 == strcmp(msg.cmd, "start")) {
            if (0 != pack_ota_info_from_mqtt_msg(&ota, &msg)) {
                return -1;
            }
            dy_syslog(LOG_DEBUG, "bauId:%d, bcuId:%d, bmuId:%d, dstLevel:%d, rack_num:%d, fileName:%s.", ota.bauId, ota.bcuId, ota.bmuId, ota.dstLevel, ota.rackNum, ota.fileName);
            if (g_exchange_with_mqtt_cbs.start_upgrade != NULL) {
                //先响应设备升级报文
                dev_upgrade_resp_mqtt_with_status(UPGRADE_START, strlen(UPGRADE_START));
                exchange_with_mqtt_send_upgrade_status(UPGRADE_START, strlen(UPGRADE_START), 0);                          
                //开始升级, 当前reset参数暂时没有使用,升级完成后,默认重置BMS
                g_exchange_with_mqtt_cbs.start_upgrade(&ota);
            }
        }
    } else if (strstr(message->topic, BINARY_FLOW)) {
        //xieneng_can 接收到消息流响应
        bool isUpgrading = false;
        if (g_exchange_with_mqtt_cbs.message_callback != NULL) {
            g_exchange_with_mqtt_cbs.message_callback(message->payload, message->payloadlen, NULL, &isUpgrading);
        }
        
        // 处于升级状态时，不处理版本号上报(即不处理升级流程中用于判断设备READY而获取到的版本号信息)
        if(isUpgrading) {
            return 0;
        }
        canrbus_xncan_header_t *canrbus = (canrbus_xncan_header_t *)message->payload;
        if (NULL == canrbus->data || NULL == canrbus->data + CAN_DATA_HEAD_LEN) {
            return -1;
        }
        
        memcpy(ver, canrbus->data + CAN_DATA_HEAD_LEN, sizeof(ver));        
        ret = sscanf(ver, "pv = %d mv = %d sv = %d cv = %d", &pv, &mv, &sv, &cv);
        if (4 != ret) {
            //版本号解析失败, 可能是其他消息流,后续需要优化
            return ret;
        }

        if (canrbus->msgFlowBauId > 0 && 0 == canrbus->msgFlowBcuId && 0 == canrbus->msgFlowBmuId) {
            canrbus->layer = 3;
        } else if (canrbus->msgFlowBauId > 0 && canrbus->msgFlowBcuId > 0 && 0 == canrbus->msgFlowBmuId) {
            canrbus->layer = 2;
        } else if (canrbus->msgFlowBauId > 0 && canrbus->msgFlowBcuId > 0 && canrbus->msgFlowBmuId > 0) {
            canrbus->layer = 1;
        } else {
            return -1;
        }
        add_rackid(canrbus->msgFlowBcuId);
        dy_syslog(LOG_DEBUG, "get version, layer:%d, bau:%d, bcu:%d, bmu:%d, pv:%d, mv:%d, sv:%d, cv:%d.", canrbus->layer, canrbus->msgFlowBauId, canrbus->msgFlowBcuId, canrbus->msgFlowBmuId, pv, mv, sv, cv);
        if (false == var->bro_show_version[canrbus->layer - 1]) {
            //读单个设备的版本号,立即上报
            snprintf(version_str, sizeof(version_str), "{\"frames\":{\"devLevel\":%d,\"version\":[{\"devicePoint\":\"%d-%d-%d\",\"deviceVersion\":\"%d.%d.%d.%d\"}]}}", canrbus->layer, canrbus->msgFlowBauId, canrbus->msgFlowBcuId, canrbus->msgFlowBmuId, pv, mv, sv, cv);
            ret = bmser_upgrade_status_pack(SHOW_VERSION, version_str, buf, sizeof(buf));
            if (0 != ret) {
                return -1;
            }
            ret = exchange_with_mqtt_session_publish(var->mosq, SHOW_VERSION, buf, strlen(buf));
            if (0 != ret) {
                return -1;
            }
        } else {
            //广播读取版本号,先保存到数组中，延时上报
            if (canrbus->msgFlowBmuId == 127) {
                snprintf(version_info, sizeof(version_info), "{\"devicePoint\":\"%d-%d-%s\", \"deviceVersion\":\"%d.%d.%d.%d\"}", canrbus->msgFlowBauId, canrbus->msgFlowBcuId, "X", pv, mv, sv, cv);
            } else {
                snprintf(version_info, sizeof(version_info), "{\"devicePoint\":\"%d-%d-%d\", \"deviceVersion\":\"%d.%d.%d.%d\"}", canrbus->msgFlowBauId, canrbus->msgFlowBcuId, canrbus->msgFlowBmuId, pv, mv, sv, cv);
            }
            add_version(canrbus->layer, version_info);
        }
        
    } else if (strstr(message->topic, SHOW_VERSION)) {
        
        ret = bmser_upgrade_message_parse(message->payload, message->payloadlen, &msg);
        if (0 != ret) {
            return -1; 
        }

        if (0 != pack_ota_info_from_mqtt_msg(&ota, &msg)) {
            return -1;
        }
        dy_syslog(LOG_DEBUG, "bms show version, bauId:%d, bcuId:%d, bmuId:%d, dstLevel:%d.", ota.bauId, ota.bcuId, ota.bmuId, ota.dstLevel);
        
        if (NULL == g_exchange_with_mqtt_cbs.show_version) {
            return -1;   
        }
        g_exchange_with_mqtt_cbs.show_version(&ota);

        uint8_t level = ota.dstLevel;
        uint8_t idx = 0;
        if (level > BAU_DEV_LEVEL || level < BMU_DEV_LEVEL) {
            return -1;
        }

        idx = level - 1;
        var->bro_show_version_level[idx] = level;
        var->bro_show_version[idx] = msg.is_broadcast;
        
        if (true == var->bro_show_version[idx]) {
            memset(var->bro_show_version_buf[idx], 0, sizeof(var->bro_show_version_buf[idx]));
            tid = pthread_create(&p_bro_show_version[idx], NULL, brocast_show_version_proc, (void*)&(var->bro_show_version_level[idx]));
            if (tid < 0) {
                dy_syslog(LOG_ERR, "p_bro_show_version pthread create fail.");
                return -1;
            }
            pthread_detach(p_bro_show_version[idx]);
        }

    } else if (strstr(message->topic, DEV_RESET)) {
            ret = bmser_upgrade_message_parse(message->payload, message->payloadlen, &msg);
            if (0 != ret) {
                return -1; 
            }
            if (0 != pack_ota_info_from_mqtt_msg(&ota, &msg)) {
                return -1;
            }
            dy_syslog(LOG_DEBUG, "bms reset, bauId:%d, bcuId:%d, bmuId:%d, dstLevel:%d.", ota.bauId, ota.bcuId, ota.bmuId, ota.dstLevel);
            if (g_exchange_with_mqtt_cbs.dev_reset) {
                g_exchange_with_mqtt_cbs.dev_reset(&ota);
                //响应复位报文
                exchange_with_mqtt_session_publish(var->mosq, DEV_RESET, message->payload, message->payloadlen);
            }

    } else if (strstr(message->topic, BRO_GET_RACKIDS)) {
        dy_syslog(LOG_DEBUG, "brocast get rackId.");
        if (NULL == g_exchange_with_mqtt_cbs.show_version) {
            return -1;
        }
        memset(var->active_rackid, 0, sizeof(var->active_rackid));
        
        ota.bauId = 1;
        ota.bcuId = 0x3f;
        ota.bmuId = 0;
        ota.dstLevel = 2;
        g_exchange_with_mqtt_cbs.show_version(&ota);
        
        tid = pthread_create(&p_bro_get_rack, NULL, brocast_get_rackid_proc, (void*)&user_arg);
        if (tid < 0) {
            dy_syslog(LOG_ERR, "p_bro_get_rack pthread create fail");
            return -1;
        }
        pthread_detach(p_bro_get_rack);

    } else if (strstr(message->topic, UPDATE_CONFIG_TOPIC_IDENTIFIER)) {

        dy_syslog(LOG_DEBUG, "update config action.");
        system(DEV_UPGRADE_CFG_SHELL);

    } else if (strstr(message->topic, UPDATE_APP_TOPIC_IDENTIFIER)) {
        ret = bmser_upgrade_ipk_message_parse(message->payload, mount_dir, sizeof(mount_dir));
        if (0 != ret) {
            return -1;
        }

        char cmd[128] = {0};
        snprintf(cmd, sizeof(cmd), "sh %s/%s/%s %s", mount_dir, IPK_DIR, DEV_UPDATE_APP_SHELL, mount_dir);
        system(cmd);

    } else if (strstr(message->topic, UPDATE_DEV_TOPIC_IDENTIFIER)) {
        ret = bmser_upgrade_ipk_message_parse(message->payload, mount_dir, sizeof(mount_dir));
        if (0 != ret) {
            return -1;
        }

        char cmd[128] = {0};
        snprintf(cmd, sizeof(cmd), "sh %s/%s/%s %s", mount_dir, UPDATE_DIR, DEV_UPDATE_SHELL, mount_dir);
        system(cmd);
    }  else if (strstr(message->topic, HTTP_UPDATE_TOPIC_IDENTIFIER)) {
        char *payloadSucess = "update successful";
        char *payloadFail = "update failed";
        char sn[64] = {0},topic[256] = {0};
        if(bmser_dev_get_board_sn(sn) == 0)
        {
            snprintf(topic, sizeof(topic), "ipc/%s/emg/00000000/alert/SetResp/httpUpdateConfig", sn);
        }
        else
        {
            snprintf(topic, sizeof(topic), "ipc/0000000000000000/emg/00000000/alert/SetResp/httpUpdateConfig");
        }
        if(0 == bmser_upgrade_httpUpdateConfigProcess(message->payload))
        {
            bmser_mqtt_session_publish_raw(var->mosq, topic, payloadSucess, strlen(payloadSucess));
        }
        else
        {
            bmser_mqtt_session_publish_raw(var->mosq, topic, payloadFail, strlen(payloadFail));
        }
    }     
    else if (strstr(message->topic, RACKID_SET_MANUALLY)) {
        memset(var->target_rackid, 0, sizeof(var->target_rackid));
        ret = bmser_upgrade_parse_get_rackid(message->payload, message->payloadlen, var->target_rackid, sizeof(var->target_rackid));
        if (0 != ret) {
            return -1;
        }
        
        if (g_exchange_with_mqtt_cbs.show_version != NULL) {
            memset(var->active_rackid, 0, sizeof(var->active_rackid));
            
            ota.bauId = 1;
            ota.bcuId = 0x3f;
            ota.bmuId = 0;
            ota.dstLevel = 2;
            g_exchange_with_mqtt_cbs.show_version(&ota);

            tid = pthread_create(&p_set_rack_id, NULL, set_rackid_proc, (void*)var->target_rackid);
            if (tid < 0) {
                dy_syslog(LOG_ERR, "p_set_rack_id pthread create fail.");
                return -1;
            }
            pthread_detach(p_set_rack_id);
        }
    } else if (strstr(message->topic, TOPIC_SHOW_UPGRADE_VERSION)) {
        dy_syslog(LOG_DEBUG, "get upgrade version.");
        int ret = 0;
        char sn[64] = {0}, version[128] = {0}, topic[256] = {0}, payload[512] = {0};
        ret = bmser_dev_get_board_sn(sn);
        if (0 != ret) {
            return -1;
        }

        snprintf(version, sizeof(version), "{\"frames\":[{\"regs\":[{\"key\":\"upgrade_version\",\"value\":\"%d.%d.%d.%d\"}]}]}", UPGRADE_PROJECT_NUM,UPGRADE_MAIN_VERSION,UPGRADE_SUB_VERSION,UPGRADE_CORRECT_VERSION);
        ret = bmser_upgrade_status_pack(TOPIC_SHOW_UPGRADE_VERSION, version, payload, sizeof(payload));
        if (0 != ret) {
            return ret;
        }
        exchange_with_mqtt_session_publish(var->mosq, TOPIC_SHOW_UPGRADE_VERSION, payload, strlen(payload));
        return ret;
        
    }
    return 0;
}

/**
 * @description: 发布mqtt消息
 * @param {char*} cmd: 消息cmd类型
 * @param {const void*} payload: MQTT payload
 * @param {int} len: MQTT payload 长度
 * @return {*} 0:success, others:fail.
 */
int exchange_with_mqtt_session_publish(struct mosquitto* mosq, const char* cmd, const void* payload, int len)
{
	char topic[MQTT_TOPIC_MAX_LEN] = {0}, sn[SN_MAX_LEN] = {0};
	if (0 != bmser_dev_get_board_sn(sn)) {
		return -1;
	}
	if (!strcmp(cmd, SHOW_VERSION) || !strcmp(cmd, BRO_GET_RACKIDS) || !strcmp(cmd, TOPIC_SHOW_UPGRADE_VERSION)) {
		snprintf(topic, MQTT_TOPIC_MAX_LEN, "ipc/%s/bms/00000000/upgrade/GetResp/%s", sn, cmd);
		dy_syslog(LOG_DEBUG, "the top is %s", topic);
	} else if (!strcmp(cmd, UPGRADE_PROCESS)|| !strcmp(cmd, DEV_UPGRADE) || !strcmp(cmd, RACKID_SET_MANUALLY) || !strcmp(cmd, DEV_RESET)) {
		snprintf(topic, MQTT_TOPIC_MAX_LEN, "ipc/%s/bms/00000000/upgrade/SetResp/%s", sn, cmd);
	} else if (!strcmp(cmd, BINARY_FLOW)) {
		snprintf(topic, MQTT_TOPIC_MAX_LEN, "ipc/%s/bms/00000000/upgrade/Set/%s", sn, cmd);
	} else {
		dy_syslog(LOG_ERR, "not supported cmd = %s.", cmd);
		return -1;
	}

	return bmser_mqtt_session_publish_raw(mosq, topic, payload, len);
}

static int bmser_mqtt_connect_status_callback(int result)
{
    int idx = 0;
    OPRT_RET ret = 0;
    EXCHANGE_WITH_MQTT_STRU *var = &g_exchange_with_mqtt_mgr;

    dy_syslog(LOG_DEBUG, "xxxxxxxxx mqtt connect result: %d xxxxxxxxxxxx", result);
    if (result == 0 ) {
        g_mqttConnState = 1;

        for (idx = 0; idx < sizeof(g_upgrade_topics)/sizeof(g_upgrade_topics[0]); idx++) {
            dy_syslog(LOG_DEBUG, "xxxxxxxxx id = %d, topic = %sxxxxxxxxxxxxxxx", idx, g_upgrade_topics[idx]);
            ret = bmser_mqtt_session_subscribe(var->mosq, g_upgrade_topics[idx]);
            if (OPRT_OK != ret) {
                dy_syslog(LOG_ERR, "subscribe fail, ret = %d.", ret);
                return ret;
            }
        }
    }
    return OPRT_OK;
}

/**
 * @description: exchange_with_mqtt_init
 * @param {exchange_with_mqtt_callback_s} *cbs 回调函数
 * @param {void *} param
 * @return {*} 0:success, others:fail.
 */
int exchange_with_mqtt_init(exchange_with_mqtt_callback_s *cbs, void * param)
{
    if (NULL == cbs) {
        return -1;
    }

    memcpy(&g_exchange_with_mqtt_cbs, cbs, sizeof(g_exchange_with_mqtt_cbs));

    int ret = 0;
    char ip[16] = "localhost";
    int port = 1883;
    char client_id[16] = "upgrade_app";

    g_exchange_with_mqtt_mgr.mosq = bmser_mqtt_session_create(client_id, ip, port, bmser_mqtt_message_callback, bmser_mqtt_connect_status_callback);
    if (g_exchange_with_mqtt_mgr.mosq == NULL) {
        dy_syslog(LOG_DEBUG, "bmser_mqtt_session_create err.\n");
        return -1;
    }
    // 偶现mqtt连接失败，所以这里做了重连
    while (!g_mqttConnState)
    {
        sleep(1);
        mosquitto_reconnect(g_exchange_with_mqtt_mgr.mosq);
    }
    return g_exchange_with_mqtt_mgr.mosq == NULL ? -1 : 0;
}