From aad1bbf14a2f597320ea88e2ff8d168fed2f0549 Mon Sep 17 00:00:00 2001 From: =?utf8?q?=E5=AD=99=E6=98=A5=E9=98=B3?= Date: Mon, 11 Aug 2025 12:32:36 +0800 Subject: [PATCH] =?utf8?q?=E4=B8=AD=E9=97=B4=E7=89=88=E6=9C=AC=EF=BC=8Cai?= =?utf8?q?=5Fserver,cat1=5Fws,=20tcp=5Fsimple=5Fmanager=20=E9=83=BD?= =?utf8?q?=E5=B7=B2=E7=BB=8F=E6=B7=BB=E5=8A=A0=E5=A5=BD=EF=BC=8Cws?= =?utf8?q?=E8=BF=98=E6=9C=89=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit --- .../brandy_evb/linker/standard/acore.prelds | 2 - .../slip/cat1Service/bk_websocket.h | 70 +- .../slip/cat1Service/cat1_ai_server.cpp | 423 +++++++++++- .../peripheral/slip/cat1Service/cat1_ws.cpp | 641 ++++++++++-------- .../slip/tcpmanager/tcp_simple_manager.c | 233 ++++++- .../slip/tcpmanager/tcp_simple_manager.h | 76 ++- .../drivers/peripheral/slip/websocket_test.c | 10 +- 7 files changed, 1113 insertions(+), 342 deletions(-) diff --git a/code/sdk/drivers/boards/brandy_evb/linker/standard/acore.prelds b/code/sdk/drivers/boards/brandy_evb/linker/standard/acore.prelds index f91df4686..7ba4a2b5d 100755 --- a/code/sdk/drivers/boards/brandy_evb/linker/standard/acore.prelds +++ b/code/sdk/drivers/boards/brandy_evb/linker/standard/acore.prelds @@ -96,7 +96,6 @@ SECTIONS SORT(*libmsg_center*.a:*.o*) (.data .data* .sdata .sdata*) SORT(*ohos*.a:*.o*) (.data .data* .sdata .sdata*) SORT(*liblwIP*.a:*.o*) (.data .data* .sdata .sdata*) - SORT(*libslip*.a:*.o*) (.data .data* .sdata .sdata*) SORT(*xts_test*.a:*.o*) (.data .data* .sdata .sdata*) *libnativelauncher*.a:*.o* (.data .data* .sdata .sdata*) SORT(*application/wearable*.a:*.o*)(.data .data* .sdata .sdata*) @@ -122,7 +121,6 @@ SECTIONS SORT(*libmsg_center*.a:*.o*) (.bss .bss* .sbss .sbss* COMMON) SORT(*ohos*.a:*.o*) (.bss .bss* .sbss .sbss* COMMON) SORT(*liblwIP*.a:*.o*) (.bss .bss* .sbss .sbss* COMMON) - SORT(*libslip*.a:*.o*) (.bss .bss* .sbss .sbss* COMMON) SORT(*xts_test*.a:*.o*) (.bss .bss* .sbss .sbss* COMMON) *libnativelauncher*.a:*.o* (.bss .bss* .sbss .sbss* COMMON) SORT(*application/wearable*.a:*.o*) (.bss .bss* .sbss .sbss* COMMON) diff --git a/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h b/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h index a3380a9cb..dbf6e3367 100644 --- a/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h +++ b/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h @@ -19,19 +19,87 @@ typedef enum { WS_PING = 0x9, WS_PONG = 0xa } WebSocketMessageType; + +/* WebSocket数据包回调函数类型 */ +typedef void (*on_ws_packet_callback)(uint8_t* data, size_t datalen, WebSocketMessageType type); + +/* WebSocket组包缓冲区配置 */ +#define WS_FRAME_BUFFER_SIZE 256 +#define WS_PAYLOAD_BUFFER_SIZE 128 + +/* WebSocket组包状态 */ +typedef enum { + WS_PARSE_HEADER, /* 解析帧头 */ + WS_PARSE_EXTENDED, /* 解析扩展长度 */ + WS_PARSE_MASK, /* 解析掩码 */ + WS_PARSE_PAYLOAD /* 解析载荷数据 */ +} ws_parse_state_t; + +/* WebSocket组包缓冲区结构 */ +typedef struct { + uint8_t frame_buffer[WS_FRAME_BUFFER_SIZE]; /* 帧缓冲区 */ + uint32_t frame_buffer_len; /* 帧缓冲区当前长度 */ + + uint8_t payload_buffer[WS_PAYLOAD_BUFFER_SIZE]; /* 载荷缓冲区 */ + uint32_t payload_buffer_len; /* 载荷缓冲区当前长度 */ + + ws_parse_state_t parse_state; /* 解析状态 */ + uint8_t fin; /* FIN标志 */ + WebSocketMessageType opcode; /* 操作码 */ + uint8_t mask; /* 掩码标志 */ + uint64_t payload_len; /* 载荷长度 */ + uint32_t payload_received; /* 已接收载荷长度 */ + uint8_t masking_key[4]; /* 掩码密钥 */ + + on_ws_packet_callback packet_callback; /* 数据包回调函数 */ +} ws_packet_parser_t; + +/* WebSocket消息队列系统 */ +#define WS_QUEUE_SIZE 4 +#define WS_MESSAGE_MAX_SIZE 512 + +typedef struct { + uint8_t data[WS_MESSAGE_MAX_SIZE]; + uint16_t length; + WebSocketMessageType type; + uint32_t timestamp; + bool in_use; +} ws_message_item_t; + +typedef struct { + ws_message_item_t messages[WS_QUEUE_SIZE]; + uint8_t head; + uint8_t tail; + uint8_t count; + osMutexId_t mutex; + osSemaphoreId_t semaphore; + bool initialized; +} ws_message_queue_t; + + // WebSocket 客户端结构体 typedef struct { - int sockfd; + void* tcp_handle; + ws_message_queue_t *msg_queue; char host[128]; char path[128]; char query[128]; int port; + + /* 组包解析器 */ + ws_packet_parser_t parser; } WebSocketClient; // 初始化 WebSocket 客户端 WebSocketClient *init_websocket_client(const char *url, const char *headers,uint8_t use_dns); +// 设置WebSocket数据包回调函数 +void ws_set_packet_callback(WebSocketClient *client, on_ws_packet_callback callback); + +// 处理接收到的原始数据 +int ws_process_received_data(WebSocketClient *client, const uint8_t *data, size_t len); + // 关闭 WebSocket 客户端 void lwip_close_websocket_client(WebSocketClient *client) ; diff --git a/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ai_server.cpp b/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ai_server.cpp index ac17fc3b0..28827b42e 100644 --- a/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ai_server.cpp +++ b/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ai_server.cpp @@ -18,6 +18,7 @@ #include // #include #include "lwip/inet.h" +#include "../tcpmanager/tcp_simple_manager.h" #include // #include // #include @@ -165,6 +166,13 @@ static uint8_t disconnect_server=0; // free +/* 各线程的消息队列 */ +static ws_message_queue_t chat_msg_queue = {0}; +static ws_message_queue_t asr_msg_queue = {0}; +static ws_message_queue_t tts_msg_queue = {0}; +// static ws_message_queue_t voice_msg_queue = {0}; + + char *asr_recv_buf=NULL;//[1024]={0}; WebSocketClient *asr_client=NULL; @@ -193,7 +201,27 @@ static int tts_client_ping=0; static int chat_client_ping=0; - +WebSocketClient * get_ws_client_by_handle(void* handle) +{ + if (chat_client != NULL && chat_client->tcp_handle == handle) + { + return chat_client; + } + else if (asr_client != NULL && asr_client->tcp_handle == handle) + { + return asr_client; + } + else if (tts_client != NULL && tts_client->tcp_handle == handle) + { + return tts_client; + } + // 注意：voice_chat_client 被注释掉了，根据用户的修改 + // else if (voice_chat_client != NULL && voice_chat_client->tcp_handle == handle) + // { + // return voice_chat_client; + // } + return NULL; +} errcode_t chat_server_init(void); errcode_t tts_server_init(void); @@ -201,6 +229,188 @@ errcode_t tts_server_init(void); // void tts_server_deinit(void); uint8 tts_break =0; +/* WebSocket消息队列管理函数 */ + +/** + * @brief 初始化WebSocket消息队列 + */ +static int ws_queue_init(ws_message_queue_t* queue) +{ + if (!queue) return -1; + + memset(queue, 0, sizeof(ws_message_queue_t)); + + /* 创建互斥锁 */ + queue->mutex = osMutexNew(NULL); + if (!queue->mutex) { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to create mutex for ws queue"); + return -1; + } + + /* 创建信号量 */ + queue->semaphore = osSemaphoreNew(WS_QUEUE_SIZE, 0, NULL); + if (!queue->semaphore) { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to create semaphore for ws queue"); + osMutexDelete(queue->mutex); + return -1; + } + + queue->initialized = true; + WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket queue initialized successfully"); + return 0; +} + +/** + * @brief 反初始化WebSocket消息队列 + */ +static void ws_queue_deinit(ws_message_queue_t* queue) +{ + if (!queue || !queue->initialized) return; + + if (queue->mutex) { + osMutexDelete(queue->mutex); + queue->mutex = NULL; + } + + if (queue->semaphore) { + osSemaphoreDelete(queue->semaphore); + queue->semaphore = NULL; + } + + memset(queue, 0, sizeof(ws_message_queue_t)); + WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket queue deinitialized"); +} + +/** + * @brief 向队列中添加消息 + */ +static int ws_queue_put(ws_message_queue_t* queue, const uint8_t* data, uint16_t length, WebSocketMessageType type) +{ + if (!queue || !queue->initialized || !data || length == 0 || length > WS_MESSAGE_MAX_SIZE) { + return -1; + } + + /* 获取互斥锁 */ + osStatus_t status = osMutexAcquire(queue->mutex, 1000); + if (status != osOK) { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to acquire mutex for ws queue put"); + return -1; + } + + /* 检查队列是否已满 */ + if (queue->count >= WS_QUEUE_SIZE) { + osMutexRelease(queue->mutex); + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WebSocket queue is full"); + return -1; + } + + /* 添加消息到队列 */ + ws_message_item_t* item = &queue->messages[queue->tail]; + memcpy(item->data, data, length); + item->length = length; + item->type = type; + item->timestamp = osKernelGetTickCount(); + item->in_use = true; + + queue->tail = (queue->tail + 1) % WS_QUEUE_SIZE; + queue->count++; + + osMutexRelease(queue->mutex); + + /* 通知有新消息 */ + osSemaphoreRelease(queue->semaphore); + + return 0; +} + +/** + * @brief 从队列中获取消息 + */ +static int ws_queue_get(ws_message_queue_t* queue, uint8_t* data, uint16_t max_length, WebSocketMessageType* type, uint32_t timeout_ms) +{ + if (!queue || !queue->initialized || !data || !type) { + return -1; + } + + /* 等待信号量 */ + osStatus_t status = osSemaphoreAcquire(queue->semaphore, timeout_ms); + if (status != osOK) { + return 0; /* 超时或错误 */ + } + + /* 获取互斥锁 */ + status = osMutexAcquire(queue->mutex, 1000); + if (status != osOK) { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to acquire mutex for ws queue get"); + return -1; + } + + /* 检查队列是否为空 */ + if (queue->count == 0) { + osMutexRelease(queue->mutex); + return 0; + } + + /* 获取消息 */ + ws_message_item_t* item = &queue->messages[queue->head]; + if (!item->in_use) { + osMutexRelease(queue->mutex); + return 0; + } + + uint16_t copy_len = (item->length > max_length) ? max_length : item->length; + memcpy(data, item->data, copy_len); + *type = item->type; + + /* 标记为已使用 */ + item->in_use = false; + queue->head = (queue->head + 1) % WS_QUEUE_SIZE; + queue->count--; + + osMutexRelease(queue->mutex); + + return copy_len; +} + +/** + * @brief 初始化所有WebSocket消息队列 + */ +static int ws_queues_init_all(void) +{ + int ret = 0; + + ret |= ws_queue_init(&chat_msg_queue); + ret |= ws_queue_init(&asr_msg_queue); + ret |= ws_queue_init(&tts_msg_queue); + // ret |= ws_queue_init(&voice_msg_queue); + + if (ret != 0) { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to initialize all WebSocket queues"); + /* 清理已初始化的队列 */ + ws_queue_deinit(&chat_msg_queue); + ws_queue_deinit(&asr_msg_queue); + ws_queue_deinit(&tts_msg_queue); + // ws_queue_deinit(&voice_msg_queue); + return -1; + } + + WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "All WebSocket queues initialized successfully"); + return 0; +} + + +/** + * @brief 反初始化所有WebSocket消息队列 + */ +static void ws_queues_deinit_all(void) +{ + ws_queue_deinit(&chat_msg_queue); + ws_queue_deinit(&asr_msg_queue); + ws_queue_deinit(&tts_msg_queue); + // ws_queue_deinit(&voice_msg_queue); + WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "All WebSocket queues deinitialized"); +} + void uapi_set_tts_break(uint8_t data) { tts_break=data; @@ -527,7 +737,34 @@ uint8_t get_bk_websocket_tts_tick(void) { return bk_websocket_tts_tick; } +void asr_on_ws_packet_callback(uint8_t* data, size_t datalen, WebSocketMessageType type) +{ + if(asr_client !=NULL && asr_client->msg_queue !=NULL) + ws_queue_put(asr_client->msg_queue,data,datalen,type); +} +void tts_on_ws_packet_callback(uint8_t* data, size_t datalen, WebSocketMessageType type) +{ + if(tts_client !=NULL && tts_client->msg_queue !=NULL) + ws_queue_put(tts_client->msg_queue,data,datalen,type); +} +void chat_on_ws_packet_callback(uint8_t* data, size_t datalen, WebSocketMessageType type) +{ + if(chat_client !=NULL && chat_client->msg_queue !=NULL) + ws_queue_put(chat_client->msg_queue,data,datalen,type); +} + +/** + * @brief 从队列中接收WebSocket消息（替代原来的receive_websocket_message） + */ +static int receive_websocket_message_from_queue(ws_message_queue_t* queue, uint8_t* buffer, uint16_t buffer_size, WebSocketMessageType* type, uint32_t timeout_ms) +{ + if (!queue || !buffer || !type) { + return -1; + } + + return ws_queue_get(queue, buffer, buffer_size, type, timeout_ms); +} void bk_websocket_demo(void *data) { uint32_t start_time = (uint32_t)osKernelGetTickCount(); uint32_t retry_count = 0; @@ -570,9 +807,13 @@ void bk_websocket_demo(void *data) { if (is_raisound_url) { chat_client = init_websocket_client(url_buf, NULL, 1); + chat_client->msg_queue = &chat_msg_queue; + chat_client->parser.packet_callback = chat_on_ws_packet_callback; } else { #ifdef SUPPORT_CHINA_MOBILE_AI_REGULATION chat_client = init_websocket_client(url_buf, NULL, 0); + chat_client->msg_queue = &chat_msg_queue; + chat_client->parser.packet_callback = chat_on_ws_packet_callback; #else const char* base_url = "ws://118.145.158.207:22080/chat/ws"; char full_url[512] = {0}; @@ -598,6 +839,8 @@ void bk_websocket_demo(void *data) { break; } chat_client = init_websocket_client(full_url, NULL, 0); + chat_client->msg_queue = &chat_msg_queue; + chat_client->parser.packet_callback = chat_on_ws_packet_callback; #endif } @@ -625,11 +868,12 @@ void bk_websocket_demo(void *data) { } // 设置socket超时 - struct timeval timeout; - timeout.tv_sec = AI_RECEIVE_TIMEOUT_MS / 1000; - timeout.tv_usec = (AI_RECEIVE_TIMEOUT_MS % 1000) * 1000; - lwip_setsockopt(chat_client->sockfd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)); - + // struct timeval timeout; + // timeout.tv_sec = AI_RECEIVE_TIMEOUT_MS / 1000; + // timeout.tv_usec = (AI_RECEIVE_TIMEOUT_MS % 1000) * 1000; + // lwip_setsockopt(chat_client->sockfd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)); + // tcp_set_timeout(chat_client->tcp_handle, AI_RECEIVE_TIMEOUT_MS); + server_state = 0x02 ^ server_state; AI_LOCK(); @@ -649,7 +893,7 @@ void bk_websocket_demo(void *data) { } memset_s(recv_buf, 1024, 0, 1024); - int len = receive_websocket_message(chat_client, (unsigned char *)recv_buf, 1024, &type); + int len = receive_websocket_message_from_queue(&chat_msg_queue, (unsigned char *)recv_buf, 1024, &type, AI_RECEIVE_TIMEOUT_MS); if (len == -1 || type == WS_CLOSE) { WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WebSocket connection closed"); @@ -1045,13 +1289,21 @@ void bk_websocket_asr(void *data) { if (g_test_server_index == 1) { asr_client = init_websocket_client("ws://118.145.158.207:22080/asr/ws", NULL, 0); + asr_client->msg_queue = &asr_msg_queue; + asr_client->parser.packet_callback = asr_on_ws_packet_callback; } else if (g_test_server_index == 2) { asr_client = init_websocket_client("ws://110.42.224.129:8910/asr/ws", NULL, 0); + asr_client->msg_queue = &asr_msg_queue; + asr_client->parser.packet_callback = asr_on_ws_packet_callback; } else { if (is_raisound_url) { asr_client = init_websocket_client(url_buf, NULL, 1); + asr_client->msg_queue = &asr_msg_queue; + asr_client->parser.packet_callback = asr_on_ws_packet_callback; } else { asr_client = init_websocket_client("ws://118.145.158.207:22080/asr/ws", NULL, 0); + asr_client->msg_queue = &asr_msg_queue; + asr_client->parser.packet_callback = asr_on_ws_packet_callback; } } @@ -1102,15 +1354,15 @@ void bk_websocket_asr(void *data) { // WEARABLE_LOGI(WEARABLE_LOG_MODULE_MAIN_SERVICE, "cat1 restart_time_sync timer ret %d", ret); // } - // 设置socket缓冲区大小 - int sndbuf_len = 1024 * 5; - lwip_setsockopt(asr_client->sockfd, SOL_SOCKET, SO_SNDBUF, &sndbuf_len, sizeof(sndbuf_len)); + // // 设置socket缓冲区大小 + // int sndbuf_len = 1024 * 5; + // lwip_setsockopt(asr_client->sockfd, SOL_SOCKET, SO_SNDBUF, &sndbuf_len, sizeof(sndbuf_len)); - // 设置接收超时 - struct timeval timeout; - timeout.tv_sec = AI_RECEIVE_TIMEOUT_MS / 1000; - timeout.tv_usec = (AI_RECEIVE_TIMEOUT_MS % 1000) * 1000; - lwip_setsockopt(asr_client->sockfd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)); + // // 设置接收超时 + // struct timeval timeout; + // timeout.tv_sec = AI_RECEIVE_TIMEOUT_MS / 1000; + // timeout.tv_usec = (AI_RECEIVE_TIMEOUT_MS % 1000) * 1000; + // lwip_setsockopt(asr_client->sockfd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)); uapi_start_asr(); @@ -1124,7 +1376,7 @@ void bk_websocket_asr(void *data) { } memset_s(asr_recv_buf, 1024, 0, 1024); - int len = receive_websocket_message(asr_client, (unsigned char *)asr_recv_buf, 1024, &type); + int len = receive_websocket_message_from_queue(&asr_msg_queue, (unsigned char *)asr_recv_buf, 1024, &type, AI_RECEIVE_TIMEOUT_MS); WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "ASR received: type=%d, len=%d", type, len); @@ -1465,8 +1717,12 @@ void bk_websocket_tts(void *data) { if (is_raisound_url == 1) { tts_client = init_websocket_client(url_buf, NULL, 1); + tts_client->msg_queue = &tts_msg_queue; + tts_client->parser.packet_callback = tts_on_ws_packet_callback; } else { tts_client = init_websocket_client("ws://118.145.158.207:22080/tts/stream?audio_buffer_size=1024", NULL, 0); + tts_client->msg_queue = &tts_msg_queue; + tts_client->parser.packet_callback = tts_on_ws_packet_callback; } if (tts_client == NULL) { @@ -1508,10 +1764,10 @@ void bk_websocket_tts(void *data) { AI_UNLOCK(); // 设置socket超时 - struct timeval timeout; - timeout.tv_sec = AI_RECEIVE_TIMEOUT_MS / 1000; - timeout.tv_usec = (AI_RECEIVE_TIMEOUT_MS % 1000) * 1000; - lwip_setsockopt(tts_client->sockfd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)); + // struct timeval timeout; + // timeout.tv_sec = AI_RECEIVE_TIMEOUT_MS / 1000; + // timeout.tv_usec = (AI_RECEIVE_TIMEOUT_MS % 1000) * 1000; + // lwip_setsockopt(tts_client->sockfd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)); uint32_t last_activity = (uint32_t)osKernelGetTickCount(); @@ -1523,7 +1779,7 @@ void bk_websocket_tts(void *data) { } memset_s(tts_recv_buf, 1024 * 4, 0, 1024 * 4); - int len = receive_websocket_message(tts_client, (unsigned char *)tts_recv_buf, 1024 * 4, &type); + int len = receive_websocket_message_from_queue(&tts_msg_queue, (unsigned char *)tts_recv_buf, 1024 * 4, &type, AI_RECEIVE_TIMEOUT_MS); if (type == WS_CLOSE || len == -1 || tts_reconnect == 1) { WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "TTS WebSocket connection closed"); @@ -2212,8 +2468,10 @@ void voice_chat_websocket(void *data) { if (is_raisound_url == 1) { voice_chat_client = init_websocket_client(url_buf, NULL, 1); + // voice_chat_client->msg_queue = &voice_msg_queue; } else { voice_chat_client = init_websocket_client("ws://118.145.158.207:22080/voice-chat/v2/ws?platform=huoshan", NULL, 0); + // voice_chat_client->msg_queue = &voice_msg_queue; } if (voice_chat_client == NULL) { @@ -2247,10 +2505,10 @@ void voice_chat_websocket(void *data) { ai_update_memory_usage(1024 * 4); // 设置socket超时 - struct timeval timeout; - timeout.tv_sec = AI_RECEIVE_TIMEOUT_MS / 1000; - timeout.tv_usec = (AI_RECEIVE_TIMEOUT_MS % 1000) * 1000; - lwip_setsockopt(voice_chat_client->sockfd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)); + // struct timeval timeout; + // timeout.tv_sec = AI_RECEIVE_TIMEOUT_MS / 1000; + // timeout.tv_usec = (AI_RECEIVE_TIMEOUT_MS % 1000) * 1000; + // lwip_setsockopt(voice_chat_client->sockfd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)); if (!start_ai_voice_chat()) { WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to start AI voice chat"); @@ -2269,7 +2527,8 @@ void voice_chat_websocket(void *data) { } memset_s(voice_chat_client_buf, 1024 * 4, 0, 1024 * 4); - int len = receive_websocket_message(voice_chat_client, (unsigned char *)voice_chat_client_buf, 1024 * 4, &type); + int len = -1; + // len = receive_websocket_message_from_queue(&voice_msg_queue, (unsigned char *)voice_chat_client_buf, 1024 * 4, &type, AI_RECEIVE_TIMEOUT_MS); if (type == WS_CLOSE || len == -1) { WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Voice chat WebSocket connection closed"); @@ -2348,18 +2607,13 @@ errcode_t voice_chat_server_init(void) } - - - - - - - - // 资源清理函数 void ai_cleanup_resources(void) { AI_LOCK(); + // 清理WebSocket消息队列 + ws_queues_deinit_all(); + // 清理互斥锁 if (g_ai_mutex) { osMutexDelete(g_ai_mutex); @@ -2377,21 +2631,41 @@ void ai_cleanup_resources(void) { AI_UNLOCK(); + // 清理TCP管理器 + tcp_simple_manager_deinit(); + WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "AI resources cleaned up"); } // 初始化AI系统 errcode_t ai_system_init(void) { + // 初始化TCP管理器 + tcp_manager_result_t tcp_result = tcp_simple_manager_init(); + if (tcp_result != TCP_MANAGER_OK) { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to initialize TCP manager: %d", tcp_result); + return ERRCODE_FAIL; + } + // 创建互斥锁 g_ai_mutex = osMutexNew(NULL); if (!g_ai_mutex) { WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to create AI mutex"); + tcp_simple_manager_deinit(); // 清理TCP管理器 return ERRCODE_FAIL; } // 初始化内存使用统计 g_ai_memory_usage = 0; + // 初始化WebSocket消息队列 + if (ws_queues_init_all() != 0) { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to initialize all WebSocket queues"); + osMutexDelete(g_ai_mutex); + tcp_simple_manager_deinit(); // 清理TCP管理器 + return ERRCODE_FAIL; + } + + WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "AI system initialized successfully"); return ERRCODE_SUCC; } @@ -2431,5 +2705,82 @@ bool ai_system_health_check(void) { } #ifdef __cplusplus -} -#endif /* __cplusplus */ \ No newline at end of file + } +#endif /* __cplusplus */ + +// /** +// * @brief WebSocket数据接收线程 +// * 负责从所有WebSocket连接接收数据并分发到对应队列 +// */ +// static void ws_receiver_thread(void* arg) +// { +// (void)arg; +// WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket receiver thread started"); + +// uint8_t temp_buffer[WS_MESSAGE_MAX_SIZE]; +// WebSocketMessageType type; + +// while (disconnect_server == 0) { +// bool any_activity = false; + +// /* 检查chat客户端 */ +// if (chat_client != NULL) { +// int len = receive_websocket_message(chat_client, temp_buffer, sizeof(temp_buffer), &type); +// if (len > 0) { +// any_activity = true; +// if (ws_queue_put(&chat_msg_queue, temp_buffer, len, type) != 0) { +// WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to put chat message to queue"); +// } +// } else if (len < 0) { +// WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Chat WebSocket connection error"); +// } +// } + +// /* 检查ASR客户端 */ +// if (asr_client != NULL) { +// int len = receive_websocket_message(asr_client, temp_buffer, sizeof(temp_buffer), &type); +// if (len > 0) { +// any_activity = true; +// if (ws_queue_put(&asr_msg_queue, temp_buffer, len, type) != 0) { +// WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to put ASR message to queue"); +// } +// } else if (len < 0) { +// WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "ASR WebSocket connection error"); +// } +// } + +// /* 检查TTS客户端 */ +// if (tts_client != NULL) { +// int len = receive_websocket_message(tts_client, temp_buffer, sizeof(temp_buffer), &type); +// if (len > 0) { +// any_activity = true; +// if (ws_queue_put(&tts_msg_queue, temp_buffer, len, type) != 0) { +// WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to put TTS message to queue"); +// } +// } else if (len < 0) { +// WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "TTS WebSocket connection error"); +// } +// } + +// /* 检查voice客户端 */ +// if (voice_chat_client != NULL) { +// int len = receive_websocket_message(voice_chat_client, temp_buffer, sizeof(temp_buffer), &type); +// if (len > 0) { +// any_activity = true; +// if (ws_queue_put(&voice_msg_queue, temp_buffer, len, type) != 0) { +// WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to put voice message to queue"); +// } +// } else if (len < 0) { +// WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Voice WebSocket connection error"); +// } +// } + +// /* 如果没有任何活动，短暂休眠避免CPU占用过高 */ +// if (!any_activity) { +// osDelay(10); /* 10ms */ +// } +// } + +// WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket receiver thread exited"); +// } + diff --git a/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ws.cpp b/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ws.cpp index c3bb8444e..f19097f2f 100644 --- a/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ws.cpp +++ b/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ws.cpp @@ -16,6 +16,11 @@ #include "lwip/sockets.h" #include "lwip/sys.h" #include + +/* 添加缺失的定义 */ +#ifndef SHUT_RDWR +#define SHUT_RDWR 2 +#endif // #include #include "lwip/inet.h" #include @@ -49,7 +54,7 @@ #include #include "cat1_ws.h" #include "dev_authentication.h" - +#include "tcp_simple_manager.h" #ifdef __cplusplus extern "C" { @@ -132,6 +137,7 @@ extern "C" { extern osStatus_t osDelay(uint32_t ticks); #define cat1_msleep(ms) LOS_Msleep(ms) /* osDelay(1) : 1ms */ +static void ws_parser_init(ws_packet_parser_t* parser); void bk_sleep(uint32_t time) @@ -407,9 +413,22 @@ void generate_accept_key(const char *key, char *accept_key) { } // 发送 WebSocket 握手请求 -int send_handshake(WebSocketClient *client,const char *headers) { +/** + * @brief 发送 WebSocket 握手请求 + * @param client 客户端指针 + * @param headers 头信息 + * @return 成功返回0，失败返回-1 + */ +int send_handshake(WebSocketClient *client,const char *headers,char *response,int response_len) { char key[WEBSOCKET_KEY_LEN]; printf("send_handshake1"); + + if(client==NULL|| response==NULL || response_len<=0) + { + printf("send_handshake client or response is NULL , or response_len <= 0\r\n"); + return -1; + } + generate_websocket_key(key); char accept_key[64]; int ret = 0; @@ -422,7 +441,6 @@ int send_handshake(WebSocketClient *client,const char *headers) { return -1; } - if(headers!= NULL) { snprintf(request, BUFFER_SIZE, @@ -451,12 +469,28 @@ int send_handshake(WebSocketClient *client,const char *headers) { } printf("send_handshake request:\r\n%s",request); - if (lwip_send(client->sockfd, request, strlen(request), 0) < 0) { - printf("lwip_send handshake failed"); - + //移植建议： 此处调用 tcp_send_data 进行发送, 不必使用 lwip_send，如果新能不行，就改用 tcp_send_data_directly + + + tcp_manager_result_t tcp_result = tcp_send_data_directly(client->tcp_handle, (uint8_t *)request, strlen(request)); + if(tcp_result != TCP_MANAGER_OK) + { + printf("tcp_send_data_directly failed: %d", tcp_result); ret = -1; } - + + // retlen = tcp_recv_data_directly(client->tcp_handle, (uint8_t *)response, response_len); + + // if(retlen<0) + // { + // printf("tcp_recv_data_directly failed"); + // ret = -1; + // } + // else + // { + // ret = retlen; + // } + free(request); printf("send_handshake3 %d",ret); return ret; @@ -518,7 +552,7 @@ int send_websocket_message(WebSocketClient *client, uint8_t *message, size_t mes // 发送帧 WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"offset %d\r\n",offset); - if (lwip_send(client->sockfd, frame, offset, 0) < 0) { + if (tcp_send_data(client->tcp_handle, frame, offset) < 0) { WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"lwip_send message failed\r\n"); ret = -1; goto send_websocket_message_ERROR; @@ -531,211 +565,40 @@ int send_websocket_message(WebSocketClient *client, uint8_t *message, size_t mes return ret; } -// 接收 WebSocket 消息,返回读到长度 -int receive_websocket_message(WebSocketClient *client, unsigned char *buf,int len,WebSocketMessageType *code) { - uint32_t timer_ms = (uint32_t)osKernelGetTickCount(); - //printf("receive_websocket_message 1 timer_ms=%d\n",timer_ms); - unsigned char frame[20]={0}; - unsigned char masking_key[4]; - int bytes_received = 0; - int recv_len = 0 ; - static uint8_t last_byte = 0; - // while(1) - // { - // recv_len = lwip_send(client->sockfd, frame+bytes_received, 2-bytes_received, 0); - // bytes_received +=recv_len; - // if(bytes_received>=2) - // break; - // } - int max_count=len; - - while(max_count) - { - bytes_received = lwip_recv(client->sockfd, frame, 1, 0); - - if (bytes_received <= 0) - { - if (bytes_received == 0) { - printf("Connection lwip_closed by server.\n"); - } else { - printf("receive message failed"); - } - return -1; - } -max_count--; - - if(frame[0]!= 0x80 && frame[0]!= 0x81 && frame[0]!= 0x82 && frame[0]!= 0x88 && frame[0]!= 0x89 && frame[0]!= 0x8A) - { - printf("receive_websocket_message:hend =0x%X,0x%X ,last_byte =0x%x ERROR!!!!!!!!!",frame[0],frame[1],last_byte); - - if(last_byte!= 0x80 && last_byte!= 0x81 && last_byte!= 0x82 && last_byte!= 0x88 && last_byte!= 0x89 && last_byte!= 0x8A) - { - printf("receive_websocket_message:hend =0x%X,0x%X ,last_byte =0x%x ERROR222222222!!!!!!!!!",frame[0],frame[1],last_byte); - continue; - } - if(frame[0]==0 && frame[1]==0 && last_byte==0) - { - break; - } - - } - else - { - break; - } - } -if(max_count==0) - { - return -1; - } - timer_ms = (uint32_t)osKernelGetTickCount(); - //printf("receive_websocket_message 2 timer_ms=%d\n",timer_ms); - bytes_received = lwip_recv(client->sockfd, frame+1, 1, 0); - if (bytes_received <= 0) { - if (bytes_received == 0) { - printf("Connection lwip_closed by server.\n"); - } else { - printf("receive message failed"); - } - return -1; - } - //unsigned char fin = (frame[0] & 0x80) >> 7; - *code = (WebSocketMessageType) (frame[0] & 0x0F); - unsigned char mask = (frame[1] & 0x80) >> 7; - unsigned int payload_len = frame[1] & 0x7F; - - - size_t offset = 2; - if (payload_len == 126) { - unsigned short extended_payload_len; - bytes_received += lwip_recv(client->sockfd, frame+offset, 2, 0); - memcpy(&extended_payload_len, frame + 2, 2); - payload_len = ntohs(extended_payload_len); - //printf("receive_websocket_message:payload_len=0x%X",extended_payload_len); - offset += 2; - } else if (payload_len == 127) { - uint64_t extended_payload_len; - bytes_received += lwip_recv(client->sockfd, frame+offset, 8, 0); - memcpy(&extended_payload_len, frame + 2, 8); - payload_len = ntohll(extended_payload_len); - offset += 8; - } - if(payload_len>len) - { - printf("receive_websocket_message:len =%d ERROR",payload_len); - - // if(*code == WS_BINARY) - // { - - // } - // fopen - lwip_recv(client->sockfd, buf, len, 0); - return -1; - } - timer_ms = (uint32_t)osKernelGetTickCount(); - //printf("receive_websocket_message 3 timer_ms=%d\n",timer_ms); - // printf("receive_websocket_message :"); - // for(int i = 0;isockfd, frame+offset, 4, 0); - memcpy(masking_key, frame + offset, 4); - offset += 4; - } - bytes_received = 0; - - while(1)//读主体内容 - { - //bk_sleep(recv_len/1000+1); - recv_len = lwip_recv(client->sockfd, buf+bytes_received, payload_len-bytes_received, 0); - if(recv_len>0) - bytes_received += recv_len; - else if(recv_len<0) - { - printf("receive_websocket_message: recv_len =%d lwip_recv ERROR!!!!!!",recv_len); - return -1; - } - //printf("receive_websocket_message:bytes_received=%d recv_len =%d",bytes_received,recv_len); - if(bytes_received >= payload_len) - { - //printf("receive_websocket_message:payload_len=%d,bytes_received=%d",payload_len,bytes_received); - break; - } - - } - if(mask) - { - for (size_t i = 0; i < payload_len; ++i) { - buf[i] = buf[i] ^ masking_key[i % 4]; - } - } - - // if (*code == WS_TEXT) { - // printf("Received text message: %.*s\n", (int)payload_len, buf); - // } else if (*code == WS_BINARY) { - // printf("Received binary message of length %zu\n", payload_len); - // } - - if (*code == WS_TEXT) - { - - }else if(*code == WS_BINARY) - { - - }else if(*code == WS_CLOSE) - { - payload_len = 0; - }else if(*code == WS_PONG) - { - payload_len = 0; - }else if(*code == WS_PING) - { - printf("send_websocket_message WS_PONG \n"); - send_websocket_message(client,buf,payload_len,WS_PONG); - payload_len = 0; - }else if(*code == WS_CONT) - { - payload_len = 0; - } - last_byte = buf[payload_len-1]; - timer_ms = (uint32_t)osKernelGetTickCount(); - //printf("receive_websocket_message 4 timer_ms=%d len = %d\n",timer_ms,payload_len); - return payload_len;//parse_frame(buf, bytes_received, buf,code); +void _tcp_on_connect_cb(tcp_handle_t handle, bool success) +{ + printf("tcp_on_connect_cb %d %d\r\n",handle,success); } - - - -struct cat1_ws_init_t +void _tcp_on_disconnect_cb(tcp_handle_t handle) { - osSemaphoreId_t dns_done_sem; - ip_addr_t ip; -}; - -static void dns_found(const char* hostname, const ip_addr_t *ipaddr, void *arg) + printf("tcp_on_disconnect_cb %d\r\n",handle); +} +void _tcp_on_data_received_cb(tcp_handle_t handle, uint8_t* data, uint16_t datalen) { - struct cat1_ws_init_t * ip_sem=(struct cat1_ws_init_t *)arg; - - printf("hostname %s ipaddr%d \r\n",hostname,ipaddr->u_addr.ip4); - memcpy(&(ip_sem->ip),ipaddr,sizeof(ipaddr)); - (void)osSemaphoreRelease(ip_sem->dns_done_sem); + printf("tcp_on_data_received_cb %d %d\r\n",handle,datalen); + extern WebSocketClient * get_ws_client_by_handle(void* handle); + WebSocketClient * client = get_ws_client_by_handle(handle); + if(client==NULL) + { + printf("get_ws_client failed\r\n"); + return; + } + // 将接收到的原始数据传递给组包解析器 + int processed = ws_process_received_data(client, data, datalen); + if (processed < 0) { + printf("WebSocket packet processing failed\n"); + return; + } + + printf("Processed %d bytes of WebSocket data\n", processed); } - - // 初始化 WebSocket 客户端 WebSocketClient *init_websocket_client(const char *url, const char *headers,uint8_t use_dns) { //printf("init_websocket_client 1\n"); char *buffer = NULL; WebSocketClient *client =NULL; BK_URL parsed_url = {0}; - struct sockaddr_in server_addr; ssize_t bytes_received; - struct in_addr in_addr_t={0}; - ip_addr_t ip; - struct cat1_ws_init_t ip_sem={0}; int opt = 1; err_t ret=0; @@ -759,81 +622,30 @@ WebSocketClient *init_websocket_client(const char *url, const char *headers,uint printf("init_websocket_client alloc error"); goto client_ERROR; } + //移植建议： 此处调用 tcp_open_url 进行连接, 不必使用 lwip_connect，异常及数据接收，都放到callback中处理 + client->tcp_handle = tcp_open_url(parsed_url.host, parsed_url.port, _tcp_on_connect_cb, _tcp_on_disconnect_cb, _tcp_on_data_received_cb); - - - client->sockfd = lwip_socket(AF_INET, SOCK_STREAM, 0); - if (client->sockfd < 0) { - - printf("socket creation failed %d errno: %d\r\n",client->sockfd,errno); - goto client_ERROR; - } - lwip_setsockopt(client->sockfd, SOL_SOCKET, SO_KEEPALIVE, (const void*)&opt, sizeof(opt)); - lwip_setsockopt(client->sockfd, SOL_SOCKET, SO_REUSEADDR,(const void*)&opt, sizeof(opt)); - // printf("init_websocket_client 3\n"); - if(use_dns) - { - ip_sem.dns_done_sem= osSemaphoreNew(1, 0, NULL); - if(ip_sem.dns_done_sem==0) - { - goto client_ERROR; - } - - ret =dns_gethostbyname(parsed_url.host,&ip,dns_found,&ip_sem); - // if(ERR_OK==ret) - { - osSemaphoreAcquire(ip_sem.dns_done_sem, 1000*30); - } - osSemaphoreDelete(ip_sem.dns_done_sem); - printf("dns_gethostbyname %d\r\n",ip_sem.ip.u_addr.ip4); - memcpy(&server_addr.sin_addr.s_addr, &ip_sem.ip.u_addr.ip4, 4); - - } - else + if(client->tcp_handle == NULL) { - inet_pton(AF_INET, parsed_url.host, &server_addr.sin_addr); - } - - - - server_addr.sin_family = AF_INET; - - server_addr.sin_port = htons(parsed_url.port); - - - // - - if (connect(client->sockfd, (struct sockaddr *)&server_addr, sizeof(server_addr)) < 0) { - printf("init_websocket_client connection failed\r\n"); - lwip_close(client->sockfd); + printf("socket creation failed\r\n"); goto client_ERROR; } - printf("socket creation succ\r\n"); - - + printf("socket creation success\r\n"); strcpy(client->host, parsed_url.host); strcpy(client->path, parsed_url.path); strcpy(client->query, parsed_url.query); client->port = parsed_url.port; - if (send_handshake(client,headers) < 0) { //发送头 + + /* 初始化WebSocket组包解析器 */ + ws_parser_init(&client->parser); + buffer=(char *)aic_zalloc(BUFFER_SIZE); + if (bytes_received = send_handshake(client,headers,buffer,BUFFER_SIZE) < 0) { //发送头 printf("socket send_handshake client_ERROR"); - lwip_close(client->sockfd); + tcp_manager_close(client->tcp_handle); goto client_ERROR; } - buffer=(char *)aic_zalloc(BUFFER_SIZE); - if(buffer==NULL) - { - printf("aic_zalloc(BUFFER_SIZE);null "); - goto client_ERROR; - } - printf("lwip_recv: ??? "); - - bytes_received = lwip_recv(client->sockfd, buffer, BUFFER_SIZE - 1, 0); - - printf("lwip_recv; "); - if (bytes_received > 0) { buffer[bytes_received] = '\0'; printf("Received handshake response:\n%s\n", buffer); @@ -870,16 +682,309 @@ client_ERROR: void lwip_close_websocket_client(WebSocketClient *client) { if(client) { + //移植建议： 此处调用 tcp_manager_close 进行关闭, 不必使用 lwip_close + printf("lwip_close_websocket_client %s %s",client->host,client->path); - lwip_shutdown(client->sockfd,SHUT_RDWR); - bk_sleep(10); - int err= lwip_close(client->sockfd); - WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"lwip_close %d %d\n",client->sockfd,err); + tcp_manager_close(client->tcp_handle); free(client); } } +/** + * @brief 初始化WebSocket组包解析器 + */ +static void ws_parser_init(ws_packet_parser_t* parser) +{ + if (!parser) return; + + memset(parser, 0, sizeof(ws_packet_parser_t)); + parser->parse_state = WS_PARSE_HEADER; +} + +/** + * @brief 重置WebSocket组包解析器状态 + */ +static void ws_parser_reset(ws_packet_parser_t* parser) +{ + if (!parser) return; + + parser->frame_buffer_len = 0; + parser->payload_buffer_len = 0; + parser->parse_state = WS_PARSE_HEADER; + parser->fin = 0; + parser->opcode = WS_CONT; + parser->mask = 0; + parser->payload_len = 0; + parser->payload_received = 0; + memset(parser->masking_key, 0, sizeof(parser->masking_key)); +} + +/** + * @brief 处理完整的WebSocket数据包 + */ +static void ws_handle_complete_packet(ws_packet_parser_t* parser) +{ + if (!parser || !parser->packet_callback) return; + + // 如果有掩码，进行解掩码处理 + if (parser->mask) { + for (uint32_t i = 0; i < parser->payload_buffer_len; ++i) { + parser->payload_buffer[i] ^= parser->masking_key[i % 4]; + } + } + + // 处理不同类型的消息 + switch (parser->opcode) { + case WS_PING: + printf("Received WebSocket PING\n"); + // TODO: 自动回复PONG + break; + + case WS_PONG: + printf("Received WebSocket PONG\n"); + break; + + case WS_CLOSE: + printf("Received WebSocket CLOSE\n"); + parser->packet_callback(parser->payload_buffer, parser->payload_buffer_len, parser->opcode); + break; + + case WS_TEXT: + case WS_BINARY: + case WS_CONT: + // 调用回调函数处理数据包 + parser->packet_callback(parser->payload_buffer, parser->payload_buffer_len, parser->opcode); + break; + + default: + printf("Unknown WebSocket opcode: 0x%02X\n", parser->opcode); + break; + } +} + +/** + * @brief 解析WebSocket帧头 + * @return 解析的字节数，-1表示需要更多数据，-2表示错误 + */ +static int ws_parse_frame_header(ws_packet_parser_t* parser) +{ + if (parser->frame_buffer_len < 2) { + return -1; // 需要更多数据 + } + + uint8_t* frame = parser->frame_buffer; + + // 解析第一个字节 + parser->fin = (frame[0] & 0x80) >> 7; + parser->opcode = (WebSocketMessageType)(frame[0] & 0x0F); + + // 解析第二个字节 + parser->mask = (frame[1] & 0x80) >> 7; + uint8_t payload_len_initial = frame[1] & 0x7F; + + // 根据载荷长度字段确定实际长度 + if (payload_len_initial < 126) { + parser->payload_len = payload_len_initial; + parser->parse_state = parser->mask ? WS_PARSE_MASK : WS_PARSE_PAYLOAD; + return 2; + } else if (payload_len_initial == 126) { + parser->parse_state = WS_PARSE_EXTENDED; + return 2; + } else if (payload_len_initial == 127) { + parser->parse_state = WS_PARSE_EXTENDED; + return 2; + } + + return -2; // 错误 +} + +/** + * @brief 解析扩展载荷长度 + * @return 解析的字节数，-1表示需要更多数据，-2表示错误 + */ +static int ws_parse_extended_length(ws_packet_parser_t* parser) +{ + uint8_t* frame = parser->frame_buffer; + uint8_t payload_len_initial = frame[1] & 0x7F; + + if (payload_len_initial == 126) { + if (parser->frame_buffer_len < 4) { + return -1; // 需要更多数据 + } + + uint16_t extended_len; + memcpy(&extended_len, frame + 2, 2); + parser->payload_len = ntohs(extended_len); + parser->parse_state = parser->mask ? WS_PARSE_MASK : WS_PARSE_PAYLOAD; + return 4; + + } else if (payload_len_initial == 127) { + if (parser->frame_buffer_len < 10) { + return -1; // 需要更多数据 + } + + uint64_t extended_len; + memcpy(&extended_len, frame + 2, 8); + parser->payload_len = ntohll(extended_len); + + // 检查载荷长度是否过大 + if (parser->payload_len > WS_PAYLOAD_BUFFER_SIZE) { + printf("WebSocket payload too large: %llu bytes\n", parser->payload_len); + return -2; // 错误 + } + + parser->parse_state = parser->mask ? WS_PARSE_MASK : WS_PARSE_PAYLOAD; + return 10; + } + + return -2; // 错误 +} + +/** + * @brief 解析掩码密钥 + * @return 解析的字节数，-1表示需要更多数据 + */ +static int ws_parse_mask_key(ws_packet_parser_t* parser, int header_len) +{ + if (parser->frame_buffer_len < header_len + 4) { + return -1; // 需要更多数据 + } + + memcpy(parser->masking_key, parser->frame_buffer + header_len, 4); + parser->parse_state = WS_PARSE_PAYLOAD; + return header_len + 4; +} + +/** + * @brief 设置WebSocket数据包回调函数 + */ +void ws_set_packet_callback(WebSocketClient *client, on_ws_packet_callback callback) +{ + if (!client) return; + + client->parser.packet_callback = callback; +} + +/** + * @brief 处理接收到的原始数据 + */ +int ws_process_received_data(WebSocketClient *client, const uint8_t *data, size_t len) +{ + if (!client || !data || len == 0) { + return -1; + } + + ws_packet_parser_t* parser = &client->parser; + const uint8_t* input = data; + size_t remaining = len; + + while (remaining > 0) { + switch (parser->parse_state) { + case WS_PARSE_HEADER: + case WS_PARSE_EXTENDED: + { + // 将数据复制到帧缓冲区 + size_t copy_len = (remaining > WS_FRAME_BUFFER_SIZE - parser->frame_buffer_len) + ? (WS_FRAME_BUFFER_SIZE - parser->frame_buffer_len) : remaining; + + if (copy_len == 0) { + printf("Frame buffer overflow\n"); + ws_parser_reset(parser); + return -1; + } + + memcpy(parser->frame_buffer + parser->frame_buffer_len, input, copy_len); + parser->frame_buffer_len += copy_len; + input += copy_len; + remaining -= copy_len; + + // 尝试解析帧头 + int parsed = 0; + if (parser->parse_state == WS_PARSE_HEADER) { + parsed = ws_parse_frame_header(parser); + } else { + parsed = ws_parse_extended_length(parser); + } + + if (parsed == -1) { + break; // 需要更多数据 + } else if (parsed == -2) { + printf("WebSocket frame parsing error\n"); + ws_parser_reset(parser); + return -1; + } + break; + } + + case WS_PARSE_MASK: + { + // 将数据复制到帧缓冲区 + size_t copy_len = (remaining > WS_FRAME_BUFFER_SIZE - parser->frame_buffer_len) + ? (WS_FRAME_BUFFER_SIZE - parser->frame_buffer_len) : remaining; + + if (copy_len == 0) { + printf("Frame buffer overflow\n"); + ws_parser_reset(parser); + return -1; + } + + memcpy(parser->frame_buffer + parser->frame_buffer_len, input, copy_len); + parser->frame_buffer_len += copy_len; + input += copy_len; + remaining -= copy_len; + + // 计算头部长度 + uint8_t payload_len_initial = parser->frame_buffer[1] & 0x7F; + int header_len = 2; + if (payload_len_initial == 126) { + header_len = 4; + } else if (payload_len_initial == 127) { + header_len = 10; + } + + // 尝试解析掩码密钥 + int parsed = ws_parse_mask_key(parser, header_len); + if (parsed == -1) { + break; // 需要更多数据 + } + break; + } + + case WS_PARSE_PAYLOAD: + { + // 计算需要接收的载荷数据长度 + uint32_t need_len = parser->payload_len - parser->payload_received; + size_t copy_len = (remaining > need_len) ? need_len : remaining; + + // 检查载荷缓冲区空间 + if (parser->payload_buffer_len + copy_len > WS_PAYLOAD_BUFFER_SIZE) { + printf("Payload buffer overflow\n"); + ws_parser_reset(parser); + return -1; + } + + // 复制载荷数据 + memcpy(parser->payload_buffer + parser->payload_buffer_len, input, copy_len); + parser->payload_buffer_len += copy_len; + parser->payload_received += copy_len; + input += copy_len; + remaining -= copy_len; + + // 检查是否接收完整个载荷 + if (parser->payload_received >= parser->payload_len) { + // 处理完整的数据包 + ws_handle_complete_packet(parser); + // 重置解析器状态 + ws_parser_reset(parser); + } + break; + } + } + } + + return len - remaining; // 返回实际处理的字节数 +} #ifdef __cplusplus diff --git a/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.c b/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.c index 8f44e0dd5..2960a3bc9 100644 --- a/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.c +++ b/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.c @@ -22,6 +22,9 @@ #include #include #include + +/* 运行时调试等级控制 */ +static int g_tcp_debug_level = TCP_DEBUG_LEVEL; /* 使用lwIP网络头文件替代标准网络头文件 */ #include "lwip/sockets.h" /* 包含lwIP socket API和fd_set等定义 */ #include "lwip/netdb.h" @@ -106,18 +109,11 @@ #endif #endif -/* 优化后的常量定义 */ -#define TCP_RECV_BUFFER_SIZE 1024 -#define TCP_SEND_QUEUE_SIZE 16 /* 每个连接的发送队列大小 */ -#define TCP_SEND_BUFFER_SIZE 512 /* 发送缓冲区大小 */ -#define TCP_CONNECT_TIMEOUT_MS 10000 /* 增加连接超时时间 */ -#define TCP_SELECT_TIMEOUT_MS 50 /* 减少select超时，提高响应性 */ -#define TCP_THREAD_STACK_SIZE 8192 -#define TCP_THREAD_PRIORITY 10 -#define TCP_DNS_TIMEOUT_MS 8000 /* 增加DNS超时时间 */ +/* TCP Keep-Alive 配置 */ #define TCP_KEEPALIVE_IDLE 30 #define TCP_KEEPALIVE_INTERVAL 5 #define TCP_KEEPALIVE_COUNT 3 +#define TCP_SELECT_TIMEOUT_MS 50 /* 减少select超时，提高响应性 */ #define TCP_CONNECTION_IDLE_TIMEOUT_MS 60000 #define TCP_CONNECT_RETRY_COUNT 3 /* 连接重试次数 */ #define TCP_CONNECT_RETRY_DELAY_MS 1000 /* 重试延迟 */ @@ -150,6 +146,7 @@ static uint32_t g_dns_cache_access_counter = 0; /* IO线程相关 */ static bool g_io_thread_running = false; static sys_thread_t g_io_thread; +static uint32_t g_io_thread_id = 0; /* IO线程ID */ /* 前向声明 */ static void tcp_manager_io_thread(void* arg); @@ -720,7 +717,9 @@ static bool process_send_queue(tcp_connection_info_t* conn) static void tcp_manager_io_thread(void* arg) { (void)arg; - printf("TCP管理器IO线程启动\n"); + /* 记录IO线程ID */ + g_io_thread_id = (uint32_t)osThreadGetId(); + printf("TCP管理器IO线程启动，线程ID: %u\n", g_io_thread_id); fd_set read_fds, write_fds, except_fds; struct timeval timeout; @@ -800,7 +799,7 @@ static void tcp_manager_io_thread(void* arg) if (FD_ISSET(conn->sockfd, &read_fds)) { /* 处理数据接收 */ - uint8_t recv_buffer[TCP_RECV_BUFFER_SIZE]; + uint8_t recv_buffer[TCP_MANAGER_RECV_BUFFER_SIZE]; ssize_t recv_len = recv(conn->sockfd, recv_buffer, sizeof(recv_buffer), MSG_DONTWAIT); if (recv_len > 0) { @@ -1070,7 +1069,7 @@ tcp_manager_result_t tcp_simple_manager_init(void) /* 创建IO线程 */ g_io_thread_running = true; g_io_thread = sys_thread_new("tcp_io", tcp_manager_io_thread, NULL, - TCP_THREAD_STACK_SIZE, TCP_THREAD_PRIORITY); + TCP_MANAGER_THREAD_STACK_SIZE, TCP_MANAGER_THREAD_PRIORITY); if (g_io_thread == (sys_thread_t)-1) { printf("TCP管理器初始化失败: 无法创建IO线程\n"); @@ -1079,7 +1078,7 @@ tcp_manager_result_t tcp_simple_manager_init(void) } g_manager_initialized = true; - printf("TCP管理器初始化成功\n"); + TCPM_INFO("TCP管理器初始化成功"); return TCP_MANAGER_OK; } @@ -1175,17 +1174,18 @@ tcp_handle_t tcp_open_url(const char* host, uint16_t port, tcp_on_data_received_cb data_cb) { if (!g_manager_initialized || !host) { + TCPM_ERR("TCP管理器未初始化或主机地址为空"); return NULL; } - printf("尝试连接到: %s:%d\n", host, port); + TCPM_INFO("尝试连接到: %s:%d", host, port); sys_mutex_lock(&g_connections_mutex); tcp_connection_info_t* conn = find_free_connection_unsafe(); if (!conn) { sys_mutex_unlock(&g_connections_mutex); - printf("连接失败: 没有可用的连接槽位\n"); + TCPM_ERR("连接失败: 没有可用的连接槽位"); return NULL; } @@ -1215,7 +1215,7 @@ tcp_handle_t tcp_open_url(const char* host, uint16_t port, /* DNS解析（带缓存） */ struct in_addr addr; if (dns_resolve_with_cache(host, &addr) < 0) { - printf("DNS解析失败: %s\n", host); + TCPM_ERR("DNS解析失败: %s", host); sys_mutex_lock(&g_connections_mutex); cleanup_connection_unsafe(conn); sys_mutex_unlock(&g_connections_mutex); @@ -1228,7 +1228,7 @@ tcp_handle_t tcp_open_url(const char* host, uint16_t port, /* 创建socket */ int sockfd = socket(AF_INET, SOCK_STREAM, 0); if (sockfd < 0) { - printf("创建socket失败: %d\n", errno); + TCPM_ERR("创建socket失败: %d", errno); sys_mutex_lock(&g_connections_mutex); cleanup_connection_unsafe(conn); sys_mutex_unlock(&g_connections_mutex); @@ -1240,7 +1240,7 @@ tcp_handle_t tcp_open_url(const char* host, uint16_t port, /* 设置非阻塞 */ if (set_socket_nonblocking(sockfd) < 0) { - printf("设置非阻塞模式失败: %d\n", errno); + TCPM_ERR("设置非阻塞模式失败: %d", errno); close(sockfd); sys_mutex_lock(&g_connections_mutex); cleanup_connection_unsafe(conn); @@ -1289,14 +1289,14 @@ tcp_handle_t tcp_open_url(const char* host, uint16_t port, g_stats.total_connections++; g_stats.active_connections++; - printf("连接进行中: %s:%d, handle=%p, sockfd=%d\n", host, port, conn->handle, sockfd); + TCPM_INFO("连接进行中: %s:%d, handle=%p, sockfd=%d", host, port, conn->handle, sockfd); sys_mutex_unlock(&g_connections_mutex); return conn->handle; } else { /* 连接失败 */ - printf("连接失败: %s:%d, 错误: %d\n", host, port, errno); + TCPM_ERR("连接失败: %s:%d, 错误: %d", host, port, errno); /* 清理资源 */ close(sockfd); @@ -1304,6 +1304,11 @@ tcp_handle_t tcp_open_url(const char* host, uint16_t port, conn->sockfd = -1; sys_mutex_unlock(&g_connections_mutex); + + /* 调用连接失败回调 */ + if (connect_cb) { + connect_cb(conn->handle, false); + } return NULL; } } @@ -1314,13 +1319,17 @@ tcp_handle_t tcp_open_url(const char* host, uint16_t port, tcp_manager_result_t tcp_send_data(tcp_handle_t handle, uint8_t* data, uint16_t datalen) { if (!g_manager_initialized || !handle || !data || datalen == 0) { + TCPM_ERR("发送数据参数无效"); return TCP_MANAGER_INVALID_HANDLE; } + TCPM_DBG("发送数据: handle=%p, len=%d", handle, datalen); + sys_mutex_lock(&g_connections_mutex); tcp_connection_info_t* conn = find_connection_by_handle_unsafe(handle); if (!conn || conn->state != TCP_STATE_CONNECTED) { + TCPM_WARN("连接无效或未连接: handle=%p, state=%d", handle, conn ? conn->state : -1); sys_mutex_unlock(&g_connections_mutex); return TCP_MANAGER_CONNECTION_FAILED; } @@ -1422,4 +1431,188 @@ tcp_manager_result_t tcp_set_timeout(tcp_handle_t handle, uint32_t timeout_ms) (void)handle; (void)timeout_ms; return TCP_MANAGER_ERROR; /* 未实现 */ +} + +/** + * @brief 检查当前线程是否为IO线程 + */ +static bool is_in_io_thread(void) +{ + return (g_io_thread_id != 0) && ((uint32_t)osThreadGetId() == g_io_thread_id); +} + +/** + * @brief 直接从连接接收数据 (仅在IO线程中调用) + */ +int tcp_recv_data_directly(tcp_handle_t handle, uint8_t* buffer, uint16_t buffer_size) +{ + if (!handle || !buffer || buffer_size == 0) { + printf("tcp_recv_data_directly: 无效参数\n"); + return -1; + } + + /* 检查是否在IO线程中 */ + if (!is_in_io_thread()) { + printf("tcp_recv_data_directly: 错误！只能在IO线程中调用\n"); + return -1; + } + + tcp_connection_info_t* conn = (tcp_connection_info_t*)handle; + + /* 检查连接状态 */ + if (!conn->in_use || conn->state != TCP_STATE_CONNECTED || conn->sockfd < 0) { + printf("tcp_recv_data_directly: 连接未建立或已关闭\n"); + return -1; + } + + /* 直接从socket接收数据 */ + int ret = recv(conn->sockfd, buffer, buffer_size, MSG_DONTWAIT); + if (ret < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK) { + /* 非阻塞模式下暂无数据可读 */ + return 0; + } else { + printf("tcp_recv_data_directly: recv失败, errno=%d\n", errno); + return -1; + } + } else if (ret == 0) { + /* 连接被对端关闭 */ + printf("tcp_recv_data_directly: 连接被对端关闭\n"); + return -1; + } + + printf("tcp_recv_data_directly: 接收到 %d 字节数据\n", ret); + return ret; +} + +/** + * @brief 同步发送数据并等待响应 (仅在IO线程中调用) + */ +int tcp_send_and_wait_response(tcp_handle_t handle, const uint8_t* send_data, uint16_t send_len, + uint8_t* recv_buffer, uint16_t recv_buffer_size, uint32_t timeout_ms) +{ + if (!handle || !send_data || send_len == 0 || !recv_buffer || recv_buffer_size == 0) { + printf("tcp_send_and_wait_response: 无效参数\n"); + return -1; + } + + /* 检查是否在IO线程中 */ + if (!is_in_io_thread()) { + printf("tcp_send_and_wait_response: 错误！只能在IO线程中调用\n"); + return -1; + } + + tcp_connection_info_t* conn = (tcp_connection_info_t*)handle; + + /* 检查连接状态 */ + if (!conn->in_use || conn->state != TCP_STATE_CONNECTED || conn->sockfd < 0) { + printf("tcp_send_and_wait_response: 连接未建立或已关闭\n"); + return -1; + } + + printf("tcp_send_and_wait_response: 发送 %d 字节数据，等待响应...\n", send_len); + + /* 1. 发送数据 */ + int sent_total = 0; + uint32_t start_time = sys_now(); + + while (sent_total < send_len) { + int sent = send(conn->sockfd, send_data + sent_total, send_len - sent_total, MSG_DONTWAIT); + if (sent < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK) { + /* 检查超时 */ + if ((sys_now() - start_time) > timeout_ms) { + printf("tcp_send_and_wait_response: 发送超时\n"); + return -1; + } + /* 短暂等待后重试 */ + sys_msleep(1); + continue; + } else { + printf("tcp_send_and_wait_response: send失败, errno=%d\n", errno); + return -1; + } + } else if (sent == 0) { + printf("tcp_send_and_wait_response: 连接被关闭\n"); + return -1; + } + + sent_total += sent; + } + + printf("tcp_send_and_wait_response: 成功发送 %d 字节\n", sent_total); + + /* 2. 等待响应数据 */ + start_time = sys_now(); + fd_set read_fds; + struct timeval select_timeout; + + while ((sys_now() - start_time) < timeout_ms) { + FD_ZERO(&read_fds); + FD_SET(conn->sockfd, &read_fds); + + /* 计算剩余超时时间 */ + uint32_t elapsed = sys_now() - start_time; + uint32_t remaining_ms = (elapsed < timeout_ms) ? (timeout_ms - elapsed) : 0; + + if (remaining_ms == 0) { + break; + } + + select_timeout.tv_sec = remaining_ms / 1000; + select_timeout.tv_usec = (remaining_ms % 1000) * 1000; + + int select_ret = select(conn->sockfd + 1, &read_fds, NULL, NULL, &select_timeout); + if (select_ret < 0) { + printf("tcp_send_and_wait_response: select失败, errno=%d\n", errno); + return -1; + } else if (select_ret == 0) { + /* 超时，继续等待 */ + continue; + } + + /* 有数据可读 */ + if (FD_ISSET(conn->sockfd, &read_fds)) { + int recv_ret = recv(conn->sockfd, recv_buffer, recv_buffer_size, MSG_DONTWAIT); + if (recv_ret < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK) { + /* 暂无数据，继续等待 */ + continue; + } else { + printf("tcp_send_and_wait_response: recv失败, errno=%d\n", errno); + return -1; + } + } else if (recv_ret == 0) { + printf("tcp_send_and_wait_response: 连接被对端关闭\n"); + return -1; + } else { + printf("tcp_send_and_wait_response: 接收到 %d 字节响应\n", recv_ret); + return recv_ret; + } + } + } + + printf("tcp_send_and_wait_response: 等待响应超时\n"); + return -1; +} + +/** + * @brief 设置调试等级 (运行时修改) + */ +void tcp_set_debug_level(int level) +{ + if (level >= TCP_DEBUG_LEVEL_NONE && level <= TCP_DEBUG_LEVEL_DEBUG) { + g_tcp_debug_level = level; + printf("TCP调试等级设置为: %d\n", level); + } else { + printf("无效的调试等级: %d (有效范围: 0-4)\n", level); + } +} + +/** + * @brief 获取当前调试等级 + */ +int tcp_get_debug_level(void) +{ + return g_tcp_debug_level; } \ No newline at end of file diff --git a/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.h b/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.h index 9c239694c..b1f35b430 100644 --- a/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.h +++ b/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.h @@ -22,22 +22,39 @@ extern "C" { #endif /* 配置参数 */ -#define TCP_MANAGER_MAX_CONNECTIONS 8 /* 最大连接数 */ +#define TCP_MANAGER_MAX_CONNECTIONS 4 /* 最大连接数 */ #define TCP_MANAGER_RECV_QUEUE_SIZE 16 /* 接收队列大小 */ #define TCP_MANAGER_CMD_QUEUE_SIZE 32 /* 命令队列大小 */ -#define TCP_MANAGER_RECV_BUFFER_SIZE 1024 /* 接收缓冲区大小 */ +#define TCP_MANAGER_RECV_BUFFER_SIZE 128 /* 接收缓冲区大小 */ #define TCP_MANAGER_THREAD_STACK_SIZE 4096 /* 线程栈大小 */ #define TCP_MANAGER_THREAD_PRIORITY 5 /* 线程优先级 */ -/* 常量定义 */ -#define TCP_MANAGER_MAX_CONNECTIONS 8 -#define TCP_MANAGER_RECV_BUFFER_SIZE 1024 -#define TCP_MANAGER_CMD_QUEUE_SIZE 32 -#define TCP_THREAD_STACK_SIZE 8192 +/* 调试等级定义 */ +#define TCP_DEBUG_LEVEL_NONE 0 /* 无调试信息 */ +#define TCP_DEBUG_LEVEL_ERROR 1 /* 仅错误信息 */ +#define TCP_DEBUG_LEVEL_WARN 2 /* 警告+错误信息 */ +#define TCP_DEBUG_LEVEL_INFO 3 /* 信息+警告+错误 */ +#define TCP_DEBUG_LEVEL_DEBUG 4 /* 全部调试信息 */ + +/* 当前调试等级 (可通过宏定义覆盖) */ +#ifndef TCP_DEBUG_LEVEL +#define TCP_DEBUG_LEVEL TCP_DEBUG_LEVEL_WARN +#endif + +/* 运行时调试等级控制 (外部声明) */ +extern int tcp_get_debug_level(void); + +/* 轻量级调试打印宏 (支持运行时控制) */ +#define TCPM_ERR(fmt, ...) do { if(tcp_get_debug_level() >= TCP_DEBUG_LEVEL_ERROR) printf("[TCPM_ERR] " fmt "\n", ##__VA_ARGS__); } while(0) +#define TCPM_WARN(fmt, ...) do { if(tcp_get_debug_level() >= TCP_DEBUG_LEVEL_WARN) printf("[TCPM_WARN] " fmt "\n", ##__VA_ARGS__); } while(0) +#define TCPM_INFO(fmt, ...) do { if(tcp_get_debug_level() >= TCP_DEBUG_LEVEL_INFO) printf("[TCPM_INFO] " fmt "\n", ##__VA_ARGS__); } while(0) +#define TCPM_DBG(fmt, ...) do { if(tcp_get_debug_level() >= TCP_DEBUG_LEVEL_DEBUG) printf("[TCPM_DBG] " fmt "\n", ##__VA_ARGS__); } while(0) + +/* 连接参数定义 */ #define TCP_CONNECT_TIMEOUT_MS 10000 #define TCP_DNS_TIMEOUT_MS 8000 -#define TCP_SEND_QUEUE_SIZE 16 -#define TCP_SEND_BUFFER_SIZE 512 +#define TCP_SEND_QUEUE_SIZE 4 +#define TCP_SEND_BUFFER_SIZE 128 #define TCP_CONNECT_RETRY_COUNT 3 #define TCP_CONNECT_RETRY_DELAY_MS 1000 @@ -155,7 +172,7 @@ typedef struct { } tcp_manager_stats_t; /* DNS缓存配置 */ -#define DNS_CACHE_MAX_ENTRIES 16 /* 最大缓存条目数 */ +#define DNS_CACHE_MAX_ENTRIES 8 /* 最大缓存条目数 */ #define DNS_CACHE_TTL_MS 300000 /* 缓存有效期: 5分钟 */ #define DNS_HOSTNAME_MAX_LEN 64 /* 主机名最大长度 */ @@ -269,6 +286,45 @@ tcp_manager_result_t tcp_manager_close_directly(tcp_handle_t handle); */ tcp_manager_result_t tcp_set_timeout(tcp_handle_t handle, uint32_t timeout_ms); +/** + * @brief 同步发送数据并等待响应 (仅在IO线程中调用) + * 注意：此函数会锁定tcp_handle，发送数据后直接等待结果返回 + * 只能在tcp_manager_io_thread中调用，确保线程安全 + * + * @param handle 连接句柄 + * @param send_data 要发送的数据 + * @param send_len 发送数据长度 + * @param recv_buffer 接收缓冲区 + * @param recv_buffer_size 接收缓冲区大小 + * @param timeout_ms 等待超时时间(毫秒) + * @return int 实际接收到的数据长度，<0表示错误 + */ +int tcp_send_and_wait_response(tcp_handle_t handle, const uint8_t* send_data, uint16_t send_len, + uint8_t* recv_buffer, uint16_t recv_buffer_size, uint32_t timeout_ms); + +/** + * @brief 直接从连接接收数据 (仅在IO线程中调用) + * 注意：此函数会直接从socket读取数据，不经过回调机制 + * 只能在tcp_manager_io_thread中调用，确保线程安全 + * + * @param handle 连接句柄 + * @param buffer 接收缓冲区 + * @param buffer_size 缓冲区大小 + * @return int 实际接收到的数据长度，<0表示错误 + */ +int tcp_recv_data_directly(tcp_handle_t handle, uint8_t* buffer, uint16_t buffer_size); + +/** + * @brief 设置调试等级 (运行时修改) + * @param level 调试等级 (0-4) + */ +void tcp_set_debug_level(int level); + +/** + * @brief 获取当前调试等级 + */ +int tcp_get_debug_level(void); + #ifdef __cplusplus } #endif diff --git a/code/sdk/drivers/peripheral/slip/websocket_test.c b/code/sdk/drivers/peripheral/slip/websocket_test.c index d73e7cea9..5cff6761a 100644 --- a/code/sdk/drivers/peripheral/slip/websocket_test.c +++ b/code/sdk/drivers/peripheral/slip/websocket_test.c @@ -74,7 +74,7 @@ static void websocket_send_thread(void *argument) while (test_running) { // 检查客户端是否有效 - if (!test_client || test_client->sockfd < 0) { + if (!test_client || test_client->tcp_handle == NULL) { WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "发送线程: 客户端无效"); break; } @@ -121,7 +121,7 @@ static void websocket_recv_thread(void *argument) while (test_running) { // 检查客户端是否有效 - if (!test_client || test_client->sockfd < 0) { + if (!test_client || test_client->tcp_handle == NULL) { WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "接收线程: 客户端无效"); break; } @@ -304,7 +304,7 @@ static bool send_test_data(uint8_t data_value) } // 添加额外的安全检查 - if (test_client->sockfd < 0) { + if (test_client->tcp_handle == NULL) { WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "发送数据失败: 无效的socket"); return false; } @@ -352,8 +352,8 @@ static bool receive_test_data(uint8_t *data_value) } WebSocketMessageType type; - int result = receive_websocket_message(test_client, received_data, TEST_DATA_SIZE, &type); - + // int result = receive_websocket_message(test_client, received_data, TEST_DATA_SIZE, &type); + int result = 0; if (result <= 0) { WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "接收数据失败: result=%d", result); return false; -- 2.17.1