chenjintao/banban
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
d6c31ffe310316e897906fb6ba665d98a4fcdac2
banban/ql-application/threadx/interface/custom_app/mqtt.c
T

1531 lines
43 KiB
C
Raw Blame History

/*================================================================
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
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 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;
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;
}
/* ==================== 解析函数 ==================== */
/**
* @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 (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)) {
strncpy(cmd->params.ota.url, url->valuestring,
sizeof(cmd->params.ota.url) - 1);
}
if (cJSON_IsString(version)) {
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);
app_play_tips_file(3);
fota_start = 1;
ql_rtos_task_sleep_ms(6000);
app_file_check_del();
ql_fullfota_app_start(cmd->params.ota.url);
/*
{'msg_id': '003', 'params': {'url': 'http://175.24.73.253/assets/firmware/20260425203347-fota.bin', 'version': '2.0.1'}}
*/
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 <= 5; 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();
}
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();
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 构建音量调节响应
*/
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;
}
/* ==================== 业务处理函数 ==================== */
/**
* @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_volume<vol){
up_down = 1;
}else{
up_down = 0;
}
if (vol <= 10) {
app_set_volume(VOLUME_10);
} else if (vol <= 20) {
app_set_volume(VOLUME_20);
} else if(vol <= 30) {
app_set_volume(VOLUME_30);
}else if(vol <= 40) {
app_set_volume(VOLUME_40);
}else if(vol <= 50) {
app_set_volume(VOLUME_50);
}else if(vol <= 60) {
app_set_volume(VOLUME_60);
}else if(vol <= 70) {
app_set_volume(VOLUME_70);
}else if(vol <= 80) {
app_set_volume(VOLUME_80);
}else if(vol <= 90) {
app_set_volume(VOLUME_90);
}else {
app_set_volume(VOLUME_100);
}
if(up_down){
app_play_tips_file(16);
}else{
app_play_tips_file(17);
}
char *resp = build_volume_response(cmd->msg_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: {
// 立即响应accepted
char *resp = build_ota_accepted(cmd->msg_id,
hal_os_get_current_version(),
cmd->params.ota.version);
/*while(get_mqtt_public_mutex()!=0){
hal_os_task_sleep_ms(100);
}*/
/*
添加删除所有本地音频动作
*/
app_mqtt_publish(resp, strlen(resp), 0);
free(resp);
// 启动OTA任务
fota_start = 1;
ql_rtos_task_sleep_ms(5000);
ql_fullfota_app_start(cmd->params.ota.url);
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 = calc_battery_level(g_vbat_vol); // 10,20,30...100
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;
}
Reference in New Issue View Git Blame Copy Permalink