From ae473e528a651a5d43de917cdf252e5fb8b284bb Mon Sep 17 00:00:00 2001 From: hcj <1224667075@qq.com> Date: Sat, 16 Aug 2025 11:28:23 +0800 Subject: [PATCH] =?utf8?q?=E4=BC=98=E5=8C=96=E7=BC=93=E5=AD=98=E6=B8=85?= =?utf8?q?=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit --- .../slip/cat1Service/bk_websocket.h | 1 + .../slip/cat1Service/cat1_ai_server.cpp | 197 +++++++++---- .../peripheral/slip/cat1Service/cat1_ws.cpp | 276 +++++++++++++++++- .../slip/tcpmanager/tcp_simple_manager.c | 134 ++++++--- .../slip/tcpmanager/tcp_simple_manager.h | 17 ++ 5 files changed, 517 insertions(+), 108 deletions(-) diff --git a/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h b/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h index bbfa75d39..fc0c3ac04 100644 --- a/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h +++ b/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h @@ -54,6 +54,7 @@ typedef struct { /* WebSocket解析器 */ ws_parser_t* parser; + int sockfd; /* socket文件描述符 */ /* 数据包回调函数 */ on_ws_packet_callback packet_callback; 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 a7d322422..360391565 100644 --- a/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ai_server.cpp +++ b/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ai_server.cpp @@ -575,6 +575,59 @@ static void ws_queues_deinit_all(void) WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "All WebSocket queues deinitialized"); } +static int ws_queue_cleanup(ws_message_queue_t* queue) { + if (!queue) { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "ws_queue_cleanup: queue is NULL"); + return -1; + } + + WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "开始清理WebSocket消息队列..."); + + // 获取互斥锁 + osStatus_t status = osMutexAcquire(queue->mutex, 1000); + if (status != osOK) { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "ws_queue_cleanup: Failed to acquire mutex, status=%d", status); + return -1; + } + + // 清理队列中的所有消息 + for (int i = 0; i < WS_QUEUE_SIZE; i++) { + ws_message_item_t* item = &queue->messages[i]; + + if (item->in_use) { + WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "清理队列项 %d: type=%d, length=%d", + i, item->type, item->length); + + // 释放动态分配的数据缓冲区 + if (item->data != NULL) { + free(item->data); + item->data = NULL; + WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "释放队列项 %d 的数据缓冲区", i); + } + + // 重置队列项状态 + item->length = 0; + item->type = (WebSocketMessageType)0; + item->timestamp = 0; + item->in_use = false; + osSemaphoreAcquire(queue->semaphore, 0); + + } + } + + // 重置队列状态 + queue->head = 0; + queue->tail = 0; + queue->count = 0; + + // 释放互斥锁 + osMutexRelease(queue->mutex); + + WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket消息队列清理完成"); + return 0; +} + + void uapi_set_tts_break(uint8_t data) { tts_break=data; @@ -599,9 +652,11 @@ static uint8_t asr_is_stop=0; void uapi_UI_start_asr(uint8_t type) { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"uapi_UI_start_asr\n"); + memset_s(result_str_asr,AI_RESULT_STR_LEN,0,AI_RESULT_STR_LEN); // uapi_stop_tts(); - + #ifdef ASR_HODE asr_is_stop=0; #endif @@ -609,16 +664,13 @@ void uapi_UI_start_asr(uint8_t type) type_ai=type; if(chat_client) { - // chat_server_deinit(); - // chat_reconnect=1; + chat_reconnect=1; } - ///////////////////////////////////////////////////// - tts_break=1; // tts_server_deinit(); - // tts_reconnect=1; + tts_reconnect=1; AIStreamFlush(); send_msg_data_into_Ai_msg_queue((uint8_t *)"tts_recv_buf",13,201); @@ -703,6 +755,7 @@ bool send_ai_txt(uint8_t type,char *msg) if (send_websocket_message(chat_client,( uint8_t *)jsonString,strlen(jsonString),WS_TEXT)==0) { +chat_reconnect=0; WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"send_websocket_message succ 0\n"); cJSON_free(jsonString); return true; @@ -960,6 +1013,8 @@ void bk_websocket_demo(void *data) { AI_UNLOCK(); char_reconnet: + ws_queue_cleanup(&chat_msg_queue); + { char url_buf[248] = {0}; uint8_t is_raisound_url = 0; @@ -1066,28 +1121,35 @@ void bk_websocket_demo(void *data) { chat_client = NULL; goto cleanup_and_exit; } - WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "Chat WebSocket: TCP连接确认成功，开始握手等待"); + // if(tts_client==NULL) + // { + // tts_server_init(); + // } + // WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "Chat WebSocket: TCP连接确认成功，开始握手等待"); + #if TEST_AI_WS + #else // 等待WebSocket握手完成，增加超时机制 uint32_t handshake_start_time = (uint32_t)osKernelGetTickCount(); const uint32_t handshake_timeout_ms = AI_WS_HANDSHAKE_TIMEOUT_MS; // 使用宏定义的握手超时 while (chat_client->handshake == 0) { - // 检查是否超时 - if ((uint32_t)osKernelGetTickCount() - handshake_start_time > handshake_timeout_ms) { - WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WebSocket握手超时 (%d ms)，重新连接", handshake_timeout_ms); - lwip_close_websocket_client(chat_client); - chat_client = NULL; - goto cleanup_and_exit; - } + // 检查是否超时 + if ((uint32_t)osKernelGetTickCount() - handshake_start_time > handshake_timeout_ms) { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WebSocket握手超时 (%d ms)，重新连接", handshake_timeout_ms); + lwip_close_websocket_client(chat_client); + chat_client = NULL; + goto cleanup_and_exit; + } - msleep(1000); - WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "等待WebSocket握手完成... (%d ms)", - (uint32_t)osKernelGetTickCount() - handshake_start_time); + msleep(1000); + WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "等待WebSocket握手完成... (%d ms)", + (uint32_t)osKernelGetTickCount() - handshake_start_time); } WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket握手完成，耗时 %d ms", - (uint32_t)osKernelGetTickCount() - handshake_start_time); + (uint32_t)osKernelGetTickCount() - handshake_start_time); + #endif // 设置socket超时 // struct timeval timeout; @@ -1175,12 +1237,16 @@ void bk_websocket_demo(void *data) { lwip_close_websocket_client(chat_client); chat_client = NULL; - if (chat_reconnect == 1) { + ws_queue_cleanup(&chat_msg_queue); + + if(disconnect_server == 0) + { chat_reconnect = 0; - retry_count = 0; + retry_count = 0; WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "Reconnecting chat client..."); goto char_reconnet; } + } cleanup_and_exit: @@ -1505,7 +1571,8 @@ ASR_SATRT: AI_LOCK(); is_bk_websocket_asr = 1; AI_UNLOCK(); - + ws_queue_cleanup(&asr_msg_queue); + uint8_t is_raisound_url = 0; { @@ -1600,29 +1667,31 @@ ASR_SATRT: asr_client = NULL; goto asr_cleanup_and_exit; } + #if TEST_AI_WS + #else WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "ASR WebSocket: TCP连接确认成功，开始握手等待"); // 等待ASR WebSocket握手完成，增加超时机制 uint32_t asr_handshake_start_time = (uint32_t)osKernelGetTickCount(); const uint32_t asr_handshake_timeout_ms = AI_WS_HANDSHAKE_TIMEOUT_MS; // 使用宏定义的握手超时 - while (asr_client->handshake == 0) { - // 检查是否超时 - if ((uint32_t)osKernelGetTickCount() - asr_handshake_start_time > asr_handshake_timeout_ms) { - WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "ASR WebSocket握手超时 (%d ms)，重新连接", asr_handshake_timeout_ms); - lwip_close_websocket_client(asr_client); - asr_client = NULL; - goto asr_cleanup_and_exit; - } + while (asr_client->handshake == 0) { + // 检查是否超时 + if ((uint32_t)osKernelGetTickCount() - asr_handshake_start_time > asr_handshake_timeout_ms) { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "ASR WebSocket握手超时 (%d ms)，重新连接", asr_handshake_timeout_ms); + lwip_close_websocket_client(asr_client); + asr_client = NULL; + goto asr_cleanup_and_exit; + } - msleep(1000); + msleep(1000); WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "等待ASR WebSocket握手完成... (%d ms)", - (uint32_t)osKernelGetTickCount() - asr_handshake_start_time); - } - - WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "ASR WebSocket握手完成，耗时 %d ms", - (uint32_t)osKernelGetTickCount() - asr_handshake_start_time); + (uint32_t)osKernelGetTickCount() - asr_handshake_start_time); + } + WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "ASR WebSocket握手完成，耗时 %d ms", + (uint32_t)osKernelGetTickCount() - asr_handshake_start_time); + #endif uapi_cat1_keep_wakeup(1); server_state = 0x01 ^ server_state; @@ -1728,6 +1797,7 @@ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "等待ASR WebSocket握手完成... (%d m lwip_close_websocket_client(asr_client); asr_client = NULL; + ws_queue_cleanup(&asr_msg_queue); asr_client_ping = 0; msleep(1000); @@ -1906,7 +1976,11 @@ void uapi_start_tts_single(void) printf("uapi_start_tts : %s\r\n",jsonString); if (send_websocket_message(tts_client,( uint8_t *)jsonString,strlen(jsonString),WS_TEXT)==0) - WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"send_websocket_message succ 0\n"); + { + tts_reconnect=0; + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP,"send_websocket_message succ 0\n"); + } + free(jsonString); } @@ -1993,7 +2067,7 @@ void bk_websocket_tts(void *data) { TTS_SATRT: uapi_cat1_keep_wakeup(1); - + ws_queue_cleanup(&tts_msg_queue); WebSocketMessageType type; bool state = true; @@ -2073,6 +2147,8 @@ TTS_SATRT: tts_client = NULL; goto tts_cleanup_and_exit; } + #if TEST_AI_WS + #else WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "TTS WebSocket: TCP连接确认成功，开始握手等待"); // 等待TTS WebSocket握手完成，增加超时机制 @@ -2080,22 +2156,23 @@ TTS_SATRT: const uint32_t tts_handshake_timeout_ms = AI_WS_HANDSHAKE_TIMEOUT_MS; // 使用宏定义的握手超时 while (tts_client->handshake == 0) { - // 检查是否超时 - if ((uint32_t)osKernelGetTickCount() - tts_handshake_start_time > tts_handshake_timeout_ms) { - WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "TTS WebSocket握手超时 (%d ms)，重新连接", tts_handshake_timeout_ms); - lwip_close_websocket_client(tts_client); - tts_client = NULL; - goto tts_cleanup_and_exit; - } + // 检查是否超时 + if ((uint32_t)osKernelGetTickCount() - tts_handshake_start_time > tts_handshake_timeout_ms) { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "TTS WebSocket握手超时 (%d ms)，重新连接", tts_handshake_timeout_ms); + lwip_close_websocket_client(tts_client); + tts_client = NULL; + goto tts_cleanup_and_exit; + } - msleep(1000); - WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "等待TTS WebSocket握手完成... (%d ms)", - (uint32_t)osKernelGetTickCount() - tts_handshake_start_time); + msleep(1000); + WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "等待TTS WebSocket握手完成... (%d ms)", + (uint32_t)osKernelGetTickCount() - tts_handshake_start_time); } WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "TTS WebSocket握手完成，耗时 %d ms", - (uint32_t)osKernelGetTickCount() - tts_handshake_start_time); + (uint32_t)osKernelGetTickCount() - tts_handshake_start_time); + #endif server_state = 0x04 ^ server_state; tts_break = 0; @@ -2142,8 +2219,8 @@ TTS_SATRT: memset_s(tts_recv_buf, 1024 * 4, 0, 1024 * 4); 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"); + if (type == WS_CLOSE || len == -1 || (bk_websocket_tts_tick!=0 && tts_reconnect == 1)) { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "TTS WebSocket connection closed %d %d %d",type,len,tts_reconnect); break; } //发送数据到AI @@ -2185,16 +2262,16 @@ TTS_SATRT: AI_SAFE_FREE(tts_recv_buf); ai_update_memory_usage(-1024 * 4); + ws_queue_cleanup(&tts_msg_queue); + lwip_close_websocket_client(tts_client); tts_client = NULL; - if (tts_reconnect) { - if (disconnect_server == 0) { - set_clear_queue_stream(1); - retry_count = 0; - WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "Reconnecting TTS client..."); - - } + if (disconnect_server==0) { + + set_clear_queue_stream(1); + retry_count = 0; + WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "Reconnecting TTS client..."); tts_reconnect = 0; goto TTS_SATRT; } @@ -2238,7 +2315,7 @@ errcode_t tts_server_init(void) tts_taskcfg.create_queue_handle = NULL; tts_taskcfg.attr.name="tts_server_init"; tts_taskcfg.attr.priority = TASK_PRIORITY_APP; - tts_taskcfg.attr.stack_size = 0x1000; + tts_taskcfg.attr.stack_size = 0x2000; tts_taskcfg.func = bk_websocket_tts; tts_taskcfg.use_mem = USE_EXTEND_MEM; @@ -3020,7 +3097,7 @@ void ai_cleanup_resources(void) { // } // 清理WebSocket消息队列 - ws_queues_deinit_all(); + // ws_queues_deinit_all(); // 清理互斥锁 if (g_ai_mutex) { @@ -3040,7 +3117,7 @@ void ai_cleanup_resources(void) { AI_UNLOCK(); // 最后清理TCP管理器 - tcp_simple_manager_deinit(); + // tcp_simple_manager_deinit(); WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "AI resources cleaned up"); } diff --git a/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ws.cpp b/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ws.cpp index 8658942bc..38f1eddd9 100644 --- a/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ws.cpp +++ b/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ws.cpp @@ -5,6 +5,11 @@ #include #include #include +#include +#include +#include +#include +#include #include #include "lwip/ip.h" @@ -59,6 +64,8 @@ // 包含新的WebSocket解析器 #include "websocket_parser_wrapper.h" +#define TEST_AI_WS 0 + #ifdef __cplusplus extern "C" { #endif /* __cplusplus */ @@ -375,6 +382,7 @@ void generate_accept_key(const char *key, char *accept_key) { * @param headers 头信息 * @return 成功返回0，失败返回-1 */ +#if (TEST_AI_WS==0) int send_handshake(WebSocketClient *client,const char *headers,char *response,int response_len) { char key[WEBSOCKET_KEY_LEN+1]; WS_LOG_INFO("send_handshake1"); @@ -390,13 +398,17 @@ int send_handshake(WebSocketClient *client,const char *headers,char *response,in int ret = 0; generate_accept_key(key, accept_key); WS_LOG_INFO("send_handshake2"); + char *request =( char *)aic_zalloc(BUFFER_SIZE); + if(request==NULL) { WS_LOG_ERROR("request aic_zalloc error"); return -1; } - + + memset(request,0,BUFFER_SIZE); + if(headers!= NULL) { snprintf(request, BUFFER_SIZE, @@ -438,6 +450,7 @@ int send_handshake(WebSocketClient *client,const char *headers,char *response,in // 检查响应内容 if (ret > 0 && response[0] != '\0') { + WS_LOG_WARNING("WebSocket %s:\r\n", client->path); WS_LOG_WARNING("WebSocket握手响应:\r\n%s", response); // 检查是否包含"HTTP/1.1 101" @@ -458,6 +471,62 @@ int send_handshake(WebSocketClient *client,const char *headers,char *response,in WS_LOG_INFO("send_handshake3 返回: %d", ret); return ret; } +#else + +int send_handshake(WebSocketClient *client,const char *headers) { + char key[WEBSOCKET_KEY_LEN+1]; + printf("send_handshake1"); + generate_websocket_key(key); + char accept_key[64]; + int ret = 0; + generate_accept_key(key, accept_key); + printf("send_handshake2"); + char *request =( char *)aic_zalloc(BUFFER_SIZE); + if(request==NULL) + { + printf("request aic_zalloc erro"); + return -1; + } + + + if(headers!= NULL) + { + snprintf(request, BUFFER_SIZE, + "GET /%s?%s HTTP/1.1\r\n" + "Host: %s:%d\r\n" + "Upgrade: websocket\r\n" + "Connection: Upgrade\r\n" + "Sec-WebSocket-Key: %s\r\n" + "Sec-WebSocket-Version: 13\r\n" + "Origin: http://%s:%d\r\n" + "%s\r\n" + "\r\n", + client->path,client->query ,client->host,client->port, accept_key,client->host,client->port,headers); + }else + { + snprintf(request, BUFFER_SIZE, + "GET /%s?%s HTTP/1.1\r\n" + "Host: %s:%d\r\n" + "Upgrade: websocket\r\n" + "Connection: Upgrade\r\n" + "Sec-WebSocket-Key: %s\r\n" + "Sec-WebSocket-Version: 13\r\n" + "Origin: http://%s:%d\r\n" + "\r\n", + client->path,client->query , client->host,client->port, accept_key,client->host,client->port); + } + + printf("send_handshake request:\r\n%s",request); + if (lwip_send(client->sockfd, request, strlen(request), 0) < 0) { + printf("lwip_send handshake failed"); + ret = -1; + } + + free(request); + printf("send_handshake3 %d",ret); + return ret; +} +#endif // 发送 WebSocket 消息 @@ -552,7 +621,9 @@ send_websocket_message_ERROR: void _tcp_on_connect_cb(tcp_handle_t handle, bool success) { WS_LOG_INFO("tcp_on_connect_cb %p %d", handle, success); - + #if TEST_AI_WS + return; + #else if (success) { // 连接成功，查找对应的WebSocket客户端 extern WebSocketClient * get_ws_client_by_handle(void* handle); @@ -606,6 +677,7 @@ void _tcp_on_connect_cb(tcp_handle_t handle, bool success) } else { WS_LOG_INFO("TCP连接失败"); } + #endif } void _tcp_on_disconnect_cb(tcp_handle_t handle) { @@ -665,14 +737,187 @@ void _tcp_on_data_received_cb(tcp_handle_t handle, uint8_t* data, uint16_t datal WS_LOG_WARNING("Processed %d bytes %s %s of WebSocket data", processed, client->host, client->path); } +#if TEST_AI_WS + +struct cat1_ws_init_t +{ + osSemaphoreId_t dns_done_sem; + ip_addr_t ip; +}; +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; + int sockfd = -1; + + err_t ret=0; + + parsed_url = parse_url(url);//解析url + + if (parsed_url.error) { + printf("init_websocket_client Error: %s", parsed_url.error_message); + goto client_ERROR; + } else { + printf("Protocol: %s", parsed_url.protocol); + printf("Host: %s\n", parsed_url.host); + printf("Port: %d\n", parsed_url.port); + printf("Path: %s\n", parsed_url.path); + printf("Query: %s\n", parsed_url.query); + + } + client = (WebSocketClient *)aic_zalloc(sizeof(WebSocketClient)) ; + if(!client) + { + printf("init_websocket_client alloc error"); + goto client_ERROR; + } + + + + sockfd = lwip_socket(AF_INET, SOCK_STREAM, 0); + client->sockfd=sockfd; + + if (sockfd < 0) { + + printf("socket creation failed %d errno: %d\r\n",sockfd,errno); + goto client_ERROR; + } + lwip_setsockopt(sockfd, SOL_SOCKET, SO_KEEPALIVE, (const void*)&opt, sizeof(opt)); + lwip_setsockopt(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 + { + 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 (lwip_connect(sockfd, (struct sockaddr *)&server_addr, sizeof(server_addr)) < 0) { + printf("init_websocket_client connection failed\r\n"); + msleep(100); + lwip_close(sockfd); + + goto client_ERROR; + + } + printf("socket creation succ\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) { //发送头 + printf("socket send_handshake client_ERROR"); + + lwip_close(sockfd); + 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(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); + } + + + + if(strstr(buffer,"HTTP/1.1 101") == NULL) + { + + goto client_ERROR; + } + printf("init_websocket_client end 1"); + + + printf("init_websocket_client end 2"); + + client->tcp_handle = tcp_add_existing_connection(sockfd, parsed_url.host, parsed_url.port, + _tcp_on_connect_cb, _tcp_on_disconnect_cb, _tcp_on_data_received_cb); + + client->parser = ws_parser_init(); + + free(buffer); + free_url(parsed_url); + + client->handshake=1; + + if (!client->parser) { + WS_LOG_ERROR("init_websocket_client Error: Failed to initialize WebSocket parser"); + lwip_close(sockfd); + goto client_ERROR; + } + + return client; +client_ERROR: + WS_LOG_ERROR("init_websocket_client Error: client_ERROR"); + if(buffer) { + free(buffer); + buffer=NULL; + } + if(client){ + free(client); + client = NULL; + } + free_url(parsed_url); + + return NULL; +} +#else // 初始化 WebSocket 客户端 WebSocketClient *init_websocket_client(const char *url, const char *headers,uint8_t use_dns) { //printf("init_websocket_client 1\n"); - WebSocketClient *client = NULL; + WebSocketClient *client = NULL; BK_URL parsed_url = {0}; int opt = 1; err_t ret = 0; - + if (!url) { WS_LOG_ERROR("init_websocket_client Error: URL is NULL"); return NULL; @@ -708,7 +953,7 @@ WebSocketClient *init_websocket_client(const char *url, const char *headers,uint goto client_ERROR; } WS_LOG_INFO("socket creation success"); - + strcpy(client->host, parsed_url.host); strcpy(client->path, parsed_url.path); strcpy(client->query, parsed_url.query); @@ -724,7 +969,7 @@ WebSocketClient *init_websocket_client(const char *url, const char *headers,uint /* 握手将在连接成功回调中发送，这里不再立即发送 */ WS_LOG_INFO("WebSocket客户端初始化完成，等待TCP连接建立"); - // if(strstr(buffer,"HTTP/1.1 101") == NULL) +// if(strstr(buffer,"HTTP/1.1 101") == NULL) // { // goto client_ERROR; @@ -735,6 +980,7 @@ WebSocketClient *init_websocket_client(const char *url, const char *headers,uint WS_LOG_INFO("init_websocket_client end 2"); return client; + client_ERROR: WS_LOG_ERROR("init_websocket_client Error: Initialization failed, cleaning up..."); if(client){ @@ -745,6 +991,7 @@ client_ERROR: return NULL; } +#endif // 关闭 WebSocket 客户端 void lwip_close_websocket_client(WebSocketClient *client) { @@ -804,11 +1051,11 @@ int ws_process_received_data(WebSocketClient *client, const uint8_t *data, size_ // printf("Raw data (hex): "); // // 打印所有数据的十六进制表示 // for (size_t i = 0; i < len; i++) { - // printf("%02X ", data[i]); - // // 每16字节换行，提高可读性 - // if ((i + 1) % 16 == 0) { - // printf("\n "); - // } + // printf("%02X ", data[i]); + // // 每16字节换行，提高可读性 + // if ((i + 1) % 16 == 0) { + // printf("\n "); + // } // } // printf("\n"); @@ -877,7 +1124,7 @@ int ws_process_received_data(WebSocketClient *client, const uint8_t *data, size_ while (remaining_len > 0 && loop_count < max_loops) { loop_count++; - WS_LOG_DEBUG("ws_process_received_data: processing remaining %zu bytes, current offset: %zu, loop %zu", + WS_LOG_DEBUG("ws_process_received_data: processing remaining %d bytes, current offset: %zu, loop %d", remaining_len, current_data - data, loop_count); // 在解析前打印当前解析器状态 @@ -936,6 +1183,7 @@ int ws_process_received_data(WebSocketClient *client, const uint8_t *data, size_ case WS_PING: WS_LOG_WARNING("Received WebSocket PING"); // TODO: 自动回复PONG + send_websocket_message(client,( uint8_t *)"PONG",strlen("PONG"),WS_PONG); break; case WS_PONG: @@ -974,7 +1222,7 @@ int ws_process_received_data(WebSocketClient *client, const uint8_t *data, size_ - WS_LOG_DEBUG("ws_process_received_data: frame processed, moving to next frame. Remaining: %zu bytes", remaining_len); + WS_LOG_DEBUG("ws_process_received_data: frame processed, moving to next frame. Remaining: %d bytes", remaining_len); } else { WS_LOG_DEBUG("ws_process_received_data: invalid payload data - header=%p, payload=%p, payload_len=%zu", @@ -986,7 +1234,7 @@ int ws_process_received_data(WebSocketClient *client, const uint8_t *data, size_ // 解析错误,重置parser ws_parser_reset(parser); - WS_LOG_DEBUG("ws_process_received_data: parsing error, find next frame. Remaining: %zu bytes", remaining_len); + WS_LOG_ERROR("ws_process_received_data: ws_stat == WS_STATE_ERROR parsing error, find next frame. Remaining: %zu bytes", remaining_len); } else { // 数据不完整，继续等待更多数据 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 c28164fe9..c02836004 100644 --- a/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.c +++ b/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.c @@ -1418,6 +1418,7 @@ tcp_manager_result_t tcp_manager_close_directly(tcp_handle_t handle) tcp_connection_state_t tcp_get_connection_state(tcp_handle_t handle) { if (!g_manager_initialized || !handle) { + printf("g_manager_initialized(%d) handle (%s)",g_manager_initialized,handle); return TCP_STATE_DISCONNECTED; } @@ -1516,21 +1517,21 @@ int tcp_send_and_wait_response(tcp_handle_t handle, const uint8_t* send_data, ui 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"); + printf("tcp_send_and_wait_response: 无效参数 (%s) (%s) (%d)\n",handle,send_data,send_len); return -1; } /* 检查是否在IO线程中 */ - if (!is_in_io_thread()) { - printf("tcp_send_and_wait_response: 错误！只能在IO线程中调用\n"); - return -1; - } + // 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"); + if (conn->state != TCP_STATE_CONNECTED || conn->sockfd < 0) { + printf("tcp_send_and_wait_response: 连接未建立或已关闭 \n"); return -1; } @@ -1542,14 +1543,14 @@ int tcp_send_and_wait_response(tcp_handle_t handle, const uint8_t* send_data, ui 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 (sent < 0) { +if (errno == EAGAIN || errno == EWOULDBLOCK) { /* 发送缓冲区满，等待后重试 */ if ((sys_now() - start_time) > timeout_ms) { printf("tcp_send_and_wait_response: 发送超时 (缓冲区满)\n"); - return -1; - } - /* 短暂等待后重试 */ + return -1; + } +/* 短暂等待后重试 */ sys_msleep(5); // 增加等待时间到5ms continue; } else if (errno == EINPROGRESS) { @@ -1583,37 +1584,37 @@ int tcp_send_and_wait_response(tcp_handle_t handle, const uint8_t* send_data, ui struct timeval select_timeout; while ((sys_now() - start_time) < timeout_ms) { - FD_ZERO(&read_fds); - FD_SET(conn->sockfd, &read_fds); + 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; + /* 计算剩余超时时间 */ + uint32_t elapsed = sys_now() - start_time; + uint32_t remaining_ms = (elapsed < timeout_ms) ? (timeout_ms - elapsed) : 0; - if (remaining_ms == 0) { - break; - } + if (remaining_ms == 0) { + break; + } - select_timeout.tv_sec = remaining_ms / 1000; - select_timeout.tv_usec = (remaining_ms % 1000) * 1000; + 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; - } + 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)) { + /* 有数据可读 */ + 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 { + } else { printf("tcp_send_and_wait_response: recv失败, errno=%d\n", errno); return -1; } @@ -1624,7 +1625,7 @@ int tcp_send_and_wait_response(tcp_handle_t handle, const uint8_t* send_data, ui printf("tcp_send_and_wait_response: 接收到 %d 字节响应\n", recv_ret); return recv_ret; } - } + } } printf("tcp_send_and_wait_response: 等待响应超时\n"); @@ -1658,4 +1659,69 @@ int tcp_get_debug_level(void) bool tcp_manager_is_initialized(void) { return g_manager_initialized; +} + +/** + * @brief 添加现有的TCP连接到管理器 + */ +tcp_handle_t tcp_add_existing_connection(int sockfd, const char* host, uint16_t port, + tcp_on_connect_cb connect_cb, + tcp_on_disconnect_cb disconnect_cb, + tcp_on_data_received_cb data_cb) +{ + if (!g_manager_initialized || sockfd < 0 || !host) { + TCPM_ERR("TCP管理器未初始化或参数无效"); + return NULL; + } + + TCPM_INFO("添加现有连接: %s:%d, sockfd=%d", host, port, sockfd); + + sys_mutex_lock(&g_connections_mutex); + + tcp_connection_info_t* conn = find_free_connection_unsafe(); + if (!conn) { + sys_mutex_unlock(&g_connections_mutex); + TCPM_ERR("添加连接失败: 没有可用的连接槽位"); + return NULL; + } + + /* 初始化连接 */ + memset(conn, 0, sizeof(tcp_connection_info_t)); + conn->in_use = true; + conn->handle = conn; + conn->state = TCP_STATE_CONNECTED; /* 已经是连接状态 */ + conn->sockfd = sockfd; + conn->remote_port = port; + conn->connect_cb = connect_cb; + conn->disconnect_cb = disconnect_cb; + + conn->data_cb = data_cb; + + conn->last_activity = sys_now(); + conn->send_queue_head = 0; + conn->send_queue_tail = 0; + conn->send_queue_count = 0; + conn->retry_count = 0; + conn->retry_pending = false; + + /* 复制主机名 */ + strncpy(conn->remote_ip, host, sizeof(conn->remote_ip) - 1); + conn->remote_ip[sizeof(conn->remote_ip) - 1] = '\0'; + + /* 设置Keep-Alive */ + setup_keepalive(sockfd); + + /* 更新统计信息 */ + g_stats.total_connections++; + g_stats.active_connections++; + + sys_mutex_unlock(&g_connections_mutex); + + printf("成功添加现有连接: %s:%d, handle=%p", host, port, conn->handle); + /* 调用连接成功回调 */ + if (connect_cb) { + connect_cb(conn->handle, true); + } + + return conn->handle; } \ 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 a7171d3c9..983b47c2a 100644 --- a/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.h +++ b/code/sdk/drivers/peripheral/slip/tcpmanager/tcp_simple_manager.h @@ -330,6 +330,23 @@ int tcp_get_debug_level(void); */ bool tcp_manager_is_initialized(void); +/** + * @brief 添加现有的TCP连接到管理器 + * 用于将已经建立的TCP连接（如通过其他方式创建的socket）添加到管理器中 + * + * @param sockfd 已建立的socket文件描述符 + * @param host 远程主机地址 + * @param port 远程端口 + * @param connect_cb 连接状态回调 + * @param disconnect_cb 断开连接回调 + * @param data_cb 数据接收回调 + * @return tcp_handle_t 连接句柄，失败返回NULL + */ +tcp_handle_t tcp_add_existing_connection(int sockfd, const char* host, uint16_t port, + tcp_on_connect_cb connect_cb, + tcp_on_disconnect_cb disconnect_cb, + tcp_on_data_received_cb data_cb); + #ifdef __cplusplus } #endif -- 2.17.1