/***************************************************************************** * @copyright 1997-2050, . POWER SUPPLY CO., LTD. * @file bsp_mqttClient.c * @brief mqtt通信客户端程序 * @author mdy * @date 2024-09-29 * @remark *****************************************************************************/ #pragma once #include #include #include #include #include #include "MQTTClient.h" #include "MQTTAsync.h" #include #include "bsp_rtdb.h" #include "bsp_comm.h" #include #include #include "kit_db.h" #include "kit_core.h" #include "logic_bcu2bsu.h" #include "logic_main.h" #include "logic_dido.h" #define D_MAX_MQTT_DEV_POINT_NUM (2000) // 每类设备最大的上云测点个数 #define D_MAX_DEV_NUM (100) // 子设备最大数量 #define D_MAX_DEV_NAME_LEN (100) #define D_MAX_DEV_KEYWORD (11) #define D_MAX_IPV4_LENGTH (16) #define D_MAX_USERNAME_LENGTH (20) #define D_MAX_PASSWORD_LENGTH (20) #define QOS 1 #define TIMEOUT 10000L #define MQTT_ROOT_TOPIC_KEY "root" #define MQTT_PERIOD_TOPIC_KEY "period" #define MQTT_CHANGE_TOPIC_KEY "change" #define MQTT_HISTORY_TOPIC_KEY "history" #define MQTT_CONTROL_TOPIC_KEY "control" #define MQTT_READ_TOPIC_KEY "read" #define MQTT_REPLY_TOPIC_KEY "reply" #define MQTT_PAYLOAD_MAX_LEN (1024 * 1024) // 能源云限制报文长度最大为1MB #define MQTT_PERIOD_SEND_FAIL_WORDS "period send fail." // 周期发送失败 #define D_MAX_TOPIC_NUM (7) #define MQTT_STATUS_DONT_TRANS -100 // 还未执行完策略，当前的配置参数不上传 #define MQTT_STATUS_CFG_UPLOADED 100 // 策略配置传送完毕 #define MAX_MQTT_HISTORY_MSG_NUM 1000 #define D_PERIODSEND_T_SEQ (0) #define D_HISTORY_T_SEQ (1) #define D_DATACHK_T_SEQ (2) #define D_MAX_THREAD_NUM (3) #define D_STR_CONNECT_WORD "已连接" #define D_STR_DISCONN_WORD "已断开" #define RECORDNETSTATUS(word) kit_insert_lost_contact_record(word); // MQTT主题枚举定义 typedef enum { kEmsTopicType_Start, kEmsTopicType_root = kEmsTopicType_Start, // 初始化主题 kEmsTopicType_period, // 周期上报主题 kEmsTopicType_change, // 突变上报主题 kEmsTopicType_history, // 历史数据上报主题 kEmsTopicType_control, // 控制指令下发主题 kEmsTopicType_read, // 读参数主题 kEmsTopicType_reply_control, // 回复控制指令下发主题 kEmsTopicType_reply_read, // 回复读参数主题 kEmsTopicType_End = kEmsTopicType_reply_read, } ems_topic_type_e; // 数据变化事件分类 typedef enum { kEvent_topo_send = 1, // 拓扑发送完成 kEvent_has_history_record = 2, // 存在历史记录record kEvent_change_devTopo = 4, // 设备列表变化 kEvent_rename_anydev = 8, // 重命名子设备 kEvent_int_change = 16, // 子设备整型测点值变化越死区 kEvent_float_change = 32, // 子设备浮点型测点值变化越死区 kEvent_pointlistcfg_change = 64, // 设备测点配置改变 kEvent_Reconnect_success = 128, // MQTT重连成功 kEvent_Reconnect_failed = 256, // MQTT重连失败 kEvent_Period_Task_begin = 512, // 周期发送任务启动完成 } kEvent_ems_data; // EMS总的工作模式 typedef enum { KEms_cfg_Start, // 无 KEms_cfg_DEBUG, // 调试模式 KEms_cfg_PEAKVALLY, // 削峰填谷模式 KEms_cfg_DEMANDRES, // 需求响应模式 KEms_cfg_LOADTRACK, // 负载跟踪模式 KEms_cfg_DMDCTRL, // 需量控制 KEms_cfg_PFCTRL, // 功率因数 KEms_cfg_Protect, // 保护策略 KEms_cfg_END, } cfgType_e; typedef enum { kQos_0,//最多交付一次 kQos_1,//至少交付一次 kQos_2,//只交付一次 }Qos_e; // 拓扑结构结构体 typedef struct { int dbId; // 键：数据库中的 ID topology_t *topology; // 值：指向 topology_t 结构体的指针 UT_hash_handle hh; // uthash 处理句柄，使结构体可哈希 } topology_hash_entry_t; // MQTT单个测点的db查找表 typedef struct { // 取测点使用参数 rtdb_type_e dbType; // 数据库类型 uint16_t devType; // 设备类型 uint16_t devId; // 区分同类型不同设备实体的标识 uint16_t pointId; // 测点id up_dis_config_type_e txOrRx; // 是上传的测点还是接收云端指令0-仅上报 1-下发 bool whether; // 接收用户输入，1-上传：0-不上传 double value; // 该测点值的备份用于检测变化事件 bool ifchg; // 是否需要变化上送 } mqtt2db_t; typedef mqtt2db_t mqttGetDBElem; typedef mqtt2db_t mqttSetDBElem; void pointListener(mqtt2db_t *tab); // 一个设备实体的测点信息表结构体 typedef struct { mqttGetDBElem rtdbGetTab[D_MAX_MQTT_DEV_POINT_NUM]; // 实时数据库查找表,对应上传点位 mqttSetDBElem rtdbSetTab[D_MAX_MQTT_DEV_POINT_NUM]; // 实时数据库查找表，对应相应云端控制的下发点位,云端写RTDB操作只能操作该表中的点 bool bChgFlag; // 变化标志 int txcount; // 记录加入payload的测点数量 int rxcount; } payloadlist_t; // mqtt通信参数结构体 typedef struct { char stationName[100]; // 站点名称 char stationID[20]; // 站点id dev_type_e devType; // 设备类别 char *devName[D_MAX_DEV_NAME_LEN]; // 设备名称 char sn[D_MAX_DEV_NUM][60]; // 设备SN char identifier[60]; // 消息服务标识 int devNum; // 设备数量 payloadlist_t pllist[D_MAX_DEV_NUM]; // 该类设备下，每个设备实体一张mqtt2db关联表，依据该表中的元素操作RTDB bool bChgFlag; } mqtt_option_map_t; // payloadlists // 保留给回调函数的上下文结构体 typedef struct { MQTTClient client; MQTTClient_connectOptions connopts; mqtt_lib_t mlib; int modeword; } ConnContext; //所有算法配置的联合体 typedef union { peakvalley_zone_tab_t pkvly; // 削峰填谷参数 IN demandRes_params_t demand; // 需求响应参数 loadTrack_params_t loadTrack; // 负载跟踪参数 debug_params_t debug; // 调试参数 IN protect_params_t protect; // 保护参数 IN power_distr_t distr; // 功率分配参数 IN } logic_Params; //中断记录与历史数据数组关联的结构体 typedef struct { break_record_t record; // 中断记录 UT_array *dataArray; // 其关联的数据存储集合 } break_record_with_data_t; //从云端MQTT报文返回的信息结构体 typedef struct { char *transaction; int modeWord; } arvcfgInfo_ret_t; typedef void (*setCfgCB)(void *); void setdbgCB(void *arg); void setPkvlCB(void *arg); void setDmdRspCB(void *arg); void setLoadTrackCB(void *arg); void setDmdCtrlCB(void *arg); void setPFCtrlCB(void *arg); /***************************************************************************** * @brief 消息传送成功触发的回调函数 * @param[in] context: MQTT上下文结构体 * @param[in] dt: 标识符 * @return NONE *****************************************************************************/ void deliveredCB(void *context, MQTTClient_deliveryToken dt); /***************************************************************************** * @brief 消息接收成功触发的回调函数 * @param[in] context: MQTT上下文结构体 * @param[in] topicName:主题 * @param[in] topicLen：主题长度 * @return *****************************************************************************/ int msgarrvdCB(void *context, char *topicName, int topicLen, MQTTClient_message *message); /***************************************************************************** * @brief 连接丢失触发的回调函数 * @param[in] context: MQTT上下文结构体 * @param[in] cause:中断原因 * @return NONE *****************************************************************************/ void connlostCB(void *context, char *cause); /***************************************************************************** * @brief 更新MQTT参数 * @param[in] mlt: MQTT连接配置 * @param[in] map:测点配置表 * @return NONE *****************************************************************************/ void updateMqttPara(mqtt_lib_t *mlt, mqtt_option_map_t *map); /***************************************************************************** * @brief 更新MQTT参数 * @param[in] mlt: MQTT连接配置 * @param[in] map:测点配置表 * @return NONE *****************************************************************************/ int chkDataChange(mqtt_option_map_t *now, mqtt_option_map_t *old); /***************************************************************************** * @brief MQTT周期发送线程 * @param[in] arg: 可选参数 * @return NONE *****************************************************************************/ void *periodSendTask(void *arg); /***************************************************************************** * @brief 数据变化监视器 * @param[in] arg: 可选参数 * @return NONE ************************************************************************/ void *dataChgMonitor(void *arg); /***************************************************************************** * @brief 从MQTT字符串中获取关键字返回主题类型 * @param[in] topic: MQTT主题字符串 * @return 主题类型枚举值 *****************************************************************************/ int getKeywordsByTopic(char *topic); /***************************************************************************** * @brief 根据设备大类组包 * @param[in] payload：目标字符串指针 * @param[in] type：设备类型 * @param[in] tot：设备点表结构体 * @return NONE *****************************************************************************/ void genDevGrpPeriodPayload(char *payload, dev_type_e type, mqtt_option_map_t *tot); /***************************************************************************** * @brief 递归添加子节点到 JSON 结构中 * @param[in] parentArray: 父节点的 JSON 数组 * @param[in] parentId: 父节点的 ID * @param[in] topologyById: 存储拓扑信息的哈希表 * @return 无 *****************************************************************************/ void addChildrenToJson(cJSON *parentArray, int parentId, topology_hash_entry_t *topologyById); /***************************************************************************** * @brief 将数据库中的 EMS 设备拓扑结构数据转换为 JSON 字符串 * @param[out] json: 用于存储生成的 JSON 字符串的缓冲区。确保该缓冲区有足够的空间容纳生成的 JSON 字符串。 * @return 0-成功 1-失败 *****************************************************************************/ int getTopologyJsonByDb(char **json); /***************************************************************************** * @brief 将设备类型枚举值转换为与云端通信协议中的字符串 * @param[out] type：设备类型枚举值 * @return 设备类型缩写字符串 *****************************************************************************/ char *devTypeToString(dev_type_e type); /***************************************************************************** * @brief 比较两个设备节点是否相同 * @param[in] a：左操作数 * @param[in] b：左操作数 * @return 0-不同，1-相同 *****************************************************************************/ bool ifSameTopoElem(const topology_t *a, const topology_t *b); /***************************************************************************** * @brief 比较两个设备拓扑是否相同 * @param[in] arr1：左操作数 * @param[in] arr2：左操作数 * @return 0-不同，1-相同 *****************************************************************************/ bool chkTopoDiff(UT_array *arr1, UT_array *arr2); /***************************************************************************** * @brief 获得当前时间戳字符串 * @return 当前时间戳字符串 *****************************************************************************/ char *getTmstr2(); /***************************************************************************** * @brief 解析云端消息 * @param[in] jsonString：云端消息json串 * @param[in] arr：指向配置参数的指针 * @return 0-成功；1-失败 *****************************************************************************/ arvcfgInfo_ret_t parseEmsCfgJson(const char *jsonString, logic_Params *lP); /***************************************************************************** * @brief 单个设备测点值是否改变 * @param[in] tot: mqtt_lib_t中的主题字符串指针 * @return false-未改变 true-改变 *****************************************************************************/ bool ifSingledevPointsChg(payloadlist_t *list); /***************************************************************************** * @brief 单个设备测点值是否改变 * @param[in] map: 设备组测点列表 * @return false-未改变 true-改变 *****************************************************************************/ bool ifDevGrpPointsChg(mqtt_option_map_t *map); /***************************************************************************** * @brief 装载不同设备组变化上送主题有效载荷 * @param[in] topic: mqtt_lib_t中的主题字符串指针 * @param[in] payload:主题类型枚举，枚举与mqtt_lib_t中的成员应对应，请确保payload的缓冲区长度足够 * @return NONE *****************************************************************************/ int genChgDevGrpPeriodPayload(char *payload, dev_type_e type, mqtt_option_map_t *tot); /***************************************************************************** * @brief 根据云端下发的配置信息，将参数写入联合体 * @return 设备类型字符串 *****************************************************************************/ int parseStrategyJsonObject(cJSON *json, int modeword, logic_Params *lp); /***************************************************************************** * @brief 根据配置信息和模式组装成 JSON 字符串，并包含额外的顶层字段 * @param[in] arvcfgInfo_ret_t cfg 结构包含模式字、交易标识等信息 * void *arg 指向配置数据的指针（例如，指向 logic_Params 的指针） * @return char* 表示 JSON 字符串（需要调用者释放） *****************************************************************************/ char *createStrategyCfgJsonString(arvcfgInfo_ret_t cfg, const void *arg, int rw); /***************************************************************************** * @brief 判断通信是否断开 * @param[in] arg：tcp连接上下文 * @return 0-中断；1-正常 *****************************************************************************/ bool is_mqtt_connected(); /***************************************************************************** * @brief MQTT主函数 * @param[in] arg: 可选参数 * @return NONE *****************************************************************************/ void *MqttTask(void *arg); /***************************************************************************** * @brief MQTT转发任务入口 * @return NONE *****************************************************************************/ void creatNetMqttTaskEntry();