From e74e13e37280ab3ec64523cde85bd85d1a83682d Mon Sep 17 00:00:00 2001 From: zzh <838331105@qq.com> Date: Mon, 22 Dec 2025 11:36:02 +0800 Subject: [PATCH] =?UTF-8?q?=E5=90=8C=E6=AD=A5=E6=94=B9=E7=89=88MQTT=20payl?= =?UTF-8?q?oad=20=E8=A7=A3=E6=9E=90=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main.c | 253 +++++++++++++++++++++++----------------- mqtt_utils/mqtt_utils.c | 136 ++++++++++++++++++--- mqtt_utils/mqtt_utils.h | 1 + 3 files changed, 267 insertions(+), 123 deletions(-) diff --git a/main.c b/main.c index e72c213..a1fc4ed 100644 --- a/main.c +++ b/main.c @@ -2011,124 +2011,156 @@ void *thread_mqtt_recv(void *arg){ // 重置变量 items_size = 0; - colors_size = 0; - // 解析Items数组 - get_size_from_json_string_arry_by_key(payload, "Items", &items_size); - LOG_I("Items size = %d\n", items_size); - - // 解析Time字段（常位时间） - get_int_from_json_string_by_key(payload,"Time",&mqtt_parm.msg_duration); - LOG_I("msg_duration = %d\n", mqtt_parm.msg_duration); - if(mqtt_parm.msg_duration<=5){ - mqtt_parm.msg_duration=5; - LOG_I("new msg_duration = %d\n", mqtt_parm.msg_duration); - } - - // 遍历Items数组 - for(int i = 0; i < items_size; i++){ - get_string_from_json_string_arry_by_key(payload,"Items",msg_item_value,1024,i); - snprintf(msg_items, sizeof(msg_items), "{%s}", msg_item_value); - LOG_I("msg_items[%d] = %s\n", i, msg_items); - - // 解析Beep字段（蜂鸣） + // 解析新格式payload + LOG_I("Parsing new payload format with params\n"); + + // 先打印完整的payload以便调试 + LOG_I("Complete payload: %s\n", payload); + + // 解析params中的各个字段 + char params_str[1024] = {0}; + char color_json[512] = {0}; // 提前声明color_json + int params_ret = get_string_from_json_string_by_key(payload, "params", params_str, sizeof(params_str)); + LOG_I("get_string_from_json_string_by_key return: %d\n", params_ret); + LOG_I("params_str: %s\n", params_str); + + // 如果params_str为空，尝试直接从payload解析 + if(strlen(params_str) == 0) { + LOG_I("params_str is empty, trying direct parse from payload\n"); + // 直接从payload解析各个字段 int beep_enable = 0; - get_int_from_json_string_by_key(msg_items,"Beep",&beep_enable); + get_int_from_json_string_by_key(payload, "beep", &beep_enable); + LOG_I("Direct parse Beep = %d\n", beep_enable); + changesound = beep_enable ? 1 : 0; + + int flashing_enable = 0; + get_int_from_json_string_by_key(payload, "flashing", &flashing_enable); + changeflash = flashing_enable ? 3 : 1; + LOG_I("Direct parse Flashing = %d, changeflash = %d\n", flashing_enable, changeflash); + + int duration = 0; + get_int_from_json_string_by_key(payload, "duration", &duration); + mqtt_parm.msg_duration = duration; + LOG_I("Direct parse Duration = %d\n", mqtt_parm.msg_duration); + if(mqtt_parm.msg_duration <= 5) { + mqtt_parm.msg_duration = 5; + LOG_I("new msg_duration = %d\n", mqtt_parm.msg_duration); + } + + // 解析颜色 + char color_str[512] = {0}; + get_string_from_json_string_by_key(payload, "color", color_str, sizeof(color_str)); + LOG_I("Direct parse color_str: %s\n", color_str); + + // 包装color_str为有效的JSON对象 + snprintf(color_json, sizeof(color_json), "{%s}", color_str); + LOG_I("color_json: %s\n", color_json); + } else { + // 包装params_str为有效的JSON对象 + char params_json[1024] = {0}; + snprintf(params_json, sizeof(params_json), "{%s}", params_str); + LOG_I("params_json: %s\n", params_json); + + int beep_enable = 0; + get_int_from_json_string_by_key(params_json, "beep", &beep_enable); LOG_I("Beep = %d\n", beep_enable); changesound = beep_enable ? 1 : 0; - - // 解析Flashing字段（闪烁） + int flashing_enable = 0; - get_int_from_json_string_by_key(msg_items,"Flashing",&flashing_enable); + get_int_from_json_string_by_key(params_json, "flashing", &flashing_enable); changeflash = flashing_enable ? 3 : 1; // 3=闪烁, 1=常亮 LOG_I("Flashing = %d, changeflash = %d\n", flashing_enable, changeflash); - - // 解析TagID字段（灯条ID） - LOG_I("Before parsing TagID, msg_items = %s\n", msg_items); - LOG_I("mqtt_parm.msg_sn buffer size = %d\n", (int)sizeof(mqtt_parm.msg_sn)); - // 尝试使用不带unescape的函数 - int tagid_ret = get_string_from_json_string_by_key(msg_items, "TagID", mqtt_parm.msg_sn,sizeof(mqtt_parm.msg_sn)); - LOG_I("TagID parsing result = %d\n", tagid_ret); - LOG_I("Original TagID = %s\n", mqtt_parm.msg_sn); - - // 只取TagID的后8位 - int tag_len = strlen(mqtt_parm.msg_sn); - LOG_I("TagID length = %d\n", tag_len); - if(tag_len > 8) { - // 将后8位移动到字符串开头 - memmove(mqtt_parm.msg_sn, mqtt_parm.msg_sn + (tag_len - 8), 8); - mqtt_parm.msg_sn[8] = '\0'; - LOG_I("TagID after truncation = %s\n", mqtt_parm.msg_sn); - } else if(tag_len == 0) { - LOG_I("Warning: TagID is empty!\n"); + + int duration = 0; + get_int_from_json_string_by_key(params_json, "duration", &duration); + mqtt_parm.msg_duration = duration; + LOG_I("Duration = %d\n", mqtt_parm.msg_duration); + if(mqtt_parm.msg_duration <= 5) { + mqtt_parm.msg_duration = 5; + LOG_I("new msg_duration = %d\n", mqtt_parm.msg_duration); } - LOG_I("TagID (last 8 chars) = %s\n", mqtt_parm.msg_sn); - - // 解析Colors数组 - get_size_from_json_string_arry_by_key(msg_items, "Colors", &colors_size); - LOG_I("Colors size = %d\n", colors_size); - - // 默认颜色设置 - changecolor = 1; // 默认蓝色 - - // 遍历Colors数组，找到第一个为true的颜色 - for(int j = 0; j < colors_size; j++){ - get_string_from_json_string_arry_by_key(msg_items,"Colors",msg_color_value,512,j); - snprintf(msg_colors, sizeof(msg_colors), "{%s}", msg_color_value); - LOG_I("msg_colors[%d] = %s\n", j, msg_colors); - - int r_enable = 0, g_enable = 0, b_enable = 0; - get_int_from_json_string_by_key(msg_colors,"R",&r_enable); - get_int_from_json_string_by_key(msg_colors,"G",&g_enable); - get_int_from_json_string_by_key(msg_colors,"B",&b_enable); - - // 根据RGB组合设置颜色 - if(!r_enable && !g_enable && !b_enable){ - changecolor = 0; // 不亮 - break; - }else if(r_enable && !g_enable && !b_enable){ - changecolor = 4; // 红色 - break; - }else if(!r_enable && g_enable && !b_enable){ - changecolor = 2; // 绿色 - break; - }else if(!r_enable && !g_enable && b_enable){ - changecolor = 1; // 蓝色 - break; - }else if(r_enable && g_enable && !b_enable){ - changecolor = 6; // 黄色 - break; - }else if(r_enable && !g_enable && b_enable){ - changecolor = 5; // 紫色 - break; - }else if(!r_enable && g_enable && b_enable){ - changecolor = 3; // 青色 - break; - }else if(r_enable && g_enable && b_enable){ - changecolor = 7; // 白色 - break; + + // 解析颜色 + char color_str[512] = {0}; + get_string_from_json_string_by_key(params_json, "color", color_str, sizeof(color_str)); + LOG_I("color_str: %s\n", color_str); + + // 包装color_str为有效的JSON对象 + snprintf(color_json, sizeof(color_json), "{%s}", color_str); + LOG_I("color_json: %s\n", color_json); + } + + // 解析RED/GREEN/BLUE或R/G/B作为键值 + int r_val = 0, g_val = 0, b_val = 0; + + // 先尝试完整名称 + get_int_from_json_string_by_key(color_json, "RED", &r_val); + get_int_from_json_string_by_key(color_json, "GREEN", &g_val); + get_int_from_json_string_by_key(color_json, "BLUE", &b_val); + + // 如果完整名称没有值，尝试缩写 + if(r_val == 0 && g_val == 0 && b_val == 0) { + get_int_from_json_string_by_key(color_json, "R", &r_val); + get_int_from_json_string_by_key(color_json, "G", &g_val); + get_int_from_json_string_by_key(color_json, "B", &b_val); + } + + LOG_I("RGB values: R=%d, G=%d, B=%d\n", r_val, g_val, b_val); + + // 根据RGB值设置颜色（0=不亮, 1=亮） + if(r_val == 0 && g_val == 0 && b_val == 0) { + changecolor = 0; // 不亮 + } else if(r_val == 1 && g_val == 0 && b_val == 0) { + changecolor = 4; // 红色 + } else if(r_val == 0 && g_val == 1 && b_val == 0) { + changecolor = 2; // 绿色 + } else if(r_val == 0 && g_val == 0 && b_val == 1) { + changecolor = 1; // 蓝色 + } else if(r_val == 1 && g_val == 1 && b_val == 0) { + changecolor = 6; // 黄色 + } else if(r_val == 1 && g_val == 0 && b_val == 1) { + changecolor = 5; // 紫色 + } else if(r_val == 0 && g_val == 1 && b_val == 1) { + changecolor = 3; // 青色 + } else if(r_val == 1 && g_val == 1 && b_val == 1) { + changecolor = 7; // 白色 + } + LOG_I("changecolor=%d\n", changecolor); + + // 判断任务类型：根据method字段判断是灯条任务还是OTA任务 + char method[128] = {0}; + int has_method = get_string_from_json_string_by_key(payload, "method", method, sizeof(method)); + + int is_lightbar_task = 0; + if(has_method == 0 && strstr(method, "lightOperate") != NULL) { + // 灯条任务 + is_lightbar_task = 1; + items_size = 1; + lightbars_size = 1; + + // 从topic解析出的deviceName获取设备ID + if(strlen(g_mqtt_deviceName) > 0) { + LOG_I("Device ID from topic: %s\n", g_mqtt_deviceName); + // 将deviceName设置为灯条ID + memset(mqtt_parm.msg_sn, 0, sizeof(mqtt_parm.msg_sn)); + strncpy(mqtt_parm.msg_sn, g_mqtt_deviceName, sizeof(mqtt_parm.msg_sn) - 1); + LOG_I("Set light bar ID: %s\n", mqtt_parm.msg_sn); + + // 获取后6位并设置到lightsn1 + int len = strlen(g_mqtt_deviceName); + if(len >= 6) { + char *last6 = g_mqtt_deviceName + (len - 6); + memset(lightsn1, 0, sizeof(lightsn1)); + strncpy(lightsn1, last6, 6); + LOG_I("Set lightsn1 (last 6 digits): %s\n", lightsn1); } } - - LOG_I("changecolor=%d\n",changecolor); - - // 将TagID存储到lightsn数组中（保持兼容性） - if(i < 30) { - char *lightsn_array[] = {lightsn1, lightsn2, lightsn3, lightsn4, lightsn5, - lightsn6, lightsn7, lightsn8, lightsn9, lightsn10, - lightsn11, lightsn12, lightsn13, lightsn14, lightsn15, - lightsn16, lightsn17, lightsn18, lightsn19, lightsn20, - lightsn21, lightsn22, lightsn23, lightsn24, lightsn25, - lightsn26, lightsn27, lightsn28, lightsn29, lightsn30}; - memset(lightsn_array[i], 0, 7); - memcpy(lightsn_array[i], mqtt_parm.msg_sn, strlen(mqtt_parm.msg_sn)); - LOG_I("lightsn[%d] = %s\n", i, lightsn_array[i]); - } - } - - // 设置lightbars_size为items_size - lightbars_size = items_size; - LOG_I("lightbars_size = %d\n", lightbars_size); + } else { + // OTA任务或其他任务 + items_size = 0; + lightbars_size = 0; + } // 根据任务类型进行处理 // 任务1：发布灯条指令 (topic: /estation/设备ID/task) @@ -2157,8 +2189,9 @@ void *thread_mqtt_recv(void *arg){ isBindTag = false; // P点灯 B绑定 &群控(不支持) *根据id点亮 - // timeout单位是5秒，所以 Time(秒) / 5 = timeout - uart_data_send_head(&uartSend, 'P', 5, mqtt_parm.msg_duration/5, lightbars_size); + // timeout单位是5秒，所以 Time(秒) / 5 = timeout + // 只发送头包，数据包在thread_uart_recv_ack中处理 + uart_data_send_head(&uartSend, 'P', 5, mqtt_parm.msg_duration/5, lightbars_size); } else { LOG_I("Processing OTA task (task_id: ota)\n"); diff --git a/mqtt_utils/mqtt_utils.c b/mqtt_utils/mqtt_utils.c index 359c1a1..4a2c19a 100644 --- a/mqtt_utils/mqtt_utils.c +++ b/mqtt_utils/mqtt_utils.c @@ -28,6 +28,9 @@ #include "mqtt_utils.h" +// 全局变量，用于存储从topic解析出的deviceName +char g_mqtt_deviceName[256] = {0}; + #include "main.h" #define PRINT_TIME_TAG #define DBG_TAG "mqtt_utils" @@ -161,16 +164,120 @@ void mqtt_utils_connection_lost(void *context, char *cause){ mqtt_net_failure(&connect_failure_times); } +static int mqtt_utils_topic_match(const char *filter, const char *topic) +{ + size_t fi = 0; + size_t ti = 0; + + if (filter == NULL || topic == NULL) { + return 0; + } + + while (1) { + size_t fseg_len = 0; + size_t tseg_len = 0; + + while (filter[fi + fseg_len] != '\0' && filter[fi + fseg_len] != '/') { + fseg_len++; + } + while (topic[ti + tseg_len] != '\0' && topic[ti + tseg_len] != '/') { + tseg_len++; + } + + if (fseg_len == 1 && filter[fi] == '#') { + return 1; + } + + if (fseg_len == 1 && filter[fi] == '+') { + } else { + if (fseg_len != tseg_len) { + return 0; + } + if (strncmp(filter + fi, topic + ti, fseg_len) != 0) { + return 0; + } + } + + fi += fseg_len; + ti += tseg_len; + + if (filter[fi] == '\0' && topic[ti] == '\0') { + return 1; + } + + if (filter[fi] == '\0' || topic[ti] == '\0') { + return 0; + } + + if (filter[fi] != '/' || topic[ti] != '/') { + return 0; + } + fi++; + ti++; + } +} + +static int mqtt_utils_parse_sys_lightOperate_invoke_topic(const char *topic, char *out1, size_t out1_len, char *out2, + size_t out2_len) +{ + char tmp1[256] = {0}; + char tmp2[256] = {0}; + + if (topic == NULL || out1 == NULL || out2 == NULL || out1_len == 0 || out2_len == 0) { + return -1; + } + + if (sscanf(topic, "/sys/%255[^/]/%255[^/]/thing/service/lightOperate/invoke", tmp1, tmp2) != 2) { + return -1; + } + + snprintf(out1, out1_len, "%s", tmp1); + snprintf(out2, out2_len, "%s", tmp2); + return 0; +} + int mqtt_utils_message_arrived(void *context, char *topicName, int topicLen, MQTTAsync_message *message){ LOG_I("---------------%s------------------\n",__func__); - LOG_I("收到消息，topic: %s\n", topicName); + LOG_I("收到消息，topic: %s\n", topicName ? topicName : "(null)"); // 检查消息是否来自任何一个订阅的topic int topic_matched = 0; for(int i = 0; i < subscribeTopicCount; i++) { - if(strcmp(topicName, subscribeTopics[i]) == 0) { + int matched = 0; + if (strchr(subscribeTopics[i], '+') != NULL || strchr(subscribeTopics[i], '#') != NULL) { + matched = mqtt_utils_topic_match(subscribeTopics[i], topicName); + } else { + matched = (strcmp(topicName, subscribeTopics[i]) == 0); + } + + if (matched) { + if (strcmp(subscribeTopics[i], "/sys/+/+/thing/service/lightOperate/invoke") == 0) { + char field1[256] = {0}; + char field2[256] = {0}; + if (mqtt_utils_parse_sys_lightOperate_invoke_topic(topicName, field1, sizeof(field1), field2, + sizeof(field2)) == 0) { + LOG_I("lightOperate/invoke 匹配字段: productKey=%s, deviceName=%s\n", field1, field2); + // 将deviceName存储到全局变量中 + memset(g_mqtt_deviceName, 0, sizeof(g_mqtt_deviceName)); + strncpy(g_mqtt_deviceName, field2, sizeof(g_mqtt_deviceName) - 1); + } else { + LOG_I("lightOperate/invoke topic字段解析失败: %s\n", topicName); + } + } + LOG_I("消息匹配topic[%d]: %s\n", i, subscribeTopics[i]); - PutDataIntoMQueue(message->payload); + if (message != NULL && message->payload != NULL && message->payloadlen > 0) { + char payload_buf[1024] = {0}; + size_t copy_len = (size_t)message->payloadlen; + if (copy_len >= sizeof(payload_buf)) { + copy_len = sizeof(payload_buf) - 1; + } + memcpy(payload_buf, message->payload, copy_len); + payload_buf[copy_len] = '\0'; + PutDataIntoMQueue(payload_buf); + } else { + LOG_I("消息payload为空，跳过入队\n"); + } topic_matched = 1; break; } @@ -181,7 +288,9 @@ int mqtt_utils_message_arrived(void *context, char *topicName, int topicLen, MQT } MQTTAsync_freeMessage(&message); - MQTTAsync_free(topicName); + if (topicName != NULL) { + MQTTAsync_free(topicName); + } return 1; } @@ -208,19 +317,19 @@ void mqtt_utils_connected(void *context, char *cause){ LOG_I("订阅topic[%d]请求结果: %d\n", i, subscribe_result); } - // 添加通配 符订阅来接收所有消息（用于调试） + // 添加通配符订阅来接收所有消息（用于调试） LOG_I("=== 添加通配符订阅用于调试 ===\n"); int wildcard_subscribe_result = mqtt_utils_subscribe(mqtt_conf, "#", 1); LOG_I("通配符订阅请求结果: %d\n", wildcard_subscribe_result); LOG_I("=== 通配符订阅完成 ===\n"); - // 订阅公共 topic用于测试 + // 订阅公共topic用于测试 LOG_I("=== 订阅公共topic用于测试 ===\n"); int public_subscribe_result = mqtt_utils_subscribe(mqtt_conf, "/test/public/topic", 0); LOG_I("公共topic订阅请求结果: %d\n", public_subscribe_result); LOG_I("=== 公共topic订阅完成 ===\n"); - system( "echo 1 > /sys/class/gpio/gpio113/value");//yellow ok + system("echo 1 > /sys/class/gpio/gpio113/value");//yellow ok station_status_report(); LOG_I("=== mqtt_utils_connected 函数执行完成 ===\n"); } @@ -259,7 +368,7 @@ void mqtt_utils_on_publish_failure(void *context, MQTTAsync_failureData *respons LOG_I("%s rc %s\n",__func__, MQTTAsync_strerror(response->code)); - // 如果是连 接相关的问题，记录详细信息 + // 如果是连接相关的问题，记录详细信息 if (response && response->code == MQTTASYNC_OPERATION_INCOMPLETE) { LOG_I("发布失败：操作在完成前被丢弃，可能是连接状态不稳定\n"); } else if (response && response->code == MQTTASYNC_DISCONNECTED) { @@ -299,21 +408,21 @@ int mqtt_utils_subscribe(mqtt_utils_t *mqtt_utils, char *topic, int qos){ return -1; } - // 跳过连接状态检查，直接进行 订阅 + // 跳过连接状态检查，直接进行订阅 LOG_I("跳过连接状态检查，直接进行订阅\n"); LOG_I("准备调用 MQTTAsync_subscribe...\n"); - rc = MQTTAsync_subscribe(mqtt_utils->client, topic, qos, &mqtt_utils ->sub_opts); + rc = MQTTAsync_subscribe(mqtt_utils->client, topic, qos, &mqtt_utils->sub_opts); LOG_I("MQTTAsync_subscribe 调用完成，返回值: %d\n", rc); - if (rc != MQTTASYNC_SUCCESS){ - LOG_I("订阅失败，错误代码: %s\n ", MQTTAsync_strerror(rc)); + if (rc != MQTTASYNC_SUCCESS){ + LOG_I("订阅失败，错误代码: %s\n", MQTTAsync_strerror(rc)); } else { LOG_I("订阅请求已发送，等待回调\n"); } return rc; } -int mqtt_utils_publish(mqtt_utils_t *mqtt_utils, char *topic, int qos, const char *data, int datalen){ +int mqtt_utils_publish(mqtt_utils_t *mqtt_utils, char *topic, int qos, const char *data, int datalen){ int rc; int isConnected = 0; @@ -358,6 +467,7 @@ int mqtt_utils_init(mqtt_utils_t *mqtt_config) sprintf(subscribeTopics[subscribeTopicCount++], "/iot/estation%s/group", mqtt_conf->username); sprintf(subscribeTopics[subscribeTopicCount++], "/iot%s/thing/ota/upgrade", mqtt_conf->username); sprintf(subscribeTopics[subscribeTopicCount++], "/sys/WcSubLightStrip/AD1000014C11/thing/service/lightOperate/invoke"); + sprintf(subscribeTopics[subscribeTopicCount++], "/sys/+/+/thing/service/lightOperate/invoke"); LOG_I("设置了%d个订阅topic:\n", subscribeTopicCount); for(int i = 0; i < subscribeTopicCount; i++) { diff --git a/mqtt_utils/mqtt_utils.h b/mqtt_utils/mqtt_utils.h index b1057d7..56a5305 100644 --- a/mqtt_utils/mqtt_utils.h +++ b/mqtt_utils/mqtt_utils.h @@ -67,6 +67,7 @@ typedef struct extern mqtt_utils_t mqtt_config; extern void PutDataIntoMQueue(char *payload); +extern char g_mqtt_deviceName[256]; typedef struct{ char msg_messageId[32];