#include #include #include #include #include #include #include #include "ipc_session.h" #include "mqtt_session.h" #include "dy_utils/dy_common.h" #include "dy_utils/cJSON.h" #include "common.h" static node_cfg_t *find_node_by_data(shanghai_S5_var_t *var, unsigned char *buf, int buf_len) { char sn[SN_MAX_LEN] = {0}; node_cfg_t *node = NULL; int i = 0; if (buf_len < 18) { dy_syslog(LOG_WARNING, "data length too short"); return node; } snprintf(sn, SN_MAX_LEN, "%02X%02X%02X%02X%02X%02X%02X%02X", buf[13], buf[12], buf[11], buf[10], buf[9], buf[8], buf[7], buf[6]); for (i = 0; i < var->nodes_cfg_table->node_cnt; i++) { if (strcmp(sn, var->nodes_cfg_table->node[i].sn) == 0 && strcmp(var->nodes_cfg_table->node[i].app_key, APP_KEY) == 0) { node = &var->nodes_cfg_table->node[i]; break; } } dy_syslog_hex(LOG_DEBUG, buf, buf_len, "buf content, sn:%s node:%p", sn, node); return node; } static int __send_data_PP(shanghai_S5_var_t *var, node_cfg_t *node, char *buf, int len) { pp_real_time_data_t real_data = {0}; regulate_cmd_backup_t *fields = NULL; time_t now = time(NULL); //restore the identifier { char key_str[64] = {0}; data_info_t *data_info = NULL; sprintf(key_str, "%s", node->sn); data_info = var->identifier_backup.get(&var->identifier_backup, key_str); if (data_info) { fields = (regulate_cmd_backup_t *)data_info->data; } } if (fields != NULL && ((now - fields->ts) < 10)) { real_data.mi = fields->mi; strncpy(real_data.src_identifier, fields->src_identifier, sizeof(real_data.src_identifier)); } real_data.ts = now; real_data.len = len; real_data.data = malloc(real_data.len); strcpy(real_data.sn, node->sn); memcpy(real_data.data, buf, real_data.len); send_to_proto_parser_by_app_key(var->session_int, &real_data, node->app_key); free(real_data.data); flush_device_status_by_sn(var->ds, ACTION_RCV, node->sn); return 0; } static int send_confirm(shanghai_S5_var_t *var, node_cfg_t *node, int PRseq, int con) { int FN; int PN = 0; char buf[32] = {0}; unsigned int tmp[8] = {0}; int da1, da2, dt1, dt2; int da1_bit, dt1_bit; int len = 14; if (con) FN = 1; else FN = 0; dt1_bit = (FN - 1) % 8; dt1 = 1 << dt1_bit; dt2 = (FN - 1) / 8; if (PN == 0) { da1 = 0; da2 = 0; } else { da1_bit = (PN - 1) % 8; da1 = 1 << da1_bit; da2 = (PN - 1) / 8 + 1; } int afn = 0; int seq = (1 << 6) | (1 << 5) | (PRseq & 0xF); buf[0] = 0x68; *((unsigned short *)&buf[1]) = len; *((unsigned short *)&buf[3]) = len; buf[5] = 0x68; sscanf(node->sn, "%02X%02X%02X%02X%02X%02X%02X%02X", &tmp[0], &tmp[1], &tmp[2], &tmp[3], &tmp[4], &tmp[5], &tmp[6], &tmp[7]); buf[13] = tmp[0]; buf[12] = tmp[1]; buf[11] = tmp[2]; buf[10] = tmp[3]; buf[9] = tmp[4]; buf[8] = tmp[5]; buf[7] = tmp[6]; buf[6] = tmp[7]; buf[14] = afn; buf[15] = seq; buf[16] = da1; buf[17] = da2; buf[18] = dt1; buf[19] = dt2; unsigned short crc = crc16(buf + 6, 14); buf[20] = (crc >> 8) & 0x00FF; buf[21] = crc & 0x00FF; buf[22] = 0x16; dy_syslog_hex(LOG_DEBUG, buf, 23, "confirm buf content, sn:%s", node->sn); send_data_to_front(var, node, buf, 23); return 0; } static int process_data(shanghai_S5_var_t *var, node_cfg_t *node, char *buf, int len) { int afn = buf[14]; int seq = buf[15]; int dir = (afn >> 7) & 1; int acd = (afn >> 5) & 1; int code = (afn & 0x1F); int con = (seq >> 4) & 1; int PRseq = seq & 0xF; dy_syslog(LOG_DEBUG, "dir:%d con:%d\n", dir, con); if (dir == 1 && con == 1) { // send confirm pkt send_confirm(var, node, PRseq, 1); } __send_data_PP(var, node, buf, len); } /** * 接收来自前置机的MQTT message * 并按照接收的数据内容进行处理，当需要返回数据时，将待返回的数据加入到list里面，由poll函数来发送 * 当收到需要通过SN的请求时，立即发送SN数据 * @param[in] var 模块变量部分. * @param[in] mqtt_msg 收到的MQTT消息内容，包含topic和payload. * @retval 成功 * @par 标识符 * 保留 * @par 其它 * 无 * @par 修改日志 * zhaoxianpeng于2019-10-28创建 */ static int handle_mqtt_from_front_msg(shanghai_S5_var_t *var, mqtt_message_t *mqtt_msg) { cJSON *root = NULL; uint8_t *buf = NULL; int port, socket; char *ip = NULL; char *client_id = NULL; char *payload = NULL; int len = 0; int ack_len = 0; char topic[TOPIC_MAX_LEN] = {0}; char *encode_str = NULL; char *data_tmp = NULL; char type[12]; if (mqtt_msg == NULL) { return -1; } root = cJSON_Parse(mqtt_msg->payload); if (root == NULL) { dy_syslog(LOG_ERR, "json parse error"); return -1; } GET_JSON_VALUE_STRING(root, "type", type); if (strcmp(type, "TCP") != 0) { dy_syslog(LOG_WARNING, "Not TCP data:%s", type); goto out; } GET_JSON_VALUE_DY_STRING(root, "payload", payload); buf = (char *)malloc(MAXBUF); len = b64_decode(payload, buf, MAXBUF); if (buf[0] == 0x68 && buf[len - 1] == 0x16) { // 上海S5协议 int ret; node_cfg_t *node = NULL; node = find_node_by_data(var, buf, len); if (node != NULL) { if (strcmp(mqtt_msg->topic, "F/SearchSN") == 0) { cJSON *rsp_sn = cJSON_CreateObject(); GET_JSON_VALUE_INT(root, "port", port); GET_JSON_VALUE_INT(root, "socket", socket); GET_JSON_VALUE_DY_STRING(root, "ip", ip); GET_JSON_VALUE_DY_STRING(root, "clientId", client_id); sprintf(topic, "GW/%s/RspSearchSN", client_id); cJSON_AddStringToObject(rsp_sn, "sn", node->sn); cJSON_AddNumberToObject(rsp_sn, "socket", socket); cJSON_AddStringToObject(rsp_sn, "ip", ip); cJSON_AddNumberToObject(rsp_sn, "port", port); // 以字符串的方式发送数据 data_tmp = cJSON_Print(rsp_sn); mqtt_session_publish(var->session_ext, topic, data_tmp, strlen(data_tmp)); cJSON_Delete(rsp_sn); free(ip); free(client_id); free(data_tmp); } process_data(var, node, buf, len); } } out: free(payload); free(buf); cJSON_Delete(root); return 0; } int external_mqtt_handle_recv_msg(void *obj, mqtt_message_t *mqtt_msg) { shanghai_S5_var_t *var = (shanghai_S5_var_t *)obj; char *topic = mqtt_msg->topic; //dy_syslog(LOG_DEBUG, " MQTT client:%s received MQTT topic:%s payload length:%d", // session->client.clientId, mqtt_msg->topic, mqtt_msg->payloadLen); if (mqtt_msg->topic == NULL) { dy_syslog(LOG_ERR, "topic is NULL"); goto out; } if (mqtt_msg->topic[0] == 'F' && topic[1] == '/') { // 前置机的topic handle_mqtt_from_front_msg(var, mqtt_msg); goto out; } out: return 0; }