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*)
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)
WS_PING = 0x9,\r
WS_PONG = 0xa\r
} WebSocketMessageType;\r
+\r
+/* WebSocket数据包回调函数类型 */\r
+typedef void (*on_ws_packet_callback)(uint8_t* data, size_t datalen, WebSocketMessageType type);\r
+\r
+/* WebSocket组包缓冲区配置 */\r
+#define WS_FRAME_BUFFER_SIZE 256\r
+#define WS_PAYLOAD_BUFFER_SIZE 128\r
+\r
+/* WebSocket组包状态 */\r
+typedef enum {\r
+ WS_PARSE_HEADER, /* 解析帧头 */\r
+ WS_PARSE_EXTENDED, /* 解析扩展长度 */\r
+ WS_PARSE_MASK, /* 解析掩码 */\r
+ WS_PARSE_PAYLOAD /* 解析载荷数据 */\r
+} ws_parse_state_t;\r
+\r
+/* WebSocket组包缓冲区结构 */\r
+typedef struct {\r
+ uint8_t frame_buffer[WS_FRAME_BUFFER_SIZE]; /* 帧缓冲区 */\r
+ uint32_t frame_buffer_len; /* 帧缓冲区当前长度 */\r
+ \r
+ uint8_t payload_buffer[WS_PAYLOAD_BUFFER_SIZE]; /* 载荷缓冲区 */\r
+ uint32_t payload_buffer_len; /* 载荷缓冲区当前长度 */\r
+ \r
+ ws_parse_state_t parse_state; /* 解析状态 */\r
+ uint8_t fin; /* FIN标志 */\r
+ WebSocketMessageType opcode; /* 操作码 */\r
+ uint8_t mask; /* 掩码标志 */\r
+ uint64_t payload_len; /* 载荷长度 */\r
+ uint32_t payload_received; /* 已接收载荷长度 */\r
+ uint8_t masking_key[4]; /* 掩码密钥 */\r
+ \r
+ on_ws_packet_callback packet_callback; /* 数据包回调函数 */\r
+} ws_packet_parser_t;\r
+\r
+/* WebSocket消息队列系统 */\r
+#define WS_QUEUE_SIZE 4\r
+#define WS_MESSAGE_MAX_SIZE 512\r
+\r
+typedef struct {\r
+ uint8_t data[WS_MESSAGE_MAX_SIZE];\r
+ uint16_t length;\r
+ WebSocketMessageType type;\r
+ uint32_t timestamp;\r
+ bool in_use;\r
+} ws_message_item_t;\r
+\r
+typedef struct {\r
+ ws_message_item_t messages[WS_QUEUE_SIZE];\r
+ uint8_t head;\r
+ uint8_t tail;\r
+ uint8_t count;\r
+ osMutexId_t mutex;\r
+ osSemaphoreId_t semaphore;\r
+ bool initialized;\r
+} ws_message_queue_t;\r
+\r
+\r
// WebSocket 客户端结构体\r
typedef struct {\r
- int sockfd;\r
+ void* tcp_handle;\r
+ ws_message_queue_t *msg_queue;\r
char host[128];\r
char path[128];\r
char query[128];\r
int port;\r
+ \r
+ /* 组包解析器 */\r
+ ws_packet_parser_t parser;\r
} WebSocketClient;\r
\r
\r
// 初始化 WebSocket 客户端\r
WebSocketClient *init_websocket_client(const char *url, const char *headers,uint8_t use_dns);\r
\r
+// 设置WebSocket数据包回调函数\r
+void ws_set_packet_callback(WebSocketClient *client, on_ws_packet_callback callback);\r
+\r
+// 处理接收到的原始数据\r
+int ws_process_received_data(WebSocketClient *client, const uint8_t *data, size_t len);\r
+\r
// 关闭 WebSocket 客户端\r
void lwip_close_websocket_client(WebSocketClient *client) ;\r
\r
#include <lwip/netdb.h>
// #include <netinet/in.h>
#include "lwip/inet.h"
+#include "../tcpmanager/tcp_simple_manager.h"
#include <stdio.h>
// #include <string.h>
// #include <sys/param.h>
// 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;
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);
// 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;
{
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;
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};
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
}
}
// 设置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();
}
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");
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;
}
}
// 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();
}
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);
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) {
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();
}
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");
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) {
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");
}
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");
}
-
-
-
-
-
-
-
-
// 资源清理函数
void ai_cleanup_resources(void) {
AI_LOCK();
+ // 清理WebSocket消息队列
+ ws_queues_deinit_all();
+
// 清理互斥锁
if (g_ai_mutex) {
osMutexDelete(g_ai_mutex);
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;
}
}
#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");
+// }
+
#include "lwip/sockets.h"
#include "lwip/sys.h"
#include <lwip/netdb.h>
+
+/* 添加缺失的定义 */
+#ifndef SHUT_RDWR
+#define SHUT_RDWR 2
+#endif
// #include <netinet/in.h>
#include "lwip/inet.h"
#include <stdio.h>
#include <bleService.h>
#include "cat1_ws.h"
#include "dev_authentication.h"
-
+#include "tcp_simple_manager.h"
#ifdef __cplusplus
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)
}
// 发送 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;
return -1;
}
-
if(headers!= NULL)
{
snprintf(request, BUFFER_SIZE,
}
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;
// 发送帧
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;
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;i<offset;i++)
- // {
- // printf("%02X",frame[i]);
- // }
- // printf("receive_websocket_message:payload_len=%d frame[0]&0x8E=%d",payload_len,frame[0]&0x8E);
- if (mask) {
-
- bytes_received = lwip_recv(client->sockfd, 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;
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);
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
#include <string.h>
#include <unistd.h>
#include <errno.h>
+
+/* 运行时调试等级控制 */
+static int g_tcp_debug_level = TCP_DEBUG_LEVEL;
/* 使用lwIP网络头文件替代标准网络头文件 */
#include "lwip/sockets.h" /* 包含lwIP socket API和fd_set等定义 */
#include "lwip/netdb.h"
#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 /* 重试延迟 */
/* 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);
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;
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) {
/* 创建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");
}
g_manager_initialized = true;
- printf("TCP管理器初始化成功\n");
+ TCPM_INFO("TCP管理器初始化成功");
return TCP_MANAGER_OK;
}
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;
}
/* 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);
/* 创建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);
/* 设置非阻塞 */
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);
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);
conn->sockfd = -1;
sys_mutex_unlock(&g_connections_mutex);
+
+ /* 调用连接失败回调 */
+ if (connect_cb) {
+ connect_cb(conn->handle, false);
+ }
return NULL;
}
}
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;
}
(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
#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
} 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 /* 主机名最大长度 */
*/
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
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;
}
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;
}
}
// 添加额外的安全检查
- if (test_client->sockfd < 0) {
+ if (test_client->tcp_handle == NULL) {
WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "发送数据失败: 无效的socket");
return false;
}
}
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;
projects / mirrors / FD230.git / commitdiff