/* WebSocket解析器 */
ws_parser_t* parser;
+ int sockfd; /* socket文件描述符 */
/* 数据包回调函数 */
on_ws_packet_callback packet_callback;
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;
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
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);
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;
AI_UNLOCK();
char_reconnet:
+ ws_queue_cleanup(&chat_msg_queue);
+
{
char url_buf[248] = {0};
uint8_t is_raisound_url = 0;
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;
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:
AI_LOCK();
is_bk_websocket_asr = 1;
AI_UNLOCK();
-
+ ws_queue_cleanup(&asr_msg_queue);
+
uint8_t is_raisound_url = 0;
{
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;
lwip_close_websocket_client(asr_client);
asr_client = NULL;
+ ws_queue_cleanup(&asr_msg_queue);
asr_client_ping = 0;
msleep(1000);
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);
}
TTS_SATRT:
uapi_cat1_keep_wakeup(1);
-
+ ws_queue_cleanup(&tts_msg_queue);
WebSocketMessageType type;
bool state = true;
tts_client = NULL;
goto tts_cleanup_and_exit;
}
+ #if TEST_AI_WS
+ #else
WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "TTS WebSocket: TCP连接确认成功，开始握手等待");
// 等待TTS WebSocket握手完成，增加超时机制
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;
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
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;
}
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;
// }
// 清理WebSocket消息队列
- ws_queues_deinit_all();
+ // ws_queues_deinit_all();
// 清理互斥锁
if (g_ai_mutex) {
AI_UNLOCK();
// 最后清理TCP管理器
- tcp_simple_manager_deinit();
+ // tcp_simple_manager_deinit();
WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "AI resources cleaned up");
}
#include <stdio.h>
#include <string.h>
#include <strings.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
#include <dirent.h>
#include "lwip/ip.h"
// 包含新的WebSocket解析器
#include "websocket_parser_wrapper.h"
+#define TEST_AI_WS 0
+
#ifdef __cplusplus
extern "C" {
#endif /* __cplusplus */
* @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");
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,
// 检查响应内容
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"
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 消息
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);
} else {
WS_LOG_INFO("TCP连接失败");
}
+ #endif
}
void _tcp_on_disconnect_cb(tcp_handle_t handle)
{
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;
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);
/* 握手将在连接成功回调中发送，这里不再立即发送 */
WS_LOG_INFO("WebSocket客户端初始化完成，等待TCP连接建立");
- // if(strstr(buffer,"HTTP/1.1 101") == NULL)
+// if(strstr(buffer,"HTTP/1.1 101") == NULL)
// {
// goto client_ERROR;
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){
return NULL;
}
+#endif
// 关闭 WebSocket 客户端
void lwip_close_websocket_client(WebSocketClient *client) {
// 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");
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);
// 在解析前打印当前解析器状态
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:
- 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",
// 解析错误,重置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 {
// 数据不完整，继续等待更多数据
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;
}
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;
}
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) {
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;
}
printf("tcp_send_and_wait_response: 接收到 %d 字节响应\n", recv_ret);
return recv_ret;
}
- }
+ }
}
printf("tcp_send_and_wait_response: 等待响应超时\n");
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
*/
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
projects / mirrors / FD230.git / commitdiff