/*================================================================ Copyright (c) 2021, Quectel Wireless Solutions Co., Ltd. All rights reserved. Quectel Wireless Solutions Proprietary and Confidential. =================================================================*/ #include "MQTTClient.h" #include "mqtt_connect_diag.h" #include "ql_data_call.h" #include "ql_application.h" #include "sockets.h" #include "app_inc.h" #include "cjson.h" #include "ql_fs.h" #include "ql_fota.h" #include "ql_sim.h" #include "ql_ping.h" #include "ql_nw.h" #include "ql_atcmd.h" #include "cJSON.h" #include "build_macro.h" #include "ql_log.h" #define PROFILE_IDX 1 //#define DEFAULT_SERVER_DOMAIN "175.24.73.253" #define DEFAULT_SERVER_DOMAIN "banban.api.talkingq.com" #define DEFAULT_SERVER_CLIENT_USER "talkingq" #define DEFAULT_SERVER_CLIENT_PASS "talkingq@2026" #define DEFAULT_SERVER_CLIENT_ID "9376265c059e474e92c2e46208ae1dc5" #define DEFAULT_SERVER_PORT 1883 #define DEFAULT_MQTT_TICK_TIME 10 #define TYPE_COMMAND 1 #define TYPE_RESPONE 2 #define DEFAULT_MQTT_EVENT_TIME 5000 #define OTA_STATE_FILE "U:/ota_state.bin" #define OTA_STATE_TMP_FILE "U:/ota_state.tmp" #define OTA_STATE_MAGIC 0x4F544131U #define OTA_STATE_FORMAT_VERSION 1U #define OTA_PROMPT_PLAY_TIMEOUT_MS 10000 #define OTA_PROMPT_POLL_MS 50 #define OTA_ALLOWED_URL_PREFIX "http://101.35.224.118:8080/assets/firmware/" static ql_timer_t mqtt_event_task_timer = NULL; static char event_register[128] = {0}; static char event_resp_register[128] = {0}; static char command_topicName[128] = {0};//服务器下发 static char respone_topicName[128] = {0};//设备返回给服务器 static char event_topicName[128] = {0};//设备上报给服务器 static char event_resp_topicName[128] = {0};//服务器返回 static MQTTClient client = {0}; static Network network = {0}; static unsigned char sendbuf[4096] = {0}, readbuf[4096] = {0}; static volatile char mqtt_connect = 0; static volatile char mqtt_public_error = 0; // 安全修复：添加互斥锁保护MQTT连接状态 static ql_mutex_t g_mqtt_conn_mutex = NULL; #define MQTT_CONN_LOCK() do{ if(g_mqtt_conn_mutex) ql_rtos_mutex_lock(g_mqtt_conn_mutex, QL_WAIT_FOREVER); }while(0) #define MQTT_CONN_UNLOCK() do{ if(g_mqtt_conn_mutex) ql_rtos_mutex_unlock(g_mqtt_conn_mutex); }while(0) static char imei_buf[32]={0}; static char device_id[32]={0}; static char mqtt_connecting_flag = 0; ql_queue_t mqtt_work_queue_monitor = NULL; //MQTT监控 //static uint8_t g_mqtt_public_mutex = 0; static ql_mutex_t g_mqtt_public_mutex = NULL; int g_nfc_bind_status = 0; extern u8 g_cur_volume; extern u8 g_pre_volume; extern unsigned short g_vbat_vol; extern bat_level_t battery_get_level(void); extern sleep_config_t g_sleep_cfg; extern uint8_t g_sleep_config_valid; extern float g_lon; extern float g_lat; extern unsigned char battery_first_update_flag; int sub_topic_status = 0; /*uint8_t get_mqtt_public_mutex(void) { return g_mqtt_public_mutex; }*/ int app_mqtt_connect_sem(char flag); // 安全修复：增大payload缓冲区并定义为宏便于管理 #define MQTT_PAYLOAD_SIZE 2048 static char payload[MQTT_PAYLOAD_SIZE] = {0}; extern int fota_start; int app_play_ota_tips_file(void); int xiaoice_get_play_counter(void); typedef struct { uint32_t magic; uint16_t format_version; uint16_t reserved; char previous_version[16]; char target_version[16]; uint32_t checksum; } ota_state_t; int app_mqtt_publish(char *buf, int len, int type); void build_register(void); static void ota_report_pending_success(void); enum{ MQTT_WORK_CONNECT, MQTT_WORK_RECV, }; int get_mqtt_connect_status(void) { int status; MQTT_CONN_LOCK(); status = mqtt_connect; MQTT_CONN_UNLOCK(); return status; } static uint32_t ota_state_checksum(const ota_state_t *state) { const uint8_t *data = (const uint8_t *)state; uint32_t hash = 2166136261U; size_t i; for (i = 0; i < sizeof(*state) - sizeof(state->checksum); i++) { hash ^= data[i]; hash *= 16777619U; } return hash; } static int ota_state_save(const char *previous_version, const char *target_version) { ota_state_t state; QFILE *fp; int ret; if (previous_version == NULL || target_version == NULL) { return -1; } memset(&state, 0, sizeof(state)); state.magic = OTA_STATE_MAGIC; state.format_version = OTA_STATE_FORMAT_VERSION; strncpy(state.previous_version, previous_version, sizeof(state.previous_version) - 1); strncpy(state.target_version, target_version, sizeof(state.target_version) - 1); state.checksum = ota_state_checksum(&state); fp = ql_fopen(OTA_STATE_TMP_FILE, "wb+"); if (fp == NULL) { RLOGE("OTA state open failed: %s", OTA_STATE_TMP_FILE); return -1; } ret = ql_fwrite(&state, sizeof(state), 1, fp); if (ret != (int)sizeof(state)) { RLOGE("OTA state write failed: %d", ret); ql_fclose(fp); ql_remove(OTA_STATE_TMP_FILE); return -1; } ret = ql_fsync(fp); ql_fclose(fp); if (ret != 0) { RLOGE("OTA state sync failed: %d", ret); ql_remove(OTA_STATE_TMP_FILE); return -1; } ret = ql_rename(OTA_STATE_TMP_FILE, OTA_STATE_FILE); if (ret != 0) { RLOGE("OTA state rename failed: %d", ret); ql_remove(OTA_STATE_TMP_FILE); return -1; } return 0; } static int ota_state_load(ota_state_t *state) { QFILE *fp; int ret; uint32_t checksum; if (state == NULL || ql_access(OTA_STATE_FILE, 0) != 0) { return -1; } fp = ql_fopen(OTA_STATE_FILE, "rb"); if (fp == NULL) { return -1; } memset(state, 0, sizeof(*state)); ret = ql_fread(state, sizeof(*state), 1, fp); ql_fclose(fp); if (ret != (int)sizeof(*state)) { RLOGE("OTA state read failed: %d", ret); return -1; } checksum = state->checksum; state->checksum = 0; if (state->magic != OTA_STATE_MAGIC || state->format_version != OTA_STATE_FORMAT_VERSION || checksum != ota_state_checksum(state)) { RLOGE("OTA state validation failed"); return -1; } state->checksum = checksum; state->previous_version[sizeof(state->previous_version) - 1] = '\0'; state->target_version[sizeof(state->target_version) - 1] = '\0'; return 0; } static void ota_state_clear(void) { ql_remove(OTA_STATE_TMP_FILE); if (ql_access(OTA_STATE_FILE, 0) == 0) { ql_remove(OTA_STATE_FILE); } } static int ota_parse_version(const char *version, unsigned int parts[3]) { char tail; if (version == NULL || version[0] == '\0') { return -1; } if (sscanf(version, "%u.%u.%u%c", &parts[0], &parts[1], &parts[2], &tail) != 3) { return -1; } return 0; } static int ota_compare_version(const char *left, const char *right, int *result) { unsigned int left_parts[3]; unsigned int right_parts[3]; int i; if (result == NULL || ota_parse_version(left, left_parts) != 0 || ota_parse_version(right, right_parts) != 0) { return -1; } *result = 0; for (i = 0; i < 3; i++) { if (left_parts[i] > right_parts[i]) { *result = 1; break; } if (left_parts[i] < right_parts[i]) { *result = -1; break; } } return 0; } static int ota_url_is_allowed(const char *url) { size_t prefix_len = strlen(OTA_ALLOWED_URL_PREFIX); if (url == NULL || url[0] == '\0') { return 0; } if (strncmp(url, OTA_ALLOWED_URL_PREFIX, prefix_len) != 0) { return 0; } if (url[prefix_len] == '\0' || strstr(url + prefix_len, "..") != NULL) { return 0; } return strstr(url + prefix_len, ".bin") != NULL; } /* ==================== 解析函数 ==================== */ /** * @brief 解析服务器下发的命令 */ int parse_command(const char *json_str, command_t *cmd) { cJSON *root = cJSON_Parse(json_str); if (root == NULL) { const char *error_ptr = cJSON_GetErrorPtr(); if (error_ptr != NULL) { printf("JSON Parse Error: %s", error_ptr); } return -1; } // 解析 msg_id cJSON *msg_id = cJSON_GetObjectItem(root, "msg_id"); if (cJSON_IsString(msg_id) && msg_id->valuestring != NULL) { strncpy(cmd->msg_id, msg_id->valuestring, sizeof(cmd->msg_id) - 1); } // 解析 time (UTC 时间戳，字符串格式) cJSON *time_obj = cJSON_GetObjectItem(root, "time"); if (cJSON_IsString(time_obj) && time_obj->valuestring != NULL) { strncpy(cmd->time, time_obj->valuestring, sizeof(cmd->time) - 1); cmd->time[sizeof(cmd->time) - 1] = '\0'; // 确保终止 printf("server time: %s\r\n", cmd->time); } else if (cJSON_IsNumber(time_obj)) { // 兼容数字格式，转为字符串 snprintf(cmd->time, sizeof(cmd->time), "%.0f", time_obj->valuedouble); printf("server time: %s\r\n", cmd->time); } else { cmd->time[0] = '\0'; // 空字符串表示未设置 } // 根据 msg_id 判断消息类型 int msg_type = atoi(cmd->msg_id); cmd->type = (msg_type >= 0 && msg_type <= 12) ? msg_type : -1; printf("cmd->type:%d\r\n", cmd->type); if(cmd->type==MSG_DEVICE_STATUS){ cJSON *type_obj = cJSON_GetObjectItem(root, "type"); if(type_obj!=NULL){ if (cJSON_IsNumber(type_obj)) { cmd->sub_type = type_obj->valueint; printf("device status type: %d\r\n", cmd->sub_type); } } } // 解析 params cJSON *params = cJSON_GetObjectItem(root, "params"); if (cmd->type == MSG_OTA_UPGRADE && params == NULL) { RLOGE("OTA command has no params"); cJSON_Delete(root); return -1; } if (params != NULL) { switch (cmd->type) { case MSG_GPS_QUERY: printf("MSG_GPS_QUERY\r\n"); /*返回当前的GPS坐标信息*/ break; case MSG_VOLUME_SET: { /*设置音量大小*/ printf("MSG_VOLUME_SET\r\n"); cJSON *level = cJSON_GetObjectItem(params, "level"); if (cJSON_IsNumber(level)) { cmd->params.volume_level = level->valueint; } break; } case MSG_OTA_UPGRADE: { printf("MSG_OTA_UPGRADE\r\n"); cJSON *url = cJSON_GetObjectItem(params, "url"); cJSON *version = cJSON_GetObjectItem(params, "version"); if (!cJSON_IsString(url) || url->valuestring == NULL || !cJSON_IsString(version) || version->valuestring == NULL || url->valuestring[0] == '\0' || version->valuestring[0] == '\0' || strlen(url->valuestring) >= sizeof(cmd->params.ota.url) || strlen(version->valuestring) >= sizeof(cmd->params.ota.version)) { RLOGE("invalid OTA command parameters"); cJSON_Delete(root); return -1; } strncpy(cmd->params.ota.url, url->valuestring, sizeof(cmd->params.ota.url) - 1); strncpy(cmd->params.ota.version, version->valuestring, sizeof(cmd->params.ota.version) - 1); printf("ota.url:%s\r\n", cmd->params.ota.url); printf("ota.version:%s\r\n", cmd->params.ota.version); break; } case MSG_NFC_NOTICE: { /*NFC下发留言提醒*/ printf("MSG_NFC_NOTICE\r\n"); cJSON *url = cJSON_GetObjectItem(params, "url"); if (cJSON_IsString(url)) { strncpy(cmd->params.nfc.url, url->valuestring, sizeof(cmd->params.nfc.url) - 1); } break; } case MSG_NFC_BIND: { /*NFC下发绑定*/ printf("MSG_NFC_BIND\r\n"); cJSON *url = cJSON_GetObjectItem(params, "url"); if (cJSON_IsString(url)) { strncpy(cmd->params.bind.url, url->valuestring, sizeof(cmd->params.bind.url) - 1); } break; } case MSG_DEVICE_STATUS: { /*查询或者设置设备状态*/ break; } case MSG_DEVICE_SLEEP: { /*设置设备休眠时间*/ printf("MSG_DEVICE_SLEEP\r\n"); cJSON *start = cJSON_GetObjectItem(params, "start"); cJSON *end = cJSON_GetObjectItem(params, "end"); if (cJSON_IsString(start) && start->valuestring != NULL) { strncpy(cmd->params.sleep.start, start->valuestring, sizeof(cmd->params.sleep.start) - 1); printf("sleep start: %s\r\n", cmd->params.sleep.start); } if (cJSON_IsString(end) && end->valuestring != NULL) { strncpy(cmd->params.sleep.end, end->valuestring, sizeof(cmd->params.sleep.end) - 1); printf("sleep end: %s\r\n", cmd->params.sleep.end); } break; } case MSG_DEVICE_WAKE: {/*实际功能由7来实现*/ /*主动唤醒设备*/ printf("MSG_DEVICE_WAKE\r\n"); cJSON *switch_obj = cJSON_GetObjectItem(params, "switch"); if (cJSON_IsString(switch_obj) && switch_obj->valuestring != NULL) { strncpy(cmd->params.wake.switch_state, switch_obj->valuestring, sizeof(cmd->params.wake.switch_state) - 1); printf("wake switch: %s\r\n", cmd->params.wake.switch_state); } break; } case MSG_DEVICE_ALARM: printf(" parse MSG_DEVICE_ALARM\r\n"); break; default: break; } } cJSON_Delete(root); return 0; } /** * @brief 解析服务器事件响应（兼容NFC留言/010告警/011留言/012注册） */ int parse_event_response(const char *json_str, response_t *resp) { int leave_message = 0; int msg_id_num = 0; cJSON *root = cJSON_Parse(json_str); if (root == NULL) { return -1; } // 解析 msg_id cJSON *msg_id = cJSON_GetObjectItem(root, "msg_id"); if (cJSON_IsString(msg_id)) { strncpy(resp->msg_id, msg_id->valuestring, sizeof(resp->msg_id) - 1); msg_id_num = atoi(resp->msg_id); } // 解析 status cJSON *status = cJSON_GetObjectItem(root, "status"); if (cJSON_IsString(status)) { strncpy(resp->status, status->valuestring, sizeof(resp->status) - 1); } // 解析 type (NFC事件/011使用) cJSON *type = cJSON_GetObjectItem(root, "type"); if (cJSON_IsNumber(type)) { resp->type = type->valueint; if (resp->type == 1) { // 启动留言 leave_message = 1; } } // 解析 params cJSON *params = cJSON_GetObjectItem(root, "params"); if (params != NULL) { // ========== 根据 msg_id 分支解析 ========== if (msg_id_num == 10 || msg_id_num == 11 || msg_id_num == 12) { if(msg_id_num == 10){ cJSON *url = cJSON_GetObjectItem(params, "url"); if (cJSON_IsString(url)) { strncpy(resp->url, url->valuestring, sizeof(resp->url) - 1); printf("URL: %s\r\n", resp->url); // 010: 播放告警确认音 add_file_to_play_player_queue(resp->url, 0); } } // ===== 011/012: 直接解析 url 字段 ===== if(msg_id_num == 11){ cJSON *url = cJSON_GetObjectItem(params, "url"); if (cJSON_IsString(url)) { strncpy(resp->url, url->valuestring, sizeof(resp->url) - 1); printf("URL: %s\r\n", resp->url); //quec_usbcdc_write(resp->url, strlen(resp->url)+1); // 011: 播放留言确认音 add_file_to_play_player_queue(resp->url, 1); } } // ===== 012: 额外解析 device_id 和 device_sn ===== if (msg_id_num == 12) { cJSON *device_id = cJSON_GetObjectItem(params, "device_id"); cJSON *device_sn = cJSON_GetObjectItem(params, "device_sn"); if (cJSON_IsString(device_id)) { strncpy(resp->device_id, device_id->valuestring, sizeof(resp->device_id) - 1); } if (cJSON_IsString(device_sn)) { strncpy(resp->device_sn, device_sn->valuestring, sizeof(resp->device_sn) - 1); } printf("Register: id=%s sn=%s\r\n", resp->device_id, resp->device_sn); } } else { // ===== 原有逻辑: NFC事件 - 动态查找 url_x 字段 ===== for (int i = 1; i <= 10; i++) { char key[16]; snprintf(key, sizeof(key), "url_%d", i); cJSON *url = cJSON_GetObjectItem(params, key); if (cJSON_IsString(url)) { printf("URL %d: %s\r\n", i, url->valuestring); if (leave_message == 1) { // 播完音频启动留言 add_file_to_play_player_queue(url->valuestring, 1); } else { add_file_to_play_player_queue(url->valuestring, 0); } } } } } cJSON_Delete(root); return 0; } /* [FIX] 安全处理 topic 名称，避免越界访问 */ static void messageArrived(MessageData* data) { int ret; int payload_log_len; uint32 queue_count = 0; app_work_item *work_item; // 安全获取 topic 字符串（复制到本地缓冲区） char topic_buf[128] = {0}; int topic_len = data->topicName->lenstring.len; if (topic_len >= sizeof(topic_buf)) { topic_len = sizeof(topic_buf) - 1; } memcpy(topic_buf, data->topicName->lenstring.data, topic_len); topic_buf[topic_len] = '\0'; payload_log_len = data->message->payloadlen > 512 ? 512 : (int)data->message->payloadlen; printf("[MQTT-RX] topic=%s len=%d payload=%.*s%s\r\n", topic_buf, (int)data->message->payloadlen, payload_log_len, (const char *)data->message->payload, data->message->payloadlen > (unsigned int)payload_log_len ? "..." : ""); // 比较 topic 类型 int is_command = (strstr(topic_buf, "command") != NULL); int is_event_resp = (strstr(topic_buf, "event_resp") != NULL) || (strstr(topic_buf, imei_buf) != NULL); work_item = hal_os_malloc(sizeof(app_work_item)); if (work_item == NULL) { RLOGE("malloc error"); return; } memset(work_item, 0, sizeof(app_work_item)); work_item->type = MQTT_WORK_RECV; work_item->data = hal_os_malloc(data->message->payloadlen + 1); if (work_item->data) { memset(work_item->data, 0, data->message->payloadlen + 1); memcpy(work_item->data, data->message->payload, data->message->payloadlen); work_item->datalen = data->message->payloadlen; // 直接用长度，不用 strlen if (is_command) { work_item->arg1 = TYPE_COMMAND; } else if (is_event_resp) { work_item->arg1 = TYPE_RESPONE; } ret = ql_rtos_queue_release(mqtt_work_queue_monitor, sizeof(work_item), &work_item, 1000); if (ret != 0) { RLOGE("queue error"); hal_os_free(work_item->data); hal_os_free(work_item); } } else { RLOGE("malloc error"); hal_os_free(work_item); } } int app_mqtt_connect() { int rc = -1; int ret = -1; MQTTPacket_connectData connectData = MQTTPacket_connectData_initializer; int mqtt_server_port=DEFAULT_SERVER_PORT; unsigned int tcp_started; int tcp_errno; MQTT_CONN_LOCK(); if(mqtt_connect==1){ MQTT_CONN_UNLOCK(); return 0; } mqtt_connect = 0; MQTT_CONN_UNLOCK(); RLOGD("========== app_mqtt_connect config =========="); NetworkInit(&network, NULL, PROFILE_IDX); MQTTClientInit(&client, &network, 30000, sendbuf, sizeof(sendbuf), readbuf, sizeof(readbuf)); char* address = DEFAULT_SERVER_DOMAIN; printf("[MQTT-DIAG] TCP begin host=%s port=%d pdp=%d\r\n", address, mqtt_server_port, app_get_data_call_net_status()); tcp_started = ql_rtos_get_systicks(); rc = NetworkConnect(&network, address, mqtt_server_port); tcp_errno = rc != 0 ? ql_get_soc_errno() : 0; printf("[MQTT-DIAG] TCP end rc=%d socket=%d elapsed_ms=%u errno_snapshot=%d\r\n", rc, network.my_socket, (unsigned int)(ql_rtos_get_systicks() - tcp_started) * 5U, tcp_errno); if (rc != 0) { RLOGE("Return code from network connect is %d", rc); MQTTClientDeinit(&client); goto conn_exit; } memset(device_id, 0, sizeof(device_id)); ret = hal_os_get_device_id(device_id); memset(imei_buf,0,sizeof(imei_buf)); ql_dev_get_imei(imei_buf,sizeof(imei_buf)); printf("imei:%s\r\n", imei_buf); sprintf(event_register,"device/%s/event",imei_buf); sprintf(event_resp_register,"device/%s/event_resp",imei_buf); connectData.MQTTVersion = 3; connectData.clientID.cstring = imei_buf; connectData.username.cstring = DEFAULT_SERVER_CLIENT_USER; connectData.password.cstring = DEFAULT_SERVER_CLIENT_PASS; connectData.cleansession = 0; connectData.keepAliveInterval = 60; printf("[MQTT-DIAG] saved_device_id=%d (registration needed when 0)\r\n", ret == 0); if ((rc = talkingq_mqtt_connect_diagnosed(&client, &connectData)) != 0) { RLOGE("Return code from MQTT connect is %d", rc); NetworkDisconnect(&network); MQTTClientDeinit(&client); goto conn_exit; } else RLOGD("MQTT Connected"); #if defined(MQTT_TASK) if ((rc = MQTTStartTask(&client)) != 0) { RLOGE("Return code from start tasks is %d", rc); NetworkDisconnect(&network); MQTTClientDeinit(&client); goto conn_exit; } #endif if(ret!=0){//默认没有device_id if ((rc = MQTTSubscribe(&client, event_resp_register, 0, messageArrived)) != 0) { RLOGE("Return code from MQTT subscribe is %d", rc); rc = MQTTDisconnect(&client); if(rc == SUCCESS) RLOGD("MQTT Disconnected by client"); else RLOGE("MQTT Disconnected failed by client"); NetworkDisconnect(&network); MQTTClientDeinit(&client); goto conn_exit; } RLOGD("MQTT_SUB:%s",event_resp_register); }else{ app_mqtt_sub_topic(); } MQTT_CONN_LOCK(); mqtt_connect = 1; MQTT_CONN_UNLOCK(); // [ADD] 连接成功提示音 app_play_tips_file(11); if(ret!=0){ build_register(); } else { ota_report_pending_success(); } RLOGD("========== app_mqtt_connect okay =========="); return 0; conn_exit: RLOGE("========== app_mqtt_connect error =========="); return rc; } int app_mqtt_sub_topic(void) { int rc = -1; memset(device_id, 0, sizeof(device_id)); hal_os_get_device_id(device_id); snprintf(command_topicName, sizeof(command_topicName), "device/%s/command", device_id); snprintf(respone_topicName, sizeof(respone_topicName), "device/%s/response", device_id); snprintf(event_topicName, sizeof(event_topicName), "device/%s/event", device_id); snprintf(event_resp_topicName, sizeof(event_resp_topicName), "device/%s/event_resp", device_id); if ((rc = MQTTSubscribe(&client, command_topicName, 0, messageArrived)) != 0) { RLOGE("Return code from MQTT subscribe is %d", rc); rc = MQTTDisconnect(&client); if(rc == SUCCESS) RLOGD("MQTT Disconnected by client"); else RLOGE("MQTT Disconnected failed by client"); NetworkDisconnect(&network); MQTTClientDeinit(&client); goto sub_exit; } RLOGD("MQTT_SUB:%s",command_topicName); if ((rc = MQTTSubscribe(&client, event_resp_topicName, 0, messageArrived)) != 0) { RLOGE("Return code from MQTT subscribe is %d", rc); rc = MQTTDisconnect(&client); if(rc == SUCCESS) RLOGD("MQTT Disconnected by client"); else RLOGE("MQTT Disconnected failed by client"); NetworkDisconnect(&network); MQTTClientDeinit(&client); goto sub_exit; } RLOGD("MQTT_SUB:%s",event_resp_topicName); sub_topic_status = 1; return 0; sub_exit: RLOGE("========== app_mqtt_sub error =========="); return rc; } //0:代表respone command //1:代表send event /* [FIX] 将连接状态检查和发送放在同一锁保护区内，避免竞争 */ int app_mqtt_publish(char *buf, int len, int type) { static uint32_t mqtt_pub_cnt_err=0; int rc = -1; int payload_log_len; MQTTMessage message; int cnt = 0; static uint8_t publish_error = 0; const char *publish_topic = NULL; if (type == 0) { publish_topic = respone_topicName; } else if (type == 1) { publish_topic = event_topicName; } else if (type == 2) { publish_topic = event_register; } else { RLOGE("invalid MQTT publish type=%d", type); return -1; } payload_log_len = len > 512 ? 512 : len; printf("[MQTT-TX] topic=%s len=%d payload=%.*s%s\r\n", publish_topic, len, payload_log_len, buf, len > payload_log_len ? "..." : ""); // 锁保护：检查连接状态并发送 int lock_ret = ql_rtos_mutex_lock(g_mqtt_public_mutex, 5000); if (lock_ret != 0) { RLOGE("MQTT mutex lock timeout"); return -1; } if (mqtt_connect == 0) { ql_rtos_mutex_unlock(g_mqtt_public_mutex); return -1; } message.qos = 1; message.retained = 0; message.payload = buf; message.payloadlen = len; while (cnt++ < 3) { u32 start = ql_rtos_get_systicks(); rc = MQTTPublish(&client, publish_topic, &message); RLOGD("Pulish time:%d len: %d\n", (ql_rtos_get_systicks()-start)*5, len); printf("[MQTT-TX] topic=%s attempt=%d rc=%d\r\n", publish_topic, cnt, rc); if (rc != 0) { mqtt_pub_cnt_err++; mqtt_public_error = 1; RLOGE("ERROR Return code from MQTT publish is %d, cnt %d", rc, mqtt_pub_cnt_err); } else { mqtt_public_error = 0; break; } hal_os_task_sleep_ms(100); } ql_rtos_mutex_unlock(g_mqtt_public_mutex); if (rc == 0) { publish_error = 0; } else { if (publish_error++ > 10) { publish_error = 0; app_mqtt_connect_sem(0); RLOGE("publish error reconnect"); } } return rc; } int app_mqtt_disconnect() { int rc; MQTT_CONN_LOCK(); if(mqtt_connect==0){ MQTT_CONN_UNLOCK(); return 0; } mqtt_connect = 0; MQTT_CONN_UNLOCK(); // [ADD] 断开连接提示音 app_play_tips_file(2); sub_topic_status = 0; if ((rc = MQTTUnsubscribe(&client, command_topicName)) != 0) RLOGE("Return code from MQTT un subscribe is %d", rc); if ((rc = MQTTUnsubscribe(&client, event_resp_topicName)) != 0) RLOGE("Return code from MQTT un subscribe is %d", rc); if ((rc = MQTTUnsubscribe(&client, event_resp_register)) != 0) RLOGE("Return code from MQTT un subscribe is %d", rc); rc = MQTTDisconnect(&client); if(rc == SUCCESS) RLOGD("MQTT Disconnected by client"); else RLOGE("MQTT Disconnected failed by client"); NetworkDisconnect(&network); MQTTClientDeinit(&client); return 0; } int app_mqtt_connect_stat() { int status; MQTT_CONN_LOCK(); status = mqtt_connect; MQTT_CONN_UNLOCK(); return status && MQTTIsConnected(&client); } /* [FIX] 对 mqtt_connecting_flag 加锁保护 */ int app_mqtt_connect_sem(char flag) { int ret; app_work_item *work_item; ql_rtos_mutex_lock(g_mqtt_public_mutex, QL_WAIT_FOREVER); if (mqtt_connecting_flag) { ql_rtos_mutex_unlock(g_mqtt_public_mutex); return -1; } mqtt_connecting_flag = 1; ql_rtos_mutex_unlock(g_mqtt_public_mutex); work_item = hal_os_malloc(sizeof(app_work_item)); if (work_item == NULL) { RLOGE("malloc error"); ql_rtos_mutex_lock(g_mqtt_public_mutex, QL_WAIT_FOREVER); mqtt_connecting_flag = 0; ql_rtos_mutex_unlock(g_mqtt_public_mutex); return -1; } memset(work_item, 0, sizeof(app_work_item)); work_item->type = MQTT_WORK_CONNECT; ret = ql_rtos_queue_release(mqtt_work_queue_monitor, sizeof(work_item), &work_item, 1000); if (ret != 0) { RLOGE("queue error"); hal_os_free(work_item); ql_rtos_mutex_lock(g_mqtt_public_mutex, QL_WAIT_FOREVER); mqtt_connecting_flag = 0; ql_rtos_mutex_unlock(g_mqtt_public_mutex); } return ret; } /** * @brief 构建GPS响应 */ char *build_gps_response(const char *msg_id, gps_data_t *gps) { cJSON *root = cJSON_CreateObject(); cJSON *data = cJSON_CreateObject(); cJSON_AddStringToObject(root, "msg_id", msg_id); cJSON_AddStringToObject(root, "status", "success"); cJSON_AddNumberToObject(data, "latitude", gps->latitude); cJSON_AddNumberToObject(data, "longitude", gps->longitude); cJSON_AddItemToObject(root, "data", data); char *json_str = cJSON_PrintUnformatted(root); cJSON_Delete(root); return json_str; } /** * @brief 构建心跳上报消息 */ static char *build_heartbeat(device_info_t *info, const char *msg_id) { cJSON *root = cJSON_CreateObject(); cJSON *data = cJSON_CreateObject(); cJSON_AddStringToObject(root, "msg_id", msg_id ? msg_id : "000"); cJSON_AddNumberToObject(data, "power", info->power); cJSON_AddNumberToObject(data, "signal", info->signal); cJSON_AddNumberToObject(data, "voice", info->voice); cJSON_AddStringToObject(data, "imei", info->imei); cJSON_AddStringToObject(data, "version", info->version); cJSON_AddNumberToObject(data, "latitude", info->real_gps.latitude); cJSON_AddNumberToObject(data, "longitude", info->real_gps.longitude); cJSON_AddNumberToObject(data, "device_type", 2); cJSON_AddItemToObject(root, "data", data); char *json_str = cJSON_PrintUnformatted(root); cJSON_Delete(root); return json_str; // 调用者需要 free } char build_alarm(void) { set_gnss_mode(GNSS_MODE_FAST); cJSON *root = cJSON_CreateObject(); cJSON_AddStringToObject(root, "msg_id","010"); char *json_str = cJSON_PrintUnformatted(root); cJSON_Delete(root); printf("build_alarm"); // char *log_str = "build_alarm\r\n"; // quec_usbcdc_write(log_str, strlen(log_str)+1); app_mqtt_publish(json_str, strlen(json_str), 1); free(json_str); return 0; } char build_leave_to_parent(void) { char buf_talkingq_id[64]={0}; cJSON *root = cJSON_CreateObject(); memset(buf_talkingq_id,0,sizeof(buf_talkingq_id)); hal_os_get_device_id(buf_talkingq_id); cJSON_AddStringToObject(root, "msg_id","011"); cJSON *params = cJSON_CreateObject(); cJSON_AddStringToObject(params, "uuid", buf_talkingq_id); cJSON_AddItemToObject(root, "params", params); char *json_str = cJSON_PrintUnformatted(root); cJSON_Delete(root); printf("build_leave_to_parent"); // char *log_str = "build_leave_to_parent\r\n"; // quec_usbcdc_write(log_str, strlen(log_str)+1); app_mqtt_publish(json_str, strlen(json_str), 1); free(json_str); return 0; } void build_register(void) { cJSON *root = cJSON_CreateObject(); cJSON_AddStringToObject(root, "msg_id","012"); char *json_str = cJSON_PrintUnformatted(root); cJSON_Delete(root); app_mqtt_publish(json_str, strlen(json_str), 2); free(json_str); } /** * @brief 构建简单的事件确认响应 */ char *build_event_response_ack(const char *msg_id) { cJSON *root = cJSON_CreateObject(); cJSON_AddStringToObject(root, "msg_id", msg_id); cJSON_AddStringToObject(root, "status", "success"); char *json_str = cJSON_PrintUnformatted(root); cJSON_Delete(root); return json_str; } char *build_bind(const char *msg_id, const char *uuid) { cJSON *root = cJSON_CreateObject(); cJSON *data = cJSON_CreateObject(); cJSON_AddStringToObject(root, "msg_id", msg_id); cJSON_AddStringToObject(root, "status", "success"); cJSON_AddStringToObject(data, "uuid", uuid); cJSON_AddItemToObject(root, "data", data); char *json_str = cJSON_PrintUnformatted(root); cJSON_Delete(root); return json_str; } char *build_device_status(const char *msg_id, int status) { cJSON *root = cJSON_CreateObject(); cJSON *data = cJSON_CreateObject(); cJSON_AddStringToObject(root, "msg_id", msg_id); cJSON_AddStringToObject(root, "status", "success"); if(status==0){ cJSON_AddStringToObject(data, "status", "off"); }else{ cJSON_AddStringToObject(data, "status", "on"); } cJSON_AddItemToObject(root, "data", data); char *json_str = cJSON_PrintUnformatted(root); cJSON_Delete(root); return json_str; } /** * @brief 构建定时休眠设置响应 */ char *build_sleep_response(const char *msg_id) { cJSON *root = cJSON_CreateObject(); cJSON_AddStringToObject(root, "msg_id", msg_id); cJSON_AddStringToObject(root, "status", "success"); char *json_str = cJSON_PrintUnformatted(root); cJSON_Delete(root); return json_str; } /** * @brief 构建远程唤醒/休眠响应 */ char *build_wake_response(const char *msg_id, const char *switch_state) { cJSON *root = cJSON_CreateObject(); cJSON_AddStringToObject(root, "msg_id", msg_id); cJSON_AddStringToObject(root, "status", "success"); // 根据协议，009 只需要返回 msg_id 和 status，不需要 data 字段 char *json_str = cJSON_PrintUnformatted(root); cJSON_Delete(root); return json_str; } /** * @brief 构建OTA接受响应 */ char *build_ota_accepted(const char *msg_id, const char *cur_ver, const char *tgt_ver) { cJSON *root = cJSON_CreateObject(); cJSON *data = cJSON_CreateObject(); if (root == NULL || data == NULL) { cJSON_Delete(root); cJSON_Delete(data); return NULL; } cJSON_AddStringToObject(root, "msg_id", msg_id); cJSON_AddStringToObject(root, "status", "accepted"); cJSON_AddStringToObject(data, "current_version", cur_ver); cJSON_AddStringToObject(data, "target_version", tgt_ver); cJSON_AddItemToObject(root, "data", data); char *json_str = cJSON_PrintUnformatted(root); cJSON_Delete(root); return json_str; } /** * @brief 构建OTA升级成功响应 */ static char *build_ota_success(const char *new_ver) { cJSON *root = cJSON_CreateObject(); cJSON *data = cJSON_CreateObject(); if (root == NULL || data == NULL) { cJSON_Delete(root); cJSON_Delete(data); return NULL; } cJSON_AddStringToObject(root, "msg_id", "003"); cJSON_AddStringToObject(root, "status", "success"); cJSON_AddStringToObject(data, "new_version", new_ver); cJSON_AddItemToObject(root, "data", data); char *json_str = cJSON_PrintUnformatted(root); cJSON_Delete(root); return json_str; } static void ota_report_pending_success(void) { ota_state_t state; char current_version[64] = {0}; char *resp; int compare_result; int ret; if (ota_state_load(&state) != 0) { return; } hal_os_get_build_ver(current_version); if (ota_compare_version(current_version, state.target_version, &compare_result) != 0 || compare_result != 0) { RLOGE("OTA pending state version mismatch: current=%s target=%s", current_version, state.target_version); return; } resp = build_ota_success(current_version); if (resp == NULL) { RLOGE("build OTA success response failed"); return; } ret = app_mqtt_publish(resp, strlen(resp), 0); free(resp); if (ret == 0) { printf("OTA success response published, version=%s\r\n", current_version); ota_state_clear(); } else { RLOGE("OTA success response publish failed: %d", ret); } } /** * @brief 构建音量调节响应 */ char *build_volume_response(const char *msg_id, int current_level, const char *time) { cJSON *root = cJSON_CreateObject(); cJSON *data = cJSON_CreateObject(); cJSON_AddStringToObject(root, "msg_id", msg_id); cJSON_AddStringToObject(root, "status", "success"); cJSON_AddNumberToObject(data, "current_level", current_level); cJSON_AddItemToObject(root, "data", data); // 新增：带上服务器下发的时间戳 if (time != NULL && time[0] != '\0') { cJSON_AddStringToObject(root, "time", time); } char *json_str = cJSON_PrintUnformatted(root); cJSON_Delete(root); return json_str; } /* ==================== 业务处理函数 ==================== */ static void handle_ota_upgrade(command_t *cmd) { char current_version[64] = {0}; char *resp; int compare_result; int ret; uint32_t prompt_start_time; if (cmd == NULL || fota_start == 1) { RLOGE("OTA request ignored: invalid command or OTA already running"); return; } if (!ota_url_is_allowed(cmd->params.ota.url)) { RLOGE("OTA URL is not allowed: %s", cmd->params.ota.url); return; } hal_os_get_build_ver(current_version); if (ota_compare_version(current_version, cmd->params.ota.version, &compare_result) != 0) { RLOGE("invalid OTA version: current=%s target=%s", current_version, cmd->params.ota.version); return; } if (compare_result > 0) { RLOGE("OTA downgrade rejected: current=%s target=%s", current_version, cmd->params.ota.version); return; } if (ota_state_save(current_version, cmd->params.ota.version) != 0) { RLOGE("OTA state save failed"); return; } resp = build_ota_accepted(cmd->msg_id, current_version, cmd->params.ota.version); if (resp == NULL) { RLOGE("build OTA accepted response failed"); ota_state_clear(); return; } ret = app_mqtt_publish(resp, strlen(resp), 0); free(resp); if (ret != 0) { RLOGE("OTA accepted response publish failed: %d", ret); ota_state_clear(); return; } if (compare_result == 0) { printf("OTA target version is already installed: %s\r\n", current_version); ota_report_pending_success(); return; } fota_start = 1; ret = app_play_ota_tips_file(); if (ret != 0) { RLOGE("OTA prompt could not start: %d", ret); } else { prompt_start_time = hal_os_up_time_ms(); while (xiaoice_get_play_counter() > 0) { ql_rtos_task_sleep_ms(OTA_PROMPT_POLL_MS); if (hal_os_up_time_ms() - prompt_start_time > OTA_PROMPT_PLAY_TIMEOUT_MS) { RLOGE("OTA prompt playback timeout, counter=%d", xiaoice_get_play_counter()); break; } } if (xiaoice_get_play_counter() == 0) { printf("OTA prompt playback completed\r\n"); } } /* Product requirement: keep deleting the configured audio files before OTA. */ app_file_check_del(); ret = ql_fullfota_app_start(cmd->params.ota.url); if (ret != 0) { RLOGE("start OTA failed: %d", ret); fota_start = 0; ota_state_clear(); } } /** * @brief 命令分发处理 */ void handle_command(command_t *cmd) { switch (cmd->type) { case MSG_GPS_QUERY: { set_gnss_mode(GNSS_MODE_FAST); // 获取GPS并响应 gps_data_t gps_resp; // 实际调用GPS模块 gps_location_info_t *gsp_real = get_gps_location_info(); gps_resp.longitude = gsp_real->longitude; gps_resp.latitude = gsp_real->latitude; char *resp = build_gps_response(cmd->msg_id, &gps_resp); app_mqtt_publish(resp, strlen(resp), 0); free(resp); break; } case MSG_VOLUME_SET: { uint8_t vol = cmd->params.volume_level; uint8_t up_down = 0; if(g_cur_volumemsg_id, cmd->params.volume_level, cmd->time); /*while(get_mqtt_public_mutex()!=0){ hal_os_task_sleep_ms(100); }*/ app_mqtt_publish(resp, strlen(resp), 0); free(resp); break; } case MSG_OTA_UPGRADE: { handle_ota_upgrade(cmd); break; } case MSG_NFC_NOTICE: { // 播放提示音 char *resp = build_event_response_ack(cmd->msg_id); /*while(get_mqtt_public_mutex()!=0){ hal_os_task_sleep_ms(100); }*/ app_mqtt_publish(resp, strlen(resp), 0); free(resp); add_file_to_play_player_queue(cmd->params.nfc.url, 0); break; } case MSG_NFC_BIND: { add_file_to_play_player_queue(cmd->params.bind.url, 0); g_nfc_bind_status = 1; break; } case MSG_DEVICE_STATUS: { printf("handle MSG_DEVICE_STATUS:%d\r\n", cmd->sub_type); /*查询或者设置设备状态*/ if(cmd->sub_type==0){//查询 }else if(cmd->sub_type==1){//打开 exit_sleep_mode(); }else if(cmd->sub_type==2){//关闭 enter_sleep_mode(); } char *resp = build_device_status(cmd->msg_id, get_sleep_mode()); app_mqtt_publish(resp, strlen(resp), 0); free(resp); //add_file_to_play_player_queue(cmd->params.nfc.url, 0); break; } case MSG_DEVICE_SLEEP: { /*设置设备休眠时间*/ printf("handle MSG_DEVICE_SLEEP\r\n"); // 解析时间字符串 "22:00" -> hour=22, min=0 int start_h = 0, start_m = 0, end_h = 0, end_m = 0; sscanf(cmd->params.sleep.start, "%d:%d", &start_h, &start_m); sscanf(cmd->params.sleep.end, "%d:%d", &end_h, &end_m); // 保存配置 g_sleep_cfg.start_hour = (uint8_t)start_h; g_sleep_cfg.start_min = (uint8_t)start_m; g_sleep_cfg.end_hour = (uint8_t)end_h; g_sleep_cfg.end_min = (uint8_t)end_m; g_sleep_cfg.enable = 1; save_sleep_config(); printf("sleep set: %02d:%02d ~ %02d:%02d\r\n", g_sleep_cfg.start_hour, g_sleep_cfg.start_min, g_sleep_cfg.end_hour, g_sleep_cfg.end_min); // 立即检查是否需要进入休眠 char time_buf[6] = {0}; hal_os_get_localtime(time_buf); if (is_in_sleep_time(time_buf[3], time_buf[4])) { printf("goto sleep now\r\n"); enter_sleep_mode(); } char *resp = build_sleep_response(cmd->msg_id); app_mqtt_publish(resp, strlen(resp), 0); free(resp); break; } case MSG_DEVICE_WAKE: { /*主动唤醒设备*/ printf("handle MSG_DEVICE_WAKE\r\n"); if(g_sleep_cfg.enable){ g_sleep_cfg.enable = 0;//只要是立即休眠或唤醒都把原有设置时间失效 save_sleep_config(); } // 根据 switch 参数执行相应操作 if (strcmp(cmd->params.wake.switch_state, "on") == 0) { printf("Remote WAKE UP device\r\n"); // TODO: 唤醒设备 exit_sleep_mode(); } else if (strcmp(cmd->params.wake.switch_state, "off") == 0) { printf("Remote SLEEP device\r\n"); if(!get_sleep_mode()){ // TODO: 休眠设备 enter_sleep_mode(); } } char *resp = build_wake_response(cmd->msg_id, cmd->params.wake.switch_state); app_mqtt_publish(resp, strlen(resp), 0); free(resp); break; } case MSG_DEVICE_ALARM: printf("handle MSG_DEVICE_ALARM\r\n"); break; default: break; } } /** * @brief NFC触发处理 */ void handle_nfc_trigger(const char *uuid) { // 上报NFC事件 cJSON *root = cJSON_CreateObject(); cJSON_AddStringToObject(root, "msg_id", "005"); cJSON *params = cJSON_CreateObject(); cJSON_AddStringToObject(params, "uuid", uuid); cJSON_AddItemToObject(root, "params", params); char *json_str = cJSON_PrintUnformatted(root); cJSON_Delete(root); /*while(get_mqtt_public_mutex()!=0){ hal_os_task_sleep_ms(100); }*/ app_mqtt_publish(json_str, strlen(json_str), 1); free(json_str); } void handle_nfc_bind(const char *uuid) { char iccid[QL_SIM_ICCID_LEN * 2 + 1] = {0}; QL_SIM_ERROR_CODE sim_ret; cJSON *root = cJSON_CreateObject(); cJSON *data = cJSON_CreateObject(); cJSON_AddStringToObject(root, "msg_id", "006"); cJSON_AddStringToObject(root, "status", "success"); cJSON_AddStringToObject(data, "uuid", uuid); sim_ret = ql_sim_get_iccid(iccid, sizeof(iccid)); if (sim_ret == QL_SIM_SUCCESS && iccid[0] != '\0') { printf("NFC bind ICCID:%s\r\n", iccid); cJSON_AddStringToObject(data, "iccid", iccid); } else { printf("NFC bind get ICCID failed, ret:%d\r\n", sim_ret); } cJSON_AddItemToObject(root, "data", data); char *json_str = cJSON_PrintUnformatted(root); cJSON_Delete(root); app_mqtt_publish(json_str, strlen(json_str), 0); free(json_str); } static void app_mqtt_task_monitor_thread(void *arvg) { char temp[1024]={0}; char update_flag = 0; int ret; device_info_t info; app_work_item *work_item = NULL; ql_rtos_queue_create(&mqtt_work_queue_monitor, sizeof(app_work_item *), 20); while (1) { ret = ql_rtos_queue_wait(mqtt_work_queue_monitor, &work_item, sizeof(app_work_item *), 60000); if(ret==0){ if(work_item->type == MQTT_WORK_CONNECT){ app_mqtt_disconnect(); ret = app_mqtt_connect(); if(ret==0){ mqtt_public_error = 0; } /* [FIX] 清除连接中标志，加锁保护 */ ql_rtos_mutex_lock(g_mqtt_public_mutex, QL_WAIT_FOREVER); mqtt_connecting_flag = 0; ql_rtos_mutex_unlock(g_mqtt_public_mutex); }else if(work_item->type == MQTT_WORK_RECV){ memset(payload, 0, sizeof(payload)); int len = work_item->datalen < sizeof(payload) - 1 ? work_item->datalen : sizeof(payload) - 1; memcpy(payload, (char*)work_item->data, len); payload[len] = '\0'; // 判断topic类型 if (work_item->arg1==TYPE_COMMAND) { printf("deal command\r\n"); // 处理命令 command_t cmd; memset(&cmd, 0, sizeof(cmd)); if (parse_command(payload, &cmd) == 0) { printf("parse_command ok\r\n"); handle_command(&cmd); } } else if (work_item->arg1==TYPE_RESPONE) { printf("deal event_resp\r\n"); response_t resp; memset(&resp, 0, sizeof(resp)); if (parse_event_response(payload, &resp) == 0) { int msg_id_num = atoi(resp.msg_id); switch (msg_id_num) { case 10: { // 010 告警响应 printf("Alarm status\r\n"); break; } case 11: { // 011 留言响应 - URL已在parse中播放 printf("leave url:%s\r\n", resp.url); break; } case 12: { // 012 注册响应 - 保存DeviceID和SN printf(" DeviceID: %s\r\n", resp.device_id); printf(" SN: %s\r\n", resp.device_sn); // TODO: 保存到文件系统 hal_os_set_device_id(resp.device_id); hal_os_set_device_sn(resp.device_sn); app_mqtt_sub_topic(); break; } default: { // 原有NFC事件处理（001-005等） printf("NFC event resp, type:%d\r\n", resp.type); break; } } } } } if(work_item){ if(work_item->data){ hal_os_free(work_item->data); } hal_os_free(work_item); } hal_os_task_sleep_ms(10); }else{ int conn_status = get_mqtt_connect_status(); if(conn_status==0 && app_get_data_call_net_status()){//网络ok 但是没连上服务器 RLOGD("Reconnected mqtt 0\n"); app_mqtt_connect_sem(0); continue; }else if(conn_status==1 && app_get_data_call_net_status()==0){//没网络 则主动disconnect RLOGD("Reconnected mqtt 2\n"); app_mqtt_connect_sem(0); continue; } } //检测是否真的mqtt在线 if(get_mqtt_connect_status() && app_get_data_call_net_status()){ if(!app_mqtt_connect_stat()){ RLOGD("Reconnected mqtt 1\n"); app_mqtt_connect_sem(0); } } hal_os_task_sleep_ms(100); } } static void app_mqtt_task_event_thread(void *arvg) { device_info_t info; QL_NW_SIGNAL_STRENGTH_INFO_T rsrp_t; int rsrp_value = 0; u32 app_heartbeat_time_record = hal_os_up_time_ms(); u32 app_volume_save_time_record = hal_os_up_time_ms(); // 心跳上报循环：FAST模式10秒上报一次，SLOW模式6分钟上报一次 while(1){ if(hal_os_up_time_ms() - app_volume_save_time_record < APP_VOLUME_SAVE_TIME){ if(g_cur_volume != g_pre_volume){ if(g_cur_volume == 0) g_cur_volume = 10; hal_kv_set("volume", &g_cur_volume, 1); g_pre_volume = g_cur_volume; printf("g_cur_volume:%d\r\n", g_cur_volume); } app_volume_save_time_record = hal_os_up_time_ms(); } if(battery_first_update_flag!=1){ // 根据当前GNSS模式决定上报间隔 if(get_gnss_mode() == GNSS_MODE_FAST){ // FAST模式：10秒上报一次 if(hal_os_up_time_ms() - app_heartbeat_time_record < GNSS_FAST_MODE_HEART_REPORT_TIME){ ql_rtos_task_sleep_ms(5000); continue; } }else{ // SLOW模式：6分钟 = 360秒上报一次 if(hal_os_up_time_ms() - app_heartbeat_time_record < GNSS_SLOW_MODE_HEART_REPORT_TIME){ ql_rtos_task_sleep_ms(5000); continue; } } } if(battery_first_update_flag==1){ battery_first_update_flag = 2; } app_heartbeat_time_record = hal_os_up_time_ms(); // 采集设备信息 info.power = battery_get_level(); // 使用与设备内部一致的最终电量档位 info.voice = g_cur_volume; // 40,60,80,100 hal_os_get_build_ver(&info.version); ql_nw_get_signal_strength(&rsrp_t); rsrp_value = rsrp_t.LTE_SignalStrength.rsrp; if(-85 <= rsrp_value){ info.signal = 5; //5格 }else if(-95 < rsrp_value && rsrp_value < -85){ info.signal = 4; }else if(-105 < rsrp_value && rsrp_value <= -95){ info.signal = 3; }else if(-115 < rsrp_value && rsrp_value <= -105){ info.signal = 2; }else if(rsrp_value <= -115){ info.signal = 1; } sprintf(info.imei, "%s", imei_buf); gps_location_info_t *gsp_real = get_gps_location_info(); info.real_gps.longitude = gsp_real->longitude; info.real_gps.latitude = gsp_real->latitude; // 构建并发送 char *heartbeat = build_heartbeat(&info, "000"); app_mqtt_publish(heartbeat, strlen(heartbeat), 1); free(heartbeat); } } int app_mqtt_init() { ql_rtos_mutex_create(&g_mqtt_public_mutex); // 安全修复：初始化连接状态互斥锁 ql_rtos_mutex_create(&g_mqtt_conn_mutex); ql_task_t appTaskRef_monitor = NULL; ql_rtos_task_create(&appTaskRef_monitor, 1024*30, 100, "monitor", app_mqtt_task_monitor_thread, NULL); ql_task_t eventTaskRef = NULL; ql_rtos_task_create(&eventTaskRef, 1024*6, 100, "task_event", app_mqtt_task_event_thread, NULL); return 0; }