From 5988dd6ddd72d6d725904ce7873b33496feac081 Mon Sep 17 00:00:00 2001 From: hcj <1224667075@qq.com> Date: Tue, 30 Dec 2025 10:50:25 +0800 Subject: [PATCH] =?utf8?q?ws=E5=A2=9E=E5=8A=A0=E5=8F=91=E9=80=81=E9=98=9F?= =?utf8?q?=E5=88=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit --- .../zhihui/websocket/os_ws/haisi_websocket.c | 238 ++++++++++++++++-- 1 file changed, 216 insertions(+), 22 deletions(-) diff --git a/code/sdk/zhihui/websocket/os_ws/haisi_websocket.c b/code/sdk/zhihui/websocket/os_ws/haisi_websocket.c index 14af74d5d..99f4e2318 100644 --- a/code/sdk/zhihui/websocket/os_ws/haisi_websocket.c +++ b/code/sdk/zhihui/websocket/os_ws/haisi_websocket.c @@ -55,6 +55,15 @@ static const char *g_log_level_str[] = { #define WS_MSG_TYPE_STOP 0x03 #define WS_MSG_TYPE_RAW_MESSAGE 0x04 +/* 发送方式控制宏 */ +#ifndef WS_USE_SEND_QUEUE +#define WS_USE_SEND_QUEUE 1 // 1=使用发送队列, 0=不使用发送队列 +#endif + +#ifndef WS_SEND_DIRECT_ENABLE +#define WS_SEND_DIRECT_ENABLE 0 // 0=通过队列发送, 1=直接发送（仅在WS_USE_SEND_QUEUE=0时有效） +#endif + /* 连接状态定义 */ #define WS_STATE_DISCONNECTED 0 // 未连接 #define WS_STATE_CONNECTING 1 // 连接中 @@ -85,9 +94,11 @@ typedef struct { static int g_ws_client_fd = -1; static volatile int g_ws_connected = WS_STATE_DISCONNECTED; static uint32_t g_ws_main_thread = 0; +static uint32_t g_ws_send_thread = 0; static volatile int g_ws_running = 0; static volatile unsigned int g_ws_reconnect_count = 0; /* 断线后递增，用于URL reconnect参数 */ static osMessageQueueId_t g_ws_msg_queue = NULL; +static osMessageQueueId_t g_ws_send_queue = NULL; static ws_call_backs_t g_ws_callbacks = {0}; /* WebSocket回调函数结构体 */ static volatile ws_close_status_t g_ws_close_status = WS_CLOSE_NOT_EXECUTED; /* WebSocket关闭状态标志位 */ volatile int g_ws_is_reconnecting = 0; /* 是否正在重连的标志位 */ @@ -172,6 +183,18 @@ void uapi_ws_queue_init(void) } } +/* 发送队列创建函数 */ +static void uapi_ws_send_queue_init(void) +{ + HAISI_WS_DEBUG("uapi_ws_send_queue_init\n"); + g_ws_send_queue = osMessageQueueNew(WS_MSG_QUEUE_COUNT, sizeof(ws_msg_info_t*), NULL); + if (g_ws_send_queue == NULL) { + HAISI_WS_ERROR("uapi_ws_send_queue_init failed: failed to create send queue\n"); + } else { + HAISI_WS_DEBUG("uapi_ws_send_queue_init success: send queue created\n"); + } +} + /* 握手成功回调：标记连接就绪 */ static void uapi_ws_on_open(int client) { @@ -353,15 +376,18 @@ int last_connection_state = WS_STATE_DISCONNECTED; // 记录上次连接状态 if (osMessageQueueGet(g_ws_msg_queue, &msg, NULL, 0) == osOK) { HAISI_WS_DEBUG("osMessageQueueGet: g_ws_msg_queue\n"); switch (msg->msg_type) { + // WS_MSG_TYPE_SEND_TEXT 由发送线程处理，这里放回队列 case WS_MSG_TYPE_SEND_TEXT: - if (g_ws_connected == WS_STATE_CONNECTED && g_ws_client_fd >= 0) { - // 对于文本消息，我们需要特殊处理，因为现在data是ws_data_msg_t类型 - // 这里需要重新设计文本消息的存储方式 - HAISI_WS_WARN("Text message type not supported in new structure\n"); - } else { - HAISI_WS_WARN("Cannot send text, connection state: %d\n", g_ws_connected); - } +#if WS_USE_SEND_QUEUE + // 发送消息由发送线程处理，放回队列 + osMessageQueuePut(g_ws_send_queue, &msg, 0, 0); + msg = NULL; +#else + // 发送队列未启用，直接释放消息 + HAISI_WS_WARN("Send queue disabled, dropping SEND_TEXT message\n"); ws_msg_free(msg); + msg = NULL; +#endif break; case WS_MSG_TYPE_FORCE_RECONNECT: @@ -638,19 +664,72 @@ int last_connection_state = WS_STATE_DISCONNECTED; // 记录上次连接状态 // g_ws_client_fd=-1 } g_ws_connected = WS_STATE_DISCONNECTED; +} + +/** + * @brief WebSocket发送线程 + * 负责处理发送队列中的消息 + */ +static void uapi_ws_send_thread(void *arg) +{ + (void)arg; + ws_msg_info_t *msg = NULL; - // 删除消息队列 - if (g_ws_msg_queue != NULL) { - ws_msg_info_t msg; - while (osMessageQueueGet(g_ws_msg_queue, &msg, NULL, 0) == osOK) { - // 原始消息数据不需要额外释放，因为存储在栈上 - HAISI_WS_DEBUG("Clearing queued message: %s\n", msg.raw_data); - ws_msg_free(&msg); + HAISI_WS_INFO("WebSocket send thread started\n"); + + while (g_ws_running) { + // 从发送队列获取消息（阻塞等待，超时100ms） + osStatus_t status = osMessageQueueGet(g_ws_send_queue, &msg, NULL, 100); + + if (status == osOK && msg != NULL) { + if (msg->msg_type == WS_MSG_TYPE_STOP) { + HAISI_WS_INFO("Send thread received stop signal\n"); + ws_msg_free(msg); + break; + } + // 发送线程只处理发送消息类型 + if (msg->msg_type == WS_MSG_TYPE_SEND_TEXT) { + // 检查连接状态 + if (g_ws_connected == WS_STATE_CONNECTED && g_ws_client_fd >= 0) { + if (msg->raw_data && msg->data_len > 0) { + HAISI_WS_DEBUG("Sending text message: %s, len: %u\n", msg->raw_data, msg->data_len); + + // 设置发送状态标志 + g_ws_is_sending = 1; + + // 发送数据 + int ret = WebSocketClientSendText(g_ws_client_fd, msg->raw_data, msg->data_len); + + HAISI_WS_DEBUG("Send text message ret: %d\n", ret); + + // 更新最后发送时间 + if (ret == 0) { + uapi_ws_update_tx_time(); + } + + // 清除发送状态标志 + g_ws_is_sending = 0; + } else { + HAISI_WS_WARN("Invalid text message data\n"); + } + } else { + HAISI_WS_WARN("Cannot send text, connection state: %d, fd: %d\n", g_ws_connected, g_ws_client_fd); + } + + // 释放消息内存 + ws_msg_free(msg); + msg = NULL; + } else { + // 其他消息类型放回队列或直接释放 + HAISI_WS_WARN("Send thread: Unknown message type %d, dropping\n", msg->msg_type); + ws_msg_free(msg); + msg = NULL; + } } - osMessageQueueDelete(g_ws_msg_queue); - g_ws_msg_queue = NULL; } + + HAISI_WS_INFO("WebSocket send thread exited\n"); } /* * 启动：开启监督线程与心跳机制（心跳线程首次连接成功后会自然启动） @@ -730,6 +809,24 @@ int uapi_ws_start(const ws_call_backs_t *callbacks) }; (void)uapi_create_osThread_custom_with_args(&main_task); } + #if WS_USE_SEND_QUEUE + // 创建发送线程 + if (!g_ws_send_thread) { + app_task_definition_ext_t send_task = { + .attr = { + .name = "uapi_ws_send", + .stack_size = 0x2000, + .priority = osPriorityNormal + }, + .func = uapi_ws_send_thread, + .task_handle = &g_ws_send_thread, + .create_queue_handle = uapi_ws_send_queue_init, // 发送队列已初始化，不需要单独创建 + .use_mem = USE_EXTEND_MEM, + .inst_args = NULL + }; + (void)uapi_create_osThread_custom_with_args(&send_task); + } + #endif return 0; } @@ -738,6 +835,39 @@ int uapi_ws_start(const ws_call_backs_t *callbacks) int uapi_ws_send_text(const char *text) { if (!text) return -1; + +#if (!WS_SEND_DIRECT_ENABLE) && WS_USE_SEND_QUEUE + // 通过队列发送模式（当WS_SEND_DIRECT_ENABLE=0 && WS_USE_SEND_QUEUE=1时） + if (!g_ws_running || !g_ws_send_queue) { + HAISI_WS_WARN("Cannot send text via queue, ws_running: %d, send_queue: %p\n", g_ws_running, g_ws_send_queue); + return -1; + } + + size_t text_len = strlen(text); + ws_msg_info_t *msg = ws_msg_create(text_len); + if (msg == NULL) { + HAISI_WS_ERROR("Failed to create send message\n"); + return -1; + } + + msg->msg_type = WS_MSG_TYPE_SEND_TEXT; + msg->timestamp = 0; + msg->isBinary = false; + memcpy(msg->raw_data, text, text_len); + msg->raw_data[text_len] = '\0'; + msg->data_len = text_len; + + osStatus_t ret = osMessageQueuePut(g_ws_send_queue, &msg, 0, 0); + if (ret != osOK) { + HAISI_WS_ERROR("Failed to put send message to queue: %d\n", ret); + ws_msg_free(msg); + return -1; + } + + HAISI_WS_DEBUG("Text message queued for sending: %s\n", text); + return 0; +#else + // 直接发送模式（WS_SEND_DIRECT_ENABLE=1 或 WS_USE_SEND_QUEUE=0） int network_status = uapi_dev_runtime_get_cat1_networkstat(); if (network_status != 1 || g_ws_connected != WS_STATE_CONNECTED || g_ws_client_fd < 0) { HAISI_WS_WARN("Cannot send text, connection state: %d , network_status: %d\n", g_ws_connected,network_status); @@ -750,7 +880,7 @@ int uapi_ws_send_text(const char *text) // 设置发送状态标志 g_ws_is_sending = 1; - HAISI_WS_DEBUG("uapi_ws_send_text: %s\n", text); + HAISI_WS_DEBUG("uapi_ws_send_text (direct): %s\n", text); // 发送数据 HAISI_WS_WARN("uapi_ws_send_text \n"); int ret = WebSocketClientSendText(g_ws_client_fd, text, strlen(text)); @@ -764,7 +894,10 @@ int uapi_ws_send_text(const char *text) g_ws_is_sending = 0; return ret; +#endif } + + uint32_t size_uapi_ws_send_BIN=0; int uapi_ws_send_BIN(uint8_t *data,int len) @@ -825,12 +958,37 @@ send_end: /* 通过队列发送文本消息（异步） */ int uapi_ws_send_text_async(const char *text) { - if (!text || !g_ws_running) return -1; +#if WS_USE_SEND_QUEUE + if (!text || !g_ws_running || !g_ws_send_queue) return -1; - // 由于新的数据结构不支持直接文本消息，这里暂时返回错误 - // 如果需要支持文本消息，需要重新设计消息类型 - HAISI_WS_WARN("Text message not supported in new queue structure\n"); - return -1; + size_t text_len = strlen(text); + ws_msg_info_t *msg = ws_msg_create(text_len); + if (msg == NULL) { + HAISI_WS_ERROR("Failed to create send message\n"); + return -1; + } + + msg->msg_type = WS_MSG_TYPE_SEND_TEXT; + msg->timestamp = 0; + msg->isBinary = false; + memcpy(msg->raw_data, text, text_len); + msg->raw_data[text_len] = '\0'; + msg->data_len = text_len; + + osStatus_t ret = osMessageQueuePut(g_ws_send_queue, &msg, 0, 0); + if (ret != osOK) { + HAISI_WS_ERROR("Failed to put send message to queue: %d\n", ret); + ws_msg_free(msg); + return -1; + } + + HAISI_WS_DEBUG("Text message queued for sending: %s\n", text); + return 0; +#else + // 发送队列未启用，直接调用同步发送函数 + HAISI_WS_WARN("Send queue disabled, calling uapi_ws_send_text directly\n"); + return uapi_ws_send_text(text); +#endif } /* 强制重连 */ @@ -885,10 +1043,31 @@ int uapi_ws_stop_async(void) int uapi_ws_stop(void) { g_ws_running = 0; + +#if WS_USE_SEND_QUEUE + // 发送停止消息到发送队列 + if (g_ws_send_queue != NULL) { + ws_msg_info_t *stop_msg = ws_msg_create(0); + if (stop_msg != NULL) { + stop_msg->msg_type = WS_MSG_TYPE_STOP; + stop_msg->timestamp = 0; + stop_msg->isBinary = false; + osMessageQueuePut(g_ws_send_queue, &stop_msg, 0, 0); + } + } +#endif + if (g_ws_main_thread) { osThreadJoin(g_ws_main_thread); g_ws_main_thread = 0; } + +#if WS_USE_SEND_QUEUE + if (g_ws_send_thread) { + osThreadJoin(g_ws_send_thread); + g_ws_send_thread = 0; + } +#endif int ret = 0; if (g_ws_client_fd >= 0) { ret = uapi_ws_close_with_status(g_ws_client_fd); @@ -910,6 +1089,21 @@ int uapi_ws_stop(void) g_ws_msg_queue = NULL; } +#if WS_USE_SEND_QUEUE + // 删除发送队列（先清理所有消息） + if (g_ws_send_queue != NULL) { + ws_msg_info_t *msg; + while (osMessageQueueGet(g_ws_send_queue, &msg, NULL, 0) == osOK) { + if (msg != NULL) { + HAISI_WS_DEBUG("Clearing queued send message before stop\n"); + ws_msg_free(msg); + } + } + osMessageQueueDelete(g_ws_send_queue); + g_ws_send_queue = NULL; + } +#endif + return ret; } -- 2.17.1