--- /dev/null
+# WebSocket测试TCP管理器集成说明
+
+## 概述
+
+为 `websocket_test.c` 添加了TCP简单管理器的自动初始化和清理功能，确保WebSocket测试能够正确使用底层的TCP连接管理。
+
+## 🔧 主要修改
+
+### 1. 添加TCP管理器支持
+
+#### 新增头文件
+```c
+#include "../tcpmanager/tcp_simple_manager.h"
+```
+
+#### 新增全局状态变量
+```c
+static bool tcp_manager_initialized = false;
+```
+
+### 2. 测试开始时自动初始化TCP管理器
+
+在 `init_websocket_test()` 函数中添加：
+
+```c
+// 初始化TCP简单管理器
+if (!tcp_manager_initialized) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "初始化TCP简单管理器");
+ tcp_manager_result_t tcp_result = tcp_simple_manager_init();
+ if (tcp_result != TCP_MANAGER_OK) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "TCP管理器初始化失败: %d", tcp_result);
+ return false;
+ }
+ tcp_manager_initialized = true;
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "TCP管理器初始化成功");
+}
+```
+
+### 3. 测试结束时自动清理TCP管理器
+
+在 `cleanup_websocket_test()` 函数中添加：
+
+```c
+// 清理TCP简单管理器
+if (tcp_manager_initialized) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "清理TCP简单管理器");
+ tcp_simple_manager_deinit();
+ tcp_manager_initialized = false;
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "TCP管理器清理完成");
+}
+```
+
+### 4. 错误处理改进
+
+在所有可能的错误路径中添加TCP管理器清理：
+
+```c
+// 如果WebSocket客户端创建失败，清理TCP管理器
+if (tcp_manager_initialized) {
+ tcp_simple_manager_deinit();
+ tcp_manager_initialized = false;
+}
+```
+
+### 5. 新增停止测试AT命令
+
+添加了 `AT^STOPWEBSOCKET` 命令处理器：
+
+```c
+int at_stop_websocket_handler(const char *cmd, const char *params)
+{
+ if (strncmp(cmd, "AT^STOPWEBSOCKET", 16) != 0) {
+ return -1;
+ }
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "接收到停止WebSocket测试命令");
+
+ if (test_state != TEST_STATE_RUNNING) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket测试当前未运行");
+ return 0;
+ }
+
+ // 停止测试
+ stop_websocket_test();
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket测试已停止");
+ return 0;
+}
+```
+
+## 📋 功能特性
+
+### 自动资源管理
+- ✅ 测试开始时自动初始化TCP管理器
+- ✅ 测试结束时自动清理TCP管理器
+- ✅ 错误情况下的资源清理
+- ✅ 重复测试的状态管理
+
+### 状态跟踪
+- ✅ TCP管理器初始化状态跟踪
+- ✅ 防止重复初始化
+- ✅ 确保清理的完整性
+
+### 错误处理
+- ✅ 初始化失败时的回滚
+- ✅ 资源创建失败时的清理
+- ✅ 强制清理机制
+
+### AT命令支持
+- ✅ `AT^TESTWEBSOCKET=<server>,<port>` - 启动测试
+- ✅ `AT^STOPWEBSOCKET` - 停止测试
+
+## 🔍 技术细节
+
+### TCP管理器生命周期
+
+1. **初始化阶段**：
+ ```
+ WebSocket测试启动 -> 检查TCP管理器状态 -> 初始化TCP管理器 -> 创建WebSocket客户端
+ ```
+
+2. **运行阶段**：
+ ```
+ TCP管理器提供连接管理 -> WebSocket客户端使用TCP连接 -> 数据传输
+ ```
+
+3. **清理阶段**：
+ ```
+ 测试停止 -> 关闭WebSocket连接 -> 清理TCP管理器 -> 释放所有资源
+ ```
+
+### 错误恢复机制
+
+- **初始化失败**: 自动回滚已分配的资源
+- **运行时错误**: 通过 `cleanup_websocket_test()` 统一清理
+- **强制清理**: AT命令处理器中的强制清理机制
+
+## 📊 使用流程
+
+### 正常使用流程
+
+1. **启动测试**：
+ ```
+ AT^TESTWEBSOCKET=snicker.com.cn,8910
+ ```
+ - 自动初始化TCP管理器
+ - 建立WebSocket连接
+ - 开始数据传输测试
+
+2. **停止测试**：
+ ```
+ AT^STOPWEBSOCKET
+ ```
+ - 停止数据传输
+ - 关闭WebSocket连接
+ - 自动清理TCP管理器
+
+### 错误恢复流程
+
+如果测试过程中出现问题：
+1. TCP管理器会自动处理连接错误
+2. WebSocket测试会检测到连接状态变化
+3. 自动触发清理流程
+4. 释放所有资源回到空闲状态
+
+## 🎯 测试验证
+
+### 测试场景
+
+1. **正常启动/停止循环**
+2. **重复测试执行**
+3. **异常情况恢复**
+4. **资源泄漏检查**
+
+### 日志输出示例
+
+```
+WS: 初始化TCP简单管理器
+WS: TCP管理器初始化成功
+WS: 连接WebSocket服务器: ws://snicker.com.cn:8910/
+WS: WebSocket测试已启动，使用AT^STOPWEBSOCKET停止测试
+
+... 测试运行 ...
+
+WS: 接收到停止WebSocket测试命令
+WS: 清理TCP简单管理器
+WS: TCP管理器清理完成
+WS: WebSocket测试已停止
+```
+
+## 🔄 兼容性
+
+### 向后兼容
+- ✅ 保持原有的测试接口不变
+- ✅ 原有的功能全部保留
+- ✅ 日志输出格式保持一致
+
+### 新功能
+- ✅ 自动TCP管理器生命周期管理
+- ✅ 增强的错误处理和恢复
+- ✅ 完整的AT命令支持
+
+## 📝 注意事项
+
+1. **TCP管理器单例**: TCP管理器在系统中是单例的，多个测试共享同一实例
+2. **状态跟踪**: 使用 `tcp_manager_initialized` 标志跟踪状态，避免重复操作
+3. **错误处理**: 所有错误路径都包含适当的清理代码
+4. **线程安全**: TCP管理器本身是线程安全的，但测试状态需要适当保护
+
+## 🚀 部署说明
+
+修改完成后，WebSocket测试程序具备了完整的TCP连接管理能力：
+
+- **无需手动管理**: TCP管理器自动初始化和清理
+- **错误恢复**: 自动处理各种异常情况
+- **资源安全**: 确保所有资源得到正确释放
+- **易于使用**: 通过AT命令即可完整测试
+
+这些改进使WebSocket测试更加健壮和易用，为嵌入式设备的WebSocket功能验证提供了可靠的测试工具。
\ No newline at end of file
${CMAKE_CURRENT_SOURCE_DIR}/cat_imei_map.cpp
${CMAKE_CURRENT_SOURCE_DIR}/cat1_ws.cpp
${CMAKE_CURRENT_SOURCE_DIR}/cat1_ai_server.cpp
+${CMAKE_CURRENT_SOURCE_DIR}/websocket_parser.cpp
--- /dev/null
+# WebSocket解析器日志功能说明
+
+## 概述
+
+本文档描述了添加到 `websocket_parser.cpp` 中的重要解析过程日志功能。这些日志帮助开发者调试WebSocket连接问题，跟踪帧解析过程，以及监控解析器状态。
+
+## 日志级别说明
+
+### 使用的日志级别
+
+- **WEARABLE_LOGI**: 信息级别 - 记录正常的解析过程和状态变化
+- **WEARABLE_LOGE**: 错误级别 - 记录解析错误和异常情况
+- **WEARABLE_LOGW**: 警告级别 - 记录潜在问题
+
+### 日志模块
+
+所有日志都使用 `WEARABLE_LOG_MODULE_APP` 模块标识。
+
+## 主要日志功能
+
+### 1. 解析器生命周期日志
+
+#### 初始化日志
+```c
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Initializing WebSocket parser");
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Allocating global buffer (%d bytes)", WS_BUFFER_SIZE);
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Parser count: %d", g_parser_count);
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Parser initialized successfully");
+```
+
+#### 销毁日志
+```c
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Destroying WebSocket parser");
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Parser count after destroy: %d", g_parser_count);
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Releasing global buffer (last parser destroyed)");
+```
+
+### 2. 帧头部解析日志
+
+#### 第一个头部字节
+```c
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Parse header1 - FIN:%d, RSV:%d%d%d, OPCODE:0x%02X",
+ parser->header.fin, parser->header.rsv1, parser->header.rsv2, parser->header.rsv3, parser->header.opcode);
+```
+
+#### 第二个头部字节
+```c
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Parse header2 - MASK:%d, LEN:%d",
+ parser->header.mask, payload_len);
+```
+
+### 3. 负载长度解析日志
+
+#### 小负载（≤125字节）
+```c
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Small payload length: %d bytes", payload_len);
+```
+
+#### 16位扩展长度
+```c
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Extended 16-bit payload length mode");
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: 16-bit payload length: %d bytes", parser->header.payload_len);
+```
+
+#### 64位扩展长度
+```c
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Extended 64-bit payload length mode");
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: 64-bit payload length: %llu bytes",
+ (unsigned long long)parser->header.payload_len);
+```
+
+### 4. 掩码处理日志
+
+```c
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Mask key read: %02X %02X %02X %02X",
+ parser->header.mask_key[0], parser->header.mask_key[1],
+ parser->header.mask_key[2], parser->header.mask_key[3]);
+```
+
+### 5. 负载数据处理日志
+
+```c
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Payload read complete (%llu bytes)",
+ (unsigned long long)parser->header.payload_len);
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Applying mask to payload data");
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Frame parsing complete");
+```
+
+### 6. 流解析日志
+
+```c
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Parsing stream data (%zu bytes)", len);
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Complete frame parsed (%d bytes total)", packet_len);
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Frame parsing incomplete, waiting for more data");
+```
+
+### 7. 错误处理日志
+
+#### 无效操作码
+```c
+WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WS: Invalid opcode: 0x%02X", parser->header.opcode);
+```
+
+#### 无效RSV位
+```c
+WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WS: Invalid RSV bits: %d%d%d",
+ parser->header.rsv1, parser->header.rsv2, parser->header.rsv3);
+```
+
+#### 控制帧长度验证失败
+```c
+WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WS: Control frame length validation failed: opcode=0x%02X, len=%d",
+ parser->header.opcode, payload_len);
+```
+
+#### 负载长度超限
+```c
+WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WS: Payload length %d exceeds maximum %llu",
+ parser->header.payload_len, (unsigned long long)g_max_payload_len);
+```
+
+#### 解析错误
+```c
+WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WS: Parse error at byte %zu, resetting parser", i);
+```
+
+### 8. 配置日志
+
+```c
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Setting max payload length to %llu bytes", (unsigned long long)max_len);
+WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Max payload length set successfully");
+```
+
+## 日志使用示例
+
+### 正常帧解析过程
+```
+WS: Parsing stream data (50 bytes)
+WS: Parse header1 - FIN:1, RSV:000, OPCODE:0x01
+WS: Parse header2 - MASK:1, LEN:20
+WS: Small payload length: 20 bytes
+WS: Mask key read: 12 34 56 78
+WS: Payload read complete (20 bytes)
+WS: Applying mask to payload data
+WS: Frame parsing complete
+WS: Complete frame parsed (26 bytes total)
+```
+
+### 错误情况日志
+```
+WS: Parsing stream data (10 bytes)
+WS: Parse header1 - FIN:1, RSV:001, OPCODE:0x01
+WS: Invalid RSV bits: 001
+WS: Parse error at byte 1, resetting parser
+```
+
+## 调试建议
+
+### 1. 连接问题调试
+- 查看解析器初始化日志确认解析器正常创建
+- 监控流数据输入日志确认数据到达
+- 检查解析完成日志确认帧处理状态
+
+### 2. 协议合规性问题
+- 检查RSV位和操作码验证日志
+- 监控控制帧长度验证日志
+- 查看负载长度限制日志
+
+### 3. 性能问题调试
+- 监控解析器计数日志确认资源使用
+- 查看缓冲区分配日志确认内存使用
+- 跟踪帧大小日志分析数据传输模式
+
+## 注意事项
+
+1. **日志开销**: 详细的日志可能影响性能，生产环境可能需要调整日志级别
+2. **内存占用**: 大量日志输出可能增加内存使用
+3. **日志文件大小**: 高频率的WebSocket通信会产生大量日志
+
+## 编译配置
+
+确保编译时包含 `wearable_log.h` 头文件，并且定义了相应的日志宏。
+
+```c
+#include "wearable_log.h"
+```
+
+日志功能已成功集成到编译系统中，无需额外配置即可使用。
\ No newline at end of file
#ifndef __BK_WEBSOCKET_H__
#define __BK_WEBSOCKET_H__
-// #include "aic_osa.h"
-// #include <bits/types.h>
-// WebSocket 消息类型枚举
+
+// 包含新的WebSocket解析器接口
+#include "websocket_parser_wrapper.h"
#ifdef __cplusplus
-#if __cplusplus
extern "C" {
#endif /* __cplusplus */
-#endif /* __cplusplus */
-typedef enum {
- WS_CONT = 0x00,
- WS_TEXT = 0x01,
- WS_BINARY = 0x02,
- WS_CLOSE = 0x8,
- WS_PING = 0x9,
- WS_PONG = 0xa
-} WebSocketMessageType;
+// 使用websocket_parser_wrapper.h中的枚举值，避免重复定义
+// 为保持向后兼容性，创建类型别名
+typedef ws_opcode_t WebSocketMessageType;
+
+// 定义向后兼容的常量（如果需要的话）
+#define WS_CONT WS_CONTINUATION
/* WebSocket数据包回调函数类型 */
typedef void (*on_ws_packet_callback)(uint8_t* data, size_t datalen, WebSocketMessageType type);
-/* WebSocket组包缓冲区配置 */
-#define WS_FRAME_BUFFER_SIZE 256
-#define WS_PAYLOAD_BUFFER_SIZE 128
-
-/* WebSocket组包状态 */
-typedef enum {
- WS_PARSE_HEADER, /* 解析帧头 */
- WS_PARSE_EXTENDED, /* 解析扩展长度 */
- WS_PARSE_MASK, /* 解析掩码 */
- WS_PARSE_PAYLOAD /* 解析载荷数据 */
-} ws_parse_state_t;
-
-/* WebSocket组包缓冲区结构 */
-typedef struct {
- uint8_t frame_buffer[WS_FRAME_BUFFER_SIZE]; /* 帧缓冲区 */
- uint32_t frame_buffer_len; /* 帧缓冲区当前长度 */
-
- uint8_t payload_buffer[WS_PAYLOAD_BUFFER_SIZE]; /* 载荷缓冲区 */
- uint32_t payload_buffer_len; /* 载荷缓冲区当前长度 */
-
- ws_parse_state_t parse_state; /* 解析状态 */
- uint8_t fin; /* FIN标志 */
- WebSocketMessageType opcode; /* 操作码 */
- uint8_t mask; /* 掩码标志 */
- uint64_t payload_len; /* 载荷长度 */
- uint32_t payload_received; /* 已接收载荷长度 */
- uint8_t masking_key[4]; /* 掩码密钥 */
-
- on_ws_packet_callback packet_callback; /* 数据包回调函数 */
-} ws_packet_parser_t;
-
/* WebSocket消息队列系统 */
#define WS_QUEUE_SIZE 4
#define WS_MESSAGE_MAX_SIZE 512
bool initialized;
} ws_message_queue_t;
-
// WebSocket 客户端结构体
typedef struct {
void* tcp_handle;
char path[128];
char query[128];
int port;
+
+ int handshake;
+
+ /* WebSocket解析器 */
+ ws_parser_t* parser;
- /* 组包解析器 */
- ws_packet_parser_t parser;
+ /* 数据包回调函数 */
+ on_ws_packet_callback packet_callback;
} WebSocketClient;
-
// 初始化 WebSocket 客户端
WebSocketClient *init_websocket_client(const char *url, const char *headers,uint8_t use_dns);
int receive_websocket_message(WebSocketClient *client, unsigned char *buf,int len,WebSocketMessageType *code);
int send_websocket_message(WebSocketClient *client, uint8_t *message, size_t message_len, WebSocketMessageType type) ;
+// WebSocket测试客户端（当SUPPORT_WEBSOCKET_TEST定义时可用）
+#ifdef SUPPORT_WEBSOCKET_TEST
+extern WebSocketClient *test_client;
+#endif
#ifdef __cplusplus
-#if __cplusplus
}
#endif /* __cplusplus */
-#endif /* __cplusplus */
// 发送 WebSocket 消息
// int send_websocket_message(WebSocketClient *client, uint8_t *message, size_t message_len, WebSocketMessageType type) ;
-
#endif
{
return tts_client;
}
+ // 检查WebSocket测试客户端
+#ifdef SUPPORT_WEBSOCKET_TEST
+ else if (test_client != NULL && test_client->tcp_handle == handle)
+ {
+ return test_client;
+ }
+#endif
// 注意：voice_chat_client 被注释掉了，根据用户的修改
// else if (voice_chat_client != NULL && voice_chat_client->tcp_handle == handle)
// {
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;
+ chat_client->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;
+ chat_client->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};
}
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;
+ chat_client->packet_callback = chat_on_ws_packet_callback;
#endif
}
goto cleanup_and_exit;
}
+ while (chat_client->handshake==0)
+ {
+ msleep(1000);
+ }
+
// 设置socket超时
// struct timeval timeout;
// timeout.tv_sec = AI_RECEIVE_TIMEOUT_MS / 1000;
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;
+ asr_client->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;
+ asr_client->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;
+ asr_client->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;
+ asr_client->packet_callback = asr_on_ws_packet_callback;
}
}
goto asr_cleanup_and_exit;
}
+ while (asr_client->handshake==0)
+ {
+ msleep(1000);
+ }
+
uapi_cat1_keep_wakeup(1);
server_state = 0x01 ^ server_state;
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;
+ tts_client->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;
+ tts_client->packet_callback = tts_on_ws_packet_callback;
}
if (tts_client == NULL) {
goto tts_cleanup_and_exit;
}
+ while (tts_client->handshake==0)
+ {
+ msleep(1000);
+ }
+
server_state = 0x04 ^ server_state;
tts_break = 0;
AI_LOCK();
// 首先关闭所有WebSocket连接
- extern void lwip_close_websocket_client(WebSocketClient *client);
if (chat_client) {
lwip_close_websocket_client(chat_client);
chat_client = NULL;
#include "dev_authentication.h"
#include "tcp_simple_manager.h"
+// 包含新的WebSocket解析器
+#include "websocket_parser_wrapper.h"
+
#ifdef __cplusplus
extern "C" {
#endif /* __cplusplus */
-
-// #include "sock_api/sock_api.h"
-// #include "app_config.h"
-// #include "system/includes.h"
-// #include "wifi/wifi_connect.h"
-// #include "lwip.h"
#include "lwip/sockets.h"
#include "lwip/netdb.h"
#include "cmsis_os2.h"
-// #include <nx_api.h>
-// #include <nx_ip.h>
-// #include <nx_ipv6.h>
-// #include <nx_tcp.h>
-// #include <nx_packet.h>
-// #include <nxd_bsd.h>
-// #include <nx_websocket_client.h>
-
-// #include <aic_init.h>
-// #include <aic_osa.h>
-// #include <aic_errno.h>
-// #include <aic_net_api.h>
-// #include <aic_net_pdp_mgr.h>
-// #include <aic_websocket_client.h>
-// #define printf(const char *format, ...) printf();
-
-// #define printf(fmt, ...) printf("[bk:%s,%d]"fmt,__func__, __LINE__,##__VA_ARGS__)
-
-// extern void bk_print_hex(uint8_t *data,int data_len);
#ifdef CONFIG_CRYPTO_MBEDTLS
#ifdef CONFIG_CRYPTO_MBEDTLS_2
#include "mbedtls/net.h"
#include <mbedtls/error.h>
#include <mbedtls/x509_crt.h>
-// typedef struct aic_websocket_mbedtls_session {
-// mbedtls_ssl_context ssl;
-// mbedtls_entropy_context entropy;
-// mbedtls_ctr_drbg_context ctr_drbg;
-// mbedtls_ssl_config conf;
-// mbedtls_net_context socket_fd;
-// CHAR status;
-// ULONG recv_wait_time;
-// } WEBSOCKET_MBEDTLS_SESSION;
#endif
#include "mbedtls/base64.h"
#define WEBSOCKET_KEY_LEN 24
#define WEBSOCKET_GUID "1234567890123456"
-// #include "fs/fs.h"
-// #include "spiflash.h"
-// #include "app_config.h"
-// #include "asm/sfc_norflash_api.h"
-// #include "event/device_event.h"
-// #include "system/includes.h"
-// #include "net_def.h"
#include "bk_websocket.h"
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)
}
-// 仿写 htonl 函数
-// static uint32_t htonl(uint32_t hostlong) {
-// // 检查主机是否为大端字节序
-// static const union {
-// uint32_t i;
-// char c[4];
-// } test = {0x01000000};
-// if (test.c[0] == 0x01) {
-// // 主机是大端字节序，不需要转换
-// return hostlong;
-// }
-
-// // 主机是小端字节序，进行字节序转换
-// return ((hostlong & 0x000000FF) << 24) |
-// ((hostlong & 0x0000FF00) << 8) |
-// ((hostlong & 0x00FF0000) >> 8) |
-// ((hostlong & 0xFF000000) >> 24);
-// }
// // 仿写 htonll 函数
static uint64_t htonll(uint64_t hostlonglong) {
// 检查主机是否为大端字节序
for (int i = 0; i < WEBSOCKET_KEY_LEN - 1; ++i) {
key[i] = charset[rand() % (sizeof(charset) - 1)];
}
- key[WEBSOCKET_KEY_LEN - 1] = '\0';
+ key[WEBSOCKET_KEY_LEN] = '\0';
unsigned char encoded[WEBSOCKET_KEY_LEN * 2];
size_t olen;
mbedtls_base64_encode(encoded, sizeof(encoded), &olen, (const unsigned char *)key, strlen(key));
printf("generate_websocket_key %s %d\r\n",encoded ,olen);
- if(olen>=(WEBSOCKET_KEY_LEN-1))
- olen=WEBSOCKET_KEY_LEN-1;
+ if(olen>=(WEBSOCKET_KEY_LEN))
+ olen=WEBSOCKET_KEY_LEN;
strncpy(key, (const char *)encoded, olen);
// 生成 WebSocket 握手响应密钥
void generate_accept_key(const char *key, char *accept_key) {
- char combined[128];
+ char combined[128]={0};
strcpy(combined, key);
strcat(combined, WEBSOCKET_GUID);
- unsigned char sha1_result[20];
+ unsigned char sha1_result[20]={0};
mbedtls_sha1((const unsigned char *)combined, strlen(combined), sha1_result);
- unsigned char encoded[64];
+ unsigned char encoded[64]={0};
size_t olen;
mbedtls_base64_encode(encoded, sizeof(encoded), &olen, sha1_result, 16);//sizeof(sha1_result));
strncpy(accept_key, (const char *)encoded, olen);
* @return 成功返回0，失败返回-1
*/
int send_handshake(WebSocketClient *client,const char *headers,char *response,int response_len) {
- char key[WEBSOCKET_KEY_LEN];
+ char key[WEBSOCKET_KEY_LEN+1];
printf("send_handshake1");
if(client==NULL|| response==NULL || response_len<=0)
//移植建议： 此处调用 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)
+ ret = tcp_send_and_wait_response(client->tcp_handle,(uint8_t *)request, strlen(request),(uint8_t *)response, response_len,1000);
+ if(ret<0)
{
- printf("tcp_send_data_directly failed: %d", tcp_result);
+ printf("tcp_send_data_directly failed: %d", ret);
ret = -1;
}
else
{
- // 发送成功，尝试接收响应
- int retlen = tcp_recv_data_directly(client->tcp_handle, (uint8_t *)response, response_len);
-
- if(retlen<0)
+ if(ret<0)
{
printf("tcp_recv_data_directly failed");
ret = -1;
}
else
{
- ret = retlen;
+ ret = ret;
+ printf("tcp_recv_data_directly: %s\r\n", response);
+ client->handshake=1;
}
}
extern WebSocketClient * get_ws_client_by_handle(void* handle);
WebSocketClient * client = get_ws_client_by_handle(handle);
if (client != NULL) {
+
printf("连接成功，开始WebSocket握手\r\n");
-
+ #if 1
// 发送WebSocket握手请求
char *buffer = (char *)aic_zalloc(BUFFER_SIZE);
if (buffer != NULL) {
- int bytes_received = send_handshake(client, "", buffer, BUFFER_SIZE);
+ int bytes_received = send_handshake(client, NULL, buffer, BUFFER_SIZE);
if (bytes_received < 0) {
printf("WebSocket握手失败\r\n");
tcp_manager_close(client->tcp_handle);
}
free(buffer);
}
+ #else
+ client->handshake=1;
+ #endif
} else {
printf("未找到对应的WebSocket客户端\r\n");
}
+
} else {
printf("TCP连接失败\r\n");
}
}
void _tcp_on_data_received_cb(tcp_handle_t handle, uint8_t* data, uint16_t datalen)
{
- printf("tcp_on_data_received_cb %d %d\r\n",handle,datalen);
+ printf("tcp_on_data_received_cb 0x%0X %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;
- }
+ // 使用WebSocket解析器处理数据
+ 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);
+ 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("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)
strcpy(client->query, parsed_url.query);
client->port = parsed_url.port;
- /* 初始化WebSocket组包解析器 */
- ws_parser_init(&client->parser);
+ /* 初始化WebSocket解析器 */
+ client->parser = ws_parser_init();
+ if (!client->parser) {
+ printf("Failed to initialize WebSocket parser\n");
+ goto client_ERROR;
+ }
/* 握手将在连接成功回调中发送，这里不再立即发送 */
printf("WebSocket客户端初始化完成，等待TCP连接建立\r\n");
//移植建议： 此处调用 tcp_manager_close 进行关闭, 不必使用 lwip_close
printf("lwip_close_websocket_client %s %s",client->host,client->path);
- 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; // 需要更多数据
+ // 释放WebSocket解析器
+ if (client->parser) {
+ ws_parser_destroy(client->parser);
+ client->parser = NULL;
}
- 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;
+ tcp_manager_close(client->tcp_handle);
+ free(client);
}
- 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;
-}
+// 旧的WebSocket解析器函数已移除，现在使用新的解析器
/**
* @brief 设置WebSocket数据包回调函数
{
if (!client) return;
- client->parser.packet_callback = callback;
+ client->packet_callback = callback;
}
/**
*/
int ws_process_received_data(WebSocketClient *client, const uint8_t *data, size_t len)
{
- if (!client || !data || len == 0) {
+ if (!client || !data || len == 0 || !client->parser) {
return -1;
}
- ws_packet_parser_t* parser = &client->parser;
- const uint8_t* input = data;
- size_t remaining = len;
+ ws_parser_t* parser = client->parser;
+ int processed = 0;
- 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;
- }
+ // 使用新的解析器处理数据
+ int result = ws_parse_stream(parser, data, len);
+
+ if (result > 0) {
+ // 解析完成一个完整的WebSocket帧
+ const ws_header_t* header = ws_get_header(parser);
+ const uint8_t* payload = ws_get_packet(parser);
+
+ if (header && payload) {
+ printf("WebSocket frame received: opcode=0x%02X, fin=%d, payload_len=%llu\n",
+ header->opcode, header->fin, header->payload_len);
- 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;
+ // 根据操作码处理不同类型的消息
+ switch (header->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");
+ // 调用回调函数处理关闭帧
+ if (client->packet_callback) {
+ client->packet_callback((uint8_t*)payload, header->payload_len, (WebSocketMessageType)WS_CLOSE);
+ }
+ break;
+
+ case WS_TEXT:
+ case WS_BINARY:
+ case WS_CONTINUATION:
+ // 调用回调函数处理数据包
+ if (client->packet_callback) {
+ WebSocketMessageType ws_type = (header->opcode == WS_TEXT) ? (WebSocketMessageType)WS_TEXT :
+ (header->opcode == WS_BINARY) ? (WebSocketMessageType)WS_BINARY : (WebSocketMessageType)WS_CONT;
+ client->packet_callback((uint8_t*)payload, header->payload_len, ws_type);
+ }
+ break;
+
+ default:
+ printf("Unknown WebSocket opcode: 0x%02X\n", header->opcode);
+ break;
}
}
+
+ processed = result;
+ } else if (result == 0) {
+ // 数据不完整，继续等待
+ processed = len;
+ } else {
+ // 解析错误
+ printf("WebSocket parsing error\n");
+ ws_parser_reset(parser);
+ processed = -1;
}
- return len - remaining; // 返回实际处理的字节数
+ return processed;
}
+// 多余的函数已移除，现在只使用一个统一的解析器
#ifdef __cplusplus
}
--- /dev/null
+#include <stdio.h>
+#include <string.h>
+#include "websocket_parser_wrapper.h"
+
+// 测试WebSocket解析器
+void test_websocket_parser() {
+ printf("Testing WebSocket Parser...\n");
+
+ // 初始化解析器
+ ws_parser_t* parser = ws_parser_init();
+ if (!parser) {
+ printf("Failed to initialize parser\n");
+ return;
+ }
+
+ // 设置最大负载长度
+ ws_set_max_payload_length(1024);
+ printf("Max payload length set to: %llu\n", ws_get_max_payload_length());
+
+ // 创建一个简单的WebSocket帧（文本消息 "Hello"）
+ uint8_t test_frame[] = {
+ 0x81, 0x85, 0x37, 0xfa, 0x21, 0x3d, 0x7f, 0x9f, 0x4d, 0x51, 0x58
+ // FIN=1, TEXT, MASK=1, LEN=5, MASK_KEY=0x37fa213d, PAYLOAD="Hello" XOR MASK
+ };
+
+ printf("Test frame: ");
+ for (int i = 0; i < sizeof(test_frame); i++) {
+ printf("%02X ", test_frame[i]);
+ }
+ printf("\n");
+
+ // 解析测试帧
+ int result = ws_parse_stream(parser, test_frame, sizeof(test_frame));
+
+ if (result > 0) {
+ printf("Parser success! Processed %d bytes\n", result);
+
+ const ws_header_t* header = ws_get_header(parser);
+ if (header) {
+ printf("Header: FIN=%d, RSV=%d%d%d, Opcode=0x%02X, Mask=%d, PayloadLen=%llu\n",
+ header->fin, header->rsv1, header->rsv2, header->rsv3,
+ header->opcode, header->mask, header->payload_len);
+ }
+
+ const uint8_t* payload = ws_get_packet(parser);
+ if (payload) {
+ printf("Payload: ");
+ for (size_t i = 0; i < header->payload_len; i++) {
+ printf("%c", payload[i]);
+ }
+ printf("\n");
+ }
+ } else if (result == 0) {
+ printf("Parser needs more data\n");
+ } else {
+ printf("Parser error: %d\n", result);
+ }
+
+ // 清理
+ ws_parser_destroy(parser);
+ printf("Test completed\n");
+}
+
+// 测试控制帧
+void test_control_frame() {
+ printf("\nTesting Control Frame...\n");
+
+ ws_parser_t* parser = ws_parser_init();
+ if (!parser) {
+ printf("Failed to initialize parser\n");
+ return;
+ }
+
+ // 创建一个PING帧
+ uint8_t ping_frame[] = {0x89, 0x00}; // FIN=1, PING, MASK=0, LEN=0
+
+ printf("Ping frame: ");
+ for (int i = 0; i < sizeof(ping_frame); i++) {
+ printf("%02X ", ping_frame[i]);
+ }
+ printf("\n");
+
+ int result = ws_parse_stream(parser, ping_frame, sizeof(ping_frame));
+
+ if (result > 0) {
+ printf("Ping frame parsed successfully!\n");
+ const ws_header_t* header = ws_get_header(parser);
+ if (header) {
+ printf("Ping: Opcode=0x%02X, PayloadLen=%llu\n", header->opcode, header->payload_len);
+ }
+ } else {
+ printf("Failed to parse ping frame: %d\n", result);
+ }
+
+ ws_parser_destroy(parser);
+}
+
+// 测试错误处理
+void test_error_handling() {
+ printf("\nTesting Error Handling...\n");
+
+ ws_parser_t* parser = ws_parser_init();
+ if (!parser) {
+ printf("Failed to initialize parser\n");
+ return;
+ }
+
+ // 创建一个无效的帧（RSV位不为0）
+ uint8_t invalid_frame[] = {0xF1, 0x00}; // RSV位不为0
+
+ printf("Invalid frame: ");
+ for (int i = 0; i < sizeof(invalid_frame); i++) {
+ printf("%02X ", invalid_frame[i]);
+ }
+ printf("\n");
+
+ int result = ws_parse_stream(parser, invalid_frame, sizeof(invalid_frame));
+
+ if (result == -1) {
+ printf("Error handling working correctly!\n");
+ if (ws_parser_has_error(parser)) {
+ printf("Parser is in error state\n");
+ }
+ } else {
+ printf("Error handling failed: %d\n", result);
+ }
+
+ ws_parser_destroy(parser);
+}
+
+int main() {
+ printf("WebSocket Parser Test Program\n");
+ printf("=============================\n\n");
+
+ test_websocket_parser();
+ test_control_frame();
+ test_error_handling();
+
+ return 0;
+}
\ No newline at end of file
--- /dev/null
+#include <stdio.h>
+#include <stdlib.h>
+#include <stdint.h>
+#include <string.h>
+#include "wearable_log.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/*
+ * WebSocket协议解析器 (RFC 6455)
+ *
+ * WebSocket单帧大小限制说明：
+ *
+ * 1. 协议层面限制：
+ * - 16位长度字段：最大 65,535 字节 (2^16 - 1)
+ * - 64位长度字段：最大 18,446,744,073,709,551,615 字节 (2^64 - 1)
+ *
+ * 2. 控制帧限制（RFC 6455规定）：
+ * - Ping/Pong帧：最大125字节负载
+ * - Close帧：最大125字节负载
+ *
+ * 3. 应用层限制：
+ * - 默认最大负载长度：65,535字节（16位）
+ * - 可配置最大负载长度：最大1MB
+ * - 可通过 ws_set_max_payload_length() 函数动态配置
+ *
+ * 4. 实际限制因素：
+ * - 内存限制
+ * - 网络MTU限制
+ * - 应用层缓冲区大小
+ * - 浏览器/客户端实现限制
+ */
+
+// WebSocket帧类型枚举
+typedef enum {
+ WS_CONTINUATION = 0x0,
+ WS_TEXT = 0x1,
+ WS_BINARY = 0x2,
+ WS_CLOSE = 0x8,
+ WS_PING = 0x9,
+ WS_PONG = 0xA
+} ws_opcode_t;
+
+// WebSocket解析状态
+typedef enum {
+ WS_STATE_START,
+ WS_STATE_READ_HEADER1,
+ WS_STATE_READ_HEADER2,
+ WS_STATE_READ_PAYLOAD_LEN,
+ WS_STATE_READ_MASK,
+ WS_STATE_READ_PAYLOAD,
+ WS_STATE_COMPLETE,
+ WS_STATE_ERROR
+} ws_parse_state_t;
+
+// WebSocket帧头部信息
+typedef struct {
+ uint8_t fin; // 帧结束标志
+ uint8_t rsv1; // 保留位1
+ uint8_t rsv2; // 保留位2
+ uint8_t rsv3; // 保留位3
+ ws_opcode_t opcode; // 操作码
+ uint8_t mask; // 掩码标志
+ uint64_t payload_len; // 负载长度
+ uint8_t mask_key[4]; // 掩码密钥
+} ws_header_t;
+
+// WebSocket解析上下文
+typedef struct {
+ ws_parse_state_t state; // 当前解析状态
+ ws_header_t header; // 解析出的头部信息
+ uint8_t* buffer; // 数据缓存
+ size_t buffer_pos; // 缓存当前位置
+ size_t payload_remaining; // 剩余需要读取的负载长度
+ size_t mask_pos; // 掩码读取位置
+} ws_parser_t;
+
+// 全局动态分配缓存（1024字节）
+uint8_t* g_ws_buffer = NULL;
+#define WS_BUFFER_SIZE 1024
+
+// WebSocket协议长度限制配置
+#define WS_MAX_PAYLOAD_LEN_DEFAULT 65535 // 默认最大负载长度（16位）
+#define WS_MAX_PAYLOAD_LEN_MAX 1048576 // 最大允许的负载长度（1MB）
+#define WS_CONTROL_FRAME_MAX_LEN 125 // 控制帧最大负载长度（RFC 6455规定）
+
+static int g_parser_count = 0; // 解析器实例计数
+static uint64_t g_max_payload_len = WS_MAX_PAYLOAD_LEN_DEFAULT; // 全局最大负载长度配置
+
+// 初始化WebSocket解析器
+ws_parser_t* ws_parser_init() {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Initializing WebSocket parser");
+
+ ws_parser_t* parser = (ws_parser_t*)malloc(sizeof(ws_parser_t));
+ if (!parser) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WS: Failed to allocate memory for parser");
+ return NULL;
+ }
+
+ // 初始化全局缓存
+ if (!g_ws_buffer) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Allocating global buffer (%d bytes)", WS_BUFFER_SIZE);
+ g_ws_buffer = (uint8_t*)malloc(WS_BUFFER_SIZE);
+ if (!g_ws_buffer) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WS: Failed to allocate global buffer");
+ free(parser);
+ return NULL;
+ }
+ }
+
+ g_parser_count++;
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Parser count: %d", g_parser_count);
+
+ // 初始化解析器状态
+ parser->state = WS_STATE_START;
+ parser->buffer = g_ws_buffer;
+ parser->buffer_pos = 0;
+ parser->payload_remaining = 0;
+ parser->mask_pos = 0;
+ memset(&parser->header, 0, sizeof(ws_header_t));
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Parser initialized successfully");
+ return parser;
+}
+
+// 重置解析器状态
+void ws_parser_reset(ws_parser_t* parser) {
+ if (!parser) return;
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Resetting parser state");
+
+ parser->state = WS_STATE_START;
+ parser->buffer_pos = 0;
+ parser->payload_remaining = 0;
+ parser->mask_pos = 0;
+ memset(&parser->header, 0, sizeof(ws_header_t));
+}
+
+// 应用掩码解码
+void apply_mask(uint8_t* data, size_t len, const uint8_t* mask_key, size_t offset) {
+ for (size_t i = 0; i < len; i++) {
+ data[i] ^= mask_key[(offset + i) % 4];
+ }
+}
+
+// 验证控制帧长度
+int validate_control_frame_length(ws_opcode_t opcode, uint64_t payload_len) {
+ if (opcode >= WS_CLOSE && opcode <= WS_PONG) {
+ if (payload_len > WS_CONTROL_FRAME_MAX_LEN) {
+ return 0; // 控制帧长度不能超过125字节
+ }
+ }
+ return 1;
+}
+
+// 配置最大负载长度
+int ws_set_max_payload_length(uint64_t max_len) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Setting max payload length to %llu bytes", (unsigned long long)max_len);
+
+ if (max_len > WS_MAX_PAYLOAD_LEN_MAX) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WS: Max payload length %llu exceeds limit %llu",
+ (unsigned long long)max_len, (unsigned long long)WS_MAX_PAYLOAD_LEN_MAX);
+ return 0; // 超过最大允许值
+ }
+ if (max_len < 1) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WS: Invalid max payload length: %llu (must be >= 1)", (unsigned long long)max_len);
+ return 0; // 不能小于1字节
+ }
+ g_max_payload_len = max_len;
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Max payload length set successfully");
+ return 1;
+}
+
+// 获取当前配置的最大负载长度
+uint64_t ws_get_max_payload_length(void) {
+ return g_max_payload_len;
+}
+
+// 解析单个字节
+// 返回值：0-继续解析，1-解析完成一个帧，-1-错误
+int ws_parse_byte(ws_parser_t* parser, uint8_t byte) {
+ if (!parser || parser->state == WS_STATE_ERROR) {
+ return -1;
+ }
+
+ switch (parser->state) {
+ case WS_STATE_START:
+ parser->state = WS_STATE_READ_HEADER1;
+ // 继续执行，处理第一个字节
+ // 注意：这里故意不break，让代码继续执行
+
+ case WS_STATE_READ_HEADER1:
+ // 解析第一个头部字节
+ parser->header.fin = (byte >> 7) & 0x1;
+ parser->header.rsv1 = (byte >> 6) & 0x1;
+ parser->header.rsv2 = (byte >> 5) & 0x1;
+ parser->header.rsv3 = (byte >> 4) & 0x1;
+ parser->header.opcode = (ws_opcode_t)(byte & 0x0F);
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Parse header1 - FIN:%d, RSV:%d%d%d, OPCODE:0x%02X",
+ parser->header.fin, parser->header.rsv1, parser->header.rsv2, parser->header.rsv3, parser->header.opcode);
+
+ // 检查 opcode 是否有效
+ if (!((parser->header.opcode >= WS_CONTINUATION && parser->header.opcode <= WS_BINARY) ||
+ (parser->header.opcode >= WS_CLOSE && parser->header.opcode <= WS_PONG))) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WS: Invalid opcode: 0x%02X", parser->header.opcode);
+ parser->state = WS_STATE_ERROR;
+ return -1;
+ }
+
+ // 检查RSV位是否都为0（根据RFC6455规范）
+ if (parser->header.rsv1 || parser->header.rsv2 || parser->header.rsv3) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WS: Invalid RSV bits: %d%d%d",
+ parser->header.rsv1, parser->header.rsv2, parser->header.rsv3);
+ parser->state = WS_STATE_ERROR;
+ return -1;
+ }
+
+ // 保存第一个头部字节到缓存
+ if (parser->buffer_pos < WS_BUFFER_SIZE) {
+ parser->buffer[parser->buffer_pos++] = byte;
+ } else {
+ parser->state = WS_STATE_ERROR;
+ return -1;
+ }
+
+ parser->state = WS_STATE_READ_HEADER2;
+ break;
+
+ case WS_STATE_READ_HEADER2: {
+ // 解析第二个头部字节
+ parser->header.mask = (byte >> 7) & 0x1;
+ uint8_t payload_len = byte & 0x7F;
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Parse header2 - MASK:%d, LEN:%d",
+ parser->header.mask, payload_len);
+
+ // 保存第二个头部字节到缓存
+ if (parser->buffer_pos < WS_BUFFER_SIZE) {
+ parser->buffer[parser->buffer_pos++] = byte;
+ } else {
+ parser->state = WS_STATE_ERROR;
+ return -1;
+ }
+
+ // 处理负载长度
+ if (payload_len <= 125) {
+ parser->header.payload_len = payload_len;
+ parser->payload_remaining = payload_len;
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Small payload length: %d bytes", payload_len);
+
+ // 验证控制帧长度
+ if (!validate_control_frame_length(parser->header.opcode, payload_len)) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WS: Control frame length validation failed: opcode=0x%02X, len=%d",
+ parser->header.opcode, payload_len);
+ parser->state = WS_STATE_ERROR;
+ return -1;
+ }
+
+ // 如果有掩码，下一步读取掩码
+ if (parser->header.mask) {
+ parser->state = WS_STATE_READ_MASK;
+ } else {
+ // 无掩码，直接读取负载
+ if (parser->payload_remaining == 0) {
+ parser->state = WS_STATE_COMPLETE;
+ return 1;
+ }
+ parser->state = WS_STATE_READ_PAYLOAD;
+ }
+ } else if (payload_len == 126) {
+ // 16位长度，需要再读取2个字节
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Extended 16-bit payload length mode");
+ parser->header.payload_len = 126; // 临时标记
+ parser->payload_remaining = 2;
+ parser->state = WS_STATE_READ_PAYLOAD_LEN;
+ } else if (payload_len == 127) {
+ // 64位长度，需要再读取8个字节
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Extended 64-bit payload length mode");
+ parser->header.payload_len = 127; // 临时标记
+ parser->payload_remaining = 8;
+ parser->state = WS_STATE_READ_PAYLOAD_LEN;
+ }
+ break;
+ }
+
+ case WS_STATE_READ_PAYLOAD_LEN: {
+ // 保存长度字节到缓存
+ if (parser->buffer_pos < WS_BUFFER_SIZE) {
+ parser->buffer[parser->buffer_pos++] = byte;
+ } else {
+ parser->state = WS_STATE_ERROR;
+ return -1;
+ }
+
+ // 累加长度数据
+ static uint8_t len_bytes[8];
+ size_t len_pos = 8 - parser->payload_remaining;
+ if (parser->header.payload_len == 126) {
+ len_bytes[len_pos] = byte;
+ parser->payload_remaining--;
+
+ if (parser->payload_remaining == 0) {
+ // 计算16位长度
+ parser->header.payload_len = (len_bytes[0] << 8) | len_bytes[1];
+ parser->payload_remaining = parser->header.payload_len;
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: 16-bit payload length: %d bytes", parser->header.payload_len);
+
+ // 验证控制帧长度
+ if (!validate_control_frame_length(parser->header.opcode, parser->header.payload_len)) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WS: Control frame length validation failed for 16-bit length");
+ parser->state = WS_STATE_ERROR;
+ return -1;
+ }
+
+ // 检查负载长度是否合理
+ if (parser->header.payload_len > g_max_payload_len) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WS: Payload length %d exceeds maximum %llu",
+ parser->header.payload_len, (unsigned long long)g_max_payload_len);
+ parser->state = WS_STATE_ERROR;
+ return -1;
+ }
+
+ // 检查缓存是否足够
+ if (parser->buffer_pos + parser->payload_remaining > WS_BUFFER_SIZE) {
+ parser->state = WS_STATE_ERROR;
+ return -1;
+ }
+
+ // 处理掩码
+ if (parser->header.mask) {
+ parser->state = WS_STATE_READ_MASK;
+ } else {
+ if (parser->payload_remaining == 0) {
+ parser->state = WS_STATE_COMPLETE;
+ return 1;
+ }
+ parser->state = WS_STATE_READ_PAYLOAD;
+ }
+ }
+ } else if (parser->header.payload_len == 127) {
+ len_bytes[len_pos] = byte;
+ parser->payload_remaining--;
+
+ if (parser->payload_remaining == 0) {
+ // 计算64位长度
+ parser->header.payload_len = 0;
+ for (int i = 0; i < 8; i++) {
+ parser->header.payload_len = (parser->header.payload_len << 8) | len_bytes[i];
+ }
+ parser->payload_remaining = parser->header.payload_len;
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: 64-bit payload length: %llu bytes",
+ (unsigned long long)parser->header.payload_len);
+
+ // 验证控制帧长度
+ if (!validate_control_frame_length(parser->header.opcode, parser->header.payload_len)) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WS: Control frame length validation failed for 64-bit length");
+ parser->state = WS_STATE_ERROR;
+ return -1;
+ }
+
+ // 检查负载长度是否合理
+ if (parser->header.payload_len > g_max_payload_len) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WS: 64-bit payload length %llu exceeds maximum %llu",
+ (unsigned long long)parser->header.payload_len, (unsigned long long)g_max_payload_len);
+ parser->state = WS_STATE_ERROR;
+ return -1;
+ }
+
+ // 检查缓存是否足够
+ if (parser->buffer_pos + parser->payload_remaining > WS_BUFFER_SIZE) {
+ parser->state = WS_STATE_ERROR;
+ return -1;
+ }
+
+ // 处理掩码
+ if (parser->header.mask) {
+ parser->state = WS_STATE_READ_MASK;
+ } else {
+ if (parser->payload_remaining == 0) {
+ parser->state = WS_STATE_COMPLETE;
+ return 1;
+ }
+ parser->state = WS_STATE_READ_PAYLOAD;
+ }
+ }
+ }
+ break;
+ }
+
+ case WS_STATE_READ_MASK:
+ // 读取掩码密钥
+ parser->header.mask_key[parser->mask_pos++] = byte;
+
+ // 保存掩码字节到缓存
+ if (parser->buffer_pos < WS_BUFFER_SIZE) {
+ parser->buffer[parser->buffer_pos++] = byte;
+ } else {
+ parser->state = WS_STATE_ERROR;
+ return -1;
+ }
+
+ // 掩码密钥共4字节
+ if (parser->mask_pos == 4) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Mask key read: %02X %02X %02X %02X",
+ parser->header.mask_key[0], parser->header.mask_key[1],
+ parser->header.mask_key[2], parser->header.mask_key[3]);
+
+ if (parser->payload_remaining == 0) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Frame parsing complete (zero payload)");
+ parser->state = WS_STATE_COMPLETE;
+ return 1;
+ }
+ parser->state = WS_STATE_READ_PAYLOAD;
+ }
+ break;
+
+ case WS_STATE_READ_PAYLOAD: {
+ // 保存负载字节到缓存
+ if (parser->buffer_pos < WS_BUFFER_SIZE) {
+ parser->buffer[parser->buffer_pos++] = byte;
+ } else {
+ parser->state = WS_STATE_ERROR;
+ return -1;
+ }
+
+ parser->payload_remaining--;
+
+ // 所有负载都已读取
+ if (parser->payload_remaining == 0) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Payload read complete (%llu bytes)",
+ (unsigned long long)parser->header.payload_len);
+
+ // 如果有掩码，对负载数据进行解码
+ if (parser->header.mask) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Applying mask to payload data");
+ size_t payload_start = parser->buffer_pos - parser->header.payload_len;
+ apply_mask(parser->buffer + payload_start,
+ parser->header.payload_len,
+ parser->header.mask_key,
+ 0);
+ }
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Frame parsing complete");
+ parser->state = WS_STATE_COMPLETE;
+ return 1;
+ }
+ break;
+ }
+
+ default:
+ break;
+ }
+
+ return 0;
+}
+
+// 处理流数据输入
+// 返回值：解析出的完整包长度，0-未完成，-1-错误
+int ws_parse_stream(ws_parser_t* parser, const uint8_t* data, size_t len) {
+ if (!parser || !data || len == 0) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WS: Invalid parameters for stream parsing");
+ return -1;
+ }
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Parsing stream data (%zu bytes)", len);
+
+ for (size_t i = 0; i < len; i++) {
+ int result = ws_parse_byte(parser, data[i]);
+
+ if (result == 1) {
+ // 解析完成一个包
+ int packet_len = parser->buffer_pos;
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Complete frame parsed (%d bytes total)", packet_len);
+ ws_parser_reset(parser);
+ return packet_len;
+ } else if (result == -1) {
+ // 解析错误，重置解析器
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WS: Parse error at byte %zu, resetting parser", i);
+ ws_parser_reset(parser);
+ return -1;
+ }
+ }
+
+ // 未完成解析
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Frame parsing incomplete, waiting for more data");
+ return 0;
+}
+
+// 释放解析器资源
+void ws_parser_destroy(ws_parser_t* parser) {
+ if (parser) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Destroying WebSocket parser");
+ free(parser);
+ g_parser_count--;
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Parser count after destroy: %d", g_parser_count);
+
+ // 只有当没有解析器实例时才释放全局缓存
+ if (g_parser_count == 0 && g_ws_buffer) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WS: Releasing global buffer (last parser destroyed)");
+ free(g_ws_buffer);
+ g_ws_buffer = NULL;
+ }
+ } else {
+ WEARABLE_LOGW(WEARABLE_LOG_MODULE_APP, "WS: Attempt to destroy NULL parser");
+ }
+}
+
+// 获取解析出的头部信息
+const ws_header_t* ws_get_header(ws_parser_t* parser) {
+ if (!parser) return NULL;
+ return &parser->header;
+}
+
+// 获取解析出的完整数据包
+const uint8_t* ws_get_packet(ws_parser_t* parser) {
+ if (!parser) return NULL;
+ return parser->buffer;
+}
+
+// 新增：检查解析器是否处于错误状态
+int ws_parser_has_error(ws_parser_t* parser) {
+ if (!parser) return 1;
+ return (parser->state == WS_STATE_ERROR);
+}
+
+// 新增：获取当前解析状态
+ws_parse_state_t ws_parser_get_state(ws_parser_t* parser) {
+ if (!parser) return WS_STATE_ERROR;
+ return parser->state;
+}
+
+#ifdef __cplusplus
+}
+#endif
--- /dev/null
+#ifndef __WEBSOCKET_PARSER_WRAPPER_H__
+#define __WEBSOCKET_PARSER_WRAPPER_H__
+
+#include <stdint.h>
+#include <stddef.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+// WebSocket帧类型枚举
+typedef enum {
+ WS_CONTINUATION = 0x0,
+ WS_TEXT = 0x1,
+ WS_BINARY = 0x2,
+ WS_CLOSE = 0x8,
+ WS_PING = 0x9,
+ WS_PONG = 0xA
+} ws_opcode_t;
+
+// WebSocket解析状态
+typedef enum {
+ WS_STATE_START,
+ WS_STATE_READ_HEADER1,
+ WS_STATE_READ_HEADER2,
+ WS_STATE_READ_PAYLOAD_LEN,
+ WS_STATE_READ_MASK,
+ WS_STATE_READ_PAYLOAD,
+ WS_STATE_COMPLETE,
+ WS_STATE_ERROR
+} ws_parse_state_t;
+
+// WebSocket帧头部信息
+typedef struct {
+ uint8_t fin; // 帧结束标志
+ uint8_t rsv1; // 保留位1
+ uint8_t rsv2; // 保留位2
+ uint8_t rsv3; // 保留位3
+ ws_opcode_t opcode; // 操作码
+ uint8_t mask; // 掩码标志
+ uint64_t payload_len; // 负载长度
+ uint8_t mask_key[4]; // 掩码密钥
+} ws_header_t;
+
+// WebSocket解析上下文
+typedef struct {
+ ws_parse_state_t state; // 当前解析状态
+ ws_header_t header; // 解析出的头部信息
+ uint8_t* buffer; // 数据缓存
+ size_t buffer_pos; // 缓存当前位置
+ size_t payload_remaining; // 剩余需要读取的负载长度
+ size_t mask_pos; // 掩码读取位置
+} ws_parser_t;
+
+// 函数声明
+ws_parser_t* ws_parser_init(void);
+void ws_parser_reset(ws_parser_t* parser);
+void ws_parser_destroy(ws_parser_t* parser);
+int ws_parse_byte(ws_parser_t* parser, uint8_t byte);
+int ws_parse_stream(ws_parser_t* parser, const uint8_t* data, size_t len);
+const ws_header_t* ws_get_header(ws_parser_t* parser);
+const uint8_t* ws_get_packet(ws_parser_t* parser);
+int ws_parser_has_error(ws_parser_t* parser);
+ws_parse_state_t ws_parser_get_state(ws_parser_t* parser);
+int ws_set_max_payload_length(uint64_t max_len);
+uint64_t ws_get_max_payload_length(void);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif // __WEBSOCKET_PARSER_WRAPPER_H__
\ No newline at end of file
#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 /* 重试延迟 */
+// #define TCP_CONNECT_RETRY_COUNT 3 /* 连接重试次数 */
+// #define TCP_CONNECT_RETRY_DELAY_MS 1000 /* 重试延迟 */
/* 结构体定义已在头文件中 */
}
if (conn->send_queue_count >= TCP_SEND_QUEUE_SIZE) {
+ printf("发送队列满: handle=%p\n", conn->handle);
return false; /* 队列满 */
}
* WebSocket测试程序
* 基于LiteOS-M系统
* 通过AT^TESTWEBSOCKET命令执行测试
+ * 使用cat1_ws接口实现
*/
#ifdef SUPPORT_WEBSOCKET_TEST
#include "cat1_ctl.h"
#include "custom_runtime.h"
#include "wearable_log.h"
+#include "../tcpmanager/tcp_simple_manager.h"
// 测试配置
#define TEST_DATA_SIZE 228
#define MAX_TEST_COUNT 1000
#define THREAD_STACK_SIZE 4096
#define THREAD_PRIORITY 10
+#define TEST_QUEUE_TIMEOUT_MS 1000
+#define TEST_RECEIVE_TIMEOUT_MS 5000
// 全局变量
static osThreadId_t send_thread_id = NULL;
static osMutexId_t test_mutex = NULL;
static osSemaphoreId_t test_semaphore = NULL;
-static WebSocketClient *test_client = NULL;
+WebSocketClient *test_client = NULL; // 声明为全局可访问，供get_ws_client_by_handle使用
+static ws_message_queue_t test_msg_queue;
static bool test_running = false;
static uint32_t send_count = 0;
static uint32_t recv_count = 0;
} test_state_t;
static test_state_t test_state = TEST_STATE_IDLE;
+static bool tcp_manager_initialized = false;
// 函数声明
static void websocket_send_thread(void *argument);
static bool receive_test_data(uint8_t *data_value);
static void print_test_results(void);
static void check_memory_status(void);
+static void test_packet_callback(uint8_t* data, size_t datalen, WebSocketMessageType type);
+
+// AT命令处理器声明
+int at_test_websocket_handler(const char *cmd, const char *params);
+int at_stop_websocket_handler(const char *cmd, const char *params);
+
+// WebSocket消息队列函数声明
+static int ws_queue_init(ws_message_queue_t* queue);
+static void ws_queue_deinit(ws_message_queue_t* queue);
+static int ws_queue_put(ws_message_queue_t* queue, const uint8_t* data, uint16_t length, WebSocketMessageType type);
+static int ws_queue_get(ws_message_queue_t* queue, uint8_t* data, uint16_t max_length, WebSocketMessageType* type, uint32_t timeout_ms);
+
+/**
+ * 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) {
+ return -1;
+ }
+
+ queue->semaphore = osSemaphoreNew(WS_QUEUE_SIZE, 0, NULL);
+ if (!queue->semaphore) {
+ osMutexDelete(queue->mutex);
+ return -1;
+ }
+
+ queue->initialized = true;
+ return 0;
+}
+
+/**
+ * WebSocket消息队列反初始化
+ */
+static void ws_queue_deinit(ws_message_queue_t* queue)
+{
+ if (!queue || !queue->initialized) return;
+
+ if (queue->semaphore) {
+ osSemaphoreDelete(queue->semaphore);
+ queue->semaphore = NULL;
+ }
+
+ if (queue->mutex) {
+ osMutexDelete(queue->mutex);
+ queue->mutex = NULL;
+ }
+
+ queue->initialized = false;
+}
+
+/**
+ * 向队列中放入消息
+ */
+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) {
+ return -1;
+ }
+
+ if (length > WS_MESSAGE_MAX_SIZE) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Message too large: %d > %d", length, WS_MESSAGE_MAX_SIZE);
+ return -1;
+ }
+
+ if (osMutexAcquire(queue->mutex, osWaitForever) != osOK) {
+ return -1;
+ }
+
+ if (queue->count >= WS_QUEUE_SIZE) {
+ osMutexRelease(queue->mutex);
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Queue 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 = (uint32_t)osKernelGetTickCount();
+ item->in_use = true;
+
+ queue->tail = (queue->tail + 1) % WS_QUEUE_SIZE;
+ queue->count++;
+
+ osMutexRelease(queue->mutex);
+ osSemaphoreRelease(queue->semaphore);
+
+ return 0;
+}
+
+/**
+ * 从队列中获取消息
+ */
+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;
+ }
+
+ if (osSemaphoreAcquire(queue->semaphore, timeout_ms) != osOK) {
+ return 0; // 超时
+ }
+
+ if (osMutexAcquire(queue->mutex, osWaitForever) != osOK) {
+ return -1;
+ }
+
+ if (queue->count == 0) {
+ osMutexRelease(queue->mutex);
+ return 0;
+ }
+
+ ws_message_item_t* item = &queue->messages[queue->head];
+ uint16_t copy_length = (item->length > max_length) ? max_length : item->length;
+
+ memcpy(data, item->data, copy_length);
+ *type = item->type;
+
+ item->in_use = false;
+ queue->head = (queue->head + 1) % WS_QUEUE_SIZE;
+ queue->count--;
+
+ osMutexRelease(queue->mutex);
+
+ return copy_length;
+}
+
+/**
+ * WebSocket数据包回调函数
+ */
+static void test_packet_callback(uint8_t* data, size_t datalen, WebSocketMessageType type)
+{
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "Received WebSocket packet: type=%d, len=%d", type, datalen);
+
+ // 将收到的数据放入队列
+ if (ws_queue_put(&test_msg_queue, data, datalen, type) != 0) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to put message to queue");
+ }
+}
/**
* WebSocket发送线程
*/
static void websocket_send_thread(void *argument)
{
+ (void)argument; // 防止未使用参数警告
uint8_t data_value = 0;
uint32_t send_interval = 100; // 100ms间隔
uint32_t error_count = 0;
*/
static void websocket_recv_thread(void *argument)
{
+ (void)argument; // 防止未使用参数警告
uint8_t data_value = 0;
uint32_t timeout_count = 0;
uint32_t error_count = 0;
WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "连接WebSocket服务器: %s", url);
+ // 初始化TCP简单管理器
+ if (!tcp_manager_initialized) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "初始化TCP简单管理器");
+ tcp_manager_result_t tcp_result = tcp_simple_manager_init();
+ if (tcp_result != TCP_MANAGER_OK) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "TCP管理器初始化失败: %d", tcp_result);
+ return false;
+ }
+ tcp_manager_initialized = true;
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "TCP管理器初始化成功");
+ }
+
+ // 初始化消息队列
+ if (ws_queue_init(&test_msg_queue) != 0) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "消息队列初始化失败");
+ return false;
+ }
+
// 创建WebSocket客户端
test_client = init_websocket_client(url, NULL, 1); // 启用DNS解析
if (test_client == NULL) {
WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WebSocket客户端创建失败");
+ ws_queue_deinit(&test_msg_queue);
+ // 如果WebSocket客户端创建失败，清理TCP管理器
+ if (tcp_manager_initialized) {
+ tcp_simple_manager_deinit();
+ tcp_manager_initialized = false;
+ }
return false;
}
+ // 设置消息队列和回调函数
+ test_client->msg_queue = &test_msg_queue;
+ ws_set_packet_callback(test_client, test_packet_callback);
+
// 创建互斥锁
test_mutex = osMutexNew(NULL);
if (test_mutex == NULL) {
WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "互斥锁创建失败");
lwip_close_websocket_client(test_client);
test_client = NULL;
+ ws_queue_deinit(&test_msg_queue);
+ if (tcp_manager_initialized) {
+ tcp_simple_manager_deinit();
+ tcp_manager_initialized = false;
+ }
return false;
}
osMutexDelete(test_mutex);
lwip_close_websocket_client(test_client);
test_client = NULL;
+ ws_queue_deinit(&test_msg_queue);
+ if (tcp_manager_initialized) {
+ tcp_simple_manager_deinit();
+ tcp_manager_initialized = false;
+ }
return false;
}
test_client = NULL;
}
+ // 清理消息队列
+ ws_queue_deinit(&test_msg_queue);
+
// 删除信号量
if (test_semaphore) {
osSemaphoreDelete(test_semaphore);
test_mutex = NULL;
}
+ // 清理TCP简单管理器
+ if (tcp_manager_initialized) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "清理TCP简单管理器");
+ tcp_simple_manager_deinit();
+ tcp_manager_initialized = false;
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "TCP管理器清理完成");
+ }
+
test_state = TEST_STATE_IDLE;
WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket测试清理完成");
}
return false;
}
- // 验证测试数据缓冲区 - 数组地址总是为true，不需要检查
- // if (!expected_data) {
- // WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "发送数据失败: 测试数据缓冲区无效");
- // return false;
- // }
-
// 准备测试数据
memset(expected_data, data_value, TEST_DATA_SIZE);
}
WebSocketMessageType type;
- // int result = receive_websocket_message(test_client, received_data, TEST_DATA_SIZE, &type);
- int result = 0;
+
+ // 使用消息队列接收数据
+ int result = ws_queue_get(&test_msg_queue, received_data, TEST_DATA_SIZE, &type, TEST_QUEUE_TIMEOUT_MS);
if (result <= 0) {
- WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "接收数据失败: result=%d", result);
- return false;
+ return false; // 超时或错误
}
if (type != WS_BINARY) {
return true;
}
-/**
- * 打印测试结果
- */
/**
* 检查内存状态
*/
if (test_state != TEST_STATE_IDLE) {
WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "测试正在进行中，强制清理并重新开始");
// 强制清理
- test_running = false;
- if (test_client) {
- lwip_close_websocket_client(test_client);
- test_client = NULL;
- }
- test_state = TEST_STATE_IDLE;
+ cleanup_websocket_test();
send_count = 0;
recv_count = 0;
lost_count = 0;
// register_at_command_handler("TESTWEBSOCKET", at_test_websocket_handler);
}
+/**
+ * AT停止测试命令处理函数
+ */
+int at_stop_websocket_handler(const char *cmd, const char *params)
+{
+ if (strncmp(cmd, "AT^STOPWEBSOCKET", 16) != 0) {
+ return -1;
+ }
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "接收到停止WebSocket测试命令");
+
+ if (test_state != TEST_STATE_RUNNING) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket测试当前未运行");
+ return 0;
+ }
+
+ // 停止测试
+ stop_websocket_test();
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket测试已停止");
+ return 0;
+}
+
#ifdef __cplusplus
}
#endif /* __cplusplus */
--- /dev/null
+# WebSocket测试服务器更新说明
+
+## 概述
+
+对 `websocket_test_server.py` 进行了重大增强，添加了连接信息自动发送功能和多种新特性，使其成为一个功能完整的WebSocket测试服务器。
+
+## 🆕 新增功能
+
+### 1. 自动连接信息发送
+
+当客户端连接到服务器时，服务器会自动发送详细的连接信息：
+
+```json
+{
+ "type": "connection_info",
+ "timestamp": "2025-08-12T10:14:19.123456",
+ "server": {
+ "address": "127.0.0.1",
+ "port": 8910,
+ "host": "0.0.0.0",
+ "path": "/",
+ "name": "WebSocket Test Server",
+ "version": "1.0.0"
+ },
+ "client": {
+ "address": "192.168.1.100",
+ "port": 45678,
+ "id": 12345,
+ "user_agent": "Mozilla/5.0...",
+ "origin": "http://localhost"
+ },
+ "connection": {
+ "protocol": "WebSocket",
+ "version": "RFC 6455",
+ "subprotocol": null,
+ "extensions": [],
+ "ping_interval": 30,
+ "ping_timeout": 10
+ },
+ "test": {
+ "data_size": 228,
+ "mode": "echo_test",
+ "description": "Send binary data for echo testing",
+ "commands": {
+ "info": "Get server status",
+ "status": "Get connection statistics"
+ }
+ },
+ "statistics": {
+ "total_clients": 1,
+ "total_received": 0,
+ "total_sent": 0,
+ "lost_packets": 0
+ },
+ "status": "connected",
+ "message": "Welcome! Connected to WebSocket test server at 127.0.0.1:8910"
+}
+```
+
+### 2. WebSocket升级请求模拟
+
+当收到包含 `"upgrade: websocket"` 的文本消息时，服务器返回模拟的HTTP握手响应：
+
+```
+HTTP/1.1 101 Switching Protocols
+Server: nginx/1.18.0 (Ubuntu)
+Date: Tue, 12 Aug 2025 01:53:29 GMT
+Connection: upgrade
+Upgrade: websocket
+Sec-WebSocket-Accept: iuPjBldv/x7mnKvv1KtzX2qDxHY=
+X-Task-Id: f42c892e1db2f1f7a890559de00dea5a
+```
+
+### 3. 状态查询命令
+
+客户端可以发送文本命令获取服务器状态：
+
+- **`info`** 或 **`status`**: 获取服务器状态信息
+ ```
+ WebSocket Test Server Status:
+ - Server: 0.0.0.0:8910
+ - Connected clients: 1
+ - Total received: 150
+ - Total sent: 150
+ - Lost packets: 0
+ - Test data size: 228 bytes
+ ```
+
+### 4. 增强的启动信息
+
+服务器启动时显示详细的配置和功能信息：
+
+```
+==================================================
+WebSocket 测试服务器
+==================================================
+服务器地址: ws://0.0.0.0:8910
+测试数据大小: 228 字节
+Ping间隔: 30 秒
+Ping超时: 10 秒
+功能特性:
+ - 二进制数据回显测试
+ - 文本消息处理
+ - 连接信息自动发送
+ - 实时统计信息
+ - WebSocket升级请求模拟
+==================================================
+🚀 WebSocket服务器启动成功，等待客户端连接...
+```
+
+### 5. 兼容性改进
+
+- 自动检测和适配不同版本的 `websockets` 库
+- 支持新旧版本的导入方式
+- 异常处理的向后兼容性
+
+## 📊 功能特性
+
+### 连接管理
+- ✅ 自动连接信息发送
+- ✅ 多客户端支持
+- ✅ 连接状态监控
+- ✅ 优雅断开处理
+
+### 数据处理
+- ✅ 二进制数据回显测试
+- ✅ 文本消息处理
+- ✅ 数据完整性验证
+- ✅ 统计信息收集
+
+### 协议支持
+- ✅ WebSocket RFC 6455
+- ✅ Ping/Pong心跳
+- ✅ 升级请求模拟
+- ✅ 多种消息类型
+
+### 监控和调试
+- ✅ 实时连接统计
+- ✅ 详细日志记录
+- ✅ 错误处理和报告
+- ✅ 性能指标收集
+
+## 🔧 使用方法
+
+### 基本启动
+```bash
+python3 websocket_test_server.py
+```
+
+### 自定义配置
+```bash
+# 指定地址和端口
+python3 websocket_test_server.py --host 192.168.1.100 --port 9000
+
+# 启用详细日志
+python3 websocket_test_server.py --verbose
+```
+
+### 客户端测试命令
+
+1. **连接测试**: 连接后自动收到连接信息
+2. **状态查询**: 发送 `"info"` 或 `"status"` 文本消息
+3. **升级模拟**: 发送包含 `"upgrade: websocket"` 的消息
+4. **数据测试**: 发送228字节的二进制数据进行回显测试
+
+## 📈 统计信息
+
+服务器实时收集以下统计信息：
+- 总连接数
+- 接收数据包数量
+- 发送数据包数量
+- 丢失数据包数量
+- 丢包率计算
+- 处理速率统计
+
+## 🛠️ 技术改进
+
+### 代码结构
+- 模块化的消息处理
+- 清晰的错误处理
+- 完善的日志系统
+- 类型提示支持
+
+### 性能优化
+- 异步消息处理
+- 连接池管理
+- 内存使用优化
+- 高并发支持
+
+### 安全性
+- 输入验证
+- 错误边界处理
+- 资源清理
+- 异常恢复
+
+## 🔍 调试功能
+
+### 日志级别
+- **INFO**: 常规操作日志
+- **WARNING**: 潜在问题警告
+- **ERROR**: 错误和异常
+
+### 调试命令
+```bash
+# 启用详细日志
+python3 websocket_test_server.py --verbose
+
+# 查看帮助信息
+python3 websocket_test_server.py --help
+```
+
+## 📋 依赖要求
+
+- Python 3.8+
+- websockets 库 (自动安装兼容版本)
+
+```bash
+python3 -m pip install websockets
+```
+
+## 🎯 测试场景
+
+### 1. 基本连接测试
+```javascript
+const ws = new WebSocket('ws://localhost:8910');
+ws.onopen = () => console.log('Connected');
+ws.onmessage = (e) => console.log('Received:', e.data);
+```
+
+### 2. 二进制数据测试
+```javascript
+const data = new Uint8Array(228).fill(42);
+ws.send(data);
+```
+
+### 3. 状态查询测试
+```javascript
+ws.send('info'); // 获取服务器状态
+```
+
+### 4. 升级请求测试
+```javascript
+ws.send('upgrade: websocket'); // 模拟升级请求
+```
+
+## 📝 更新记录
+
+### v1.0.0 (最新版本)
+- ✅ 添加自动连接信息发送
+- ✅ 实现WebSocket升级请求模拟
+- ✅ 增强状态查询功能
+- ✅ 改进启动信息显示
+- ✅ 提升库兼容性
+- ✅ 优化错误处理
+- ✅ 完善文档说明
+
+这个增强版的WebSocket测试服务器现在提供了完整的测试环境，支持各种WebSocket客户端的连接测试和数据验证需求。
\ No newline at end of file
# 导入 websockets
try:
import websockets
- from websockets.asyncio.server import serve
- # 避免导入可能导致兼容性问题的异常类
+ try:
+ # 尝试新版本的导入方式
+ from websockets.asyncio.server import serve
+ except ImportError:
+ # 如果失败，尝试旧版本的导入方式
+ from websockets.server import serve
+
+ # 导入异常类
+ try:
+ from websockets.exceptions import ConnectionClosedOK, ConnectionClosedError
+ except ImportError:
+ # 如果新版本没有这些异常类，定义兼容的替代
+ ConnectionClosedOK = websockets.ConnectionClosed
+ ConnectionClosedError = websockets.ConnectionClosed
+
except ImportError as e:
print(f"错误: 无法导入 websockets 模块: {e}")
print("请确保已安装 websockets: pip install websockets")
"""处理客户端连接"""
client_id = id(websocket)
client_ip, client_port = websocket.remote_address
+ server_ip, server_port = websocket.local_address
logger.info(f"客户端 {client_id} 连接: {client_ip}:{client_port}, 路径: {path}")
self.clients.add(websocket)
+ # 发送连接确认信息给客户端
+ await self.send_connection_info(websocket, client_ip, client_port, server_ip, server_port, path)
+
try:
async for message in websocket:
await self.process_message(websocket, message)
self.clients.discard(websocket)
logger.info(f"客户端 {client_id} 断开连接")
+ async def send_connection_info(self, websocket, client_ip, client_port, server_ip, server_port, path):
+ """发送连接信息给客户端"""
+ import json
+ from datetime import datetime
+
+ # 获取请求头信息
+ headers = getattr(websocket, 'request_headers', {})
+ user_agent = headers.get('User-Agent', 'Unknown')
+ origin = headers.get('Origin', 'Unknown')
+
+ connection_info = {
+ "type": "connection_info",
+ "timestamp": datetime.now().isoformat(),
+ "server": {
+ "address": server_ip,
+ "port": server_port,
+ "host": self.host,
+ "path": path,
+ "name": "WebSocket Test Server",
+ "version": "1.0.0"
+ },
+ "client": {
+ "address": client_ip,
+ "port": client_port,
+ "id": id(websocket),
+ "user_agent": user_agent,
+ "origin": origin
+ },
+ "connection": {
+ "protocol": "WebSocket",
+ "version": "RFC 6455",
+ "subprotocol": getattr(websocket, 'subprotocol', None),
+ "extensions": list(getattr(websocket, 'extensions', [])),
+ "ping_interval": 30,
+ "ping_timeout": 10
+ },
+ "test": {
+ "data_size": TEST_DATA_SIZE,
+ "mode": "echo_test",
+ "description": "Send binary data for echo testing",
+ "commands": {
+ "info": "Get server status",
+ "status": "Get connection statistics"
+ }
+ },
+ "statistics": {
+ "total_clients": len(self.clients),
+ "total_received": self.test_stats['total_received'],
+ "total_sent": self.test_stats['total_sent'],
+ "lost_packets": self.test_stats['lost_packets']
+ },
+ "status": "connected",
+ "message": f"Welcome! Connected to WebSocket test server at {server_ip}:{server_port}"
+ }
+
+ try:
+ # 发送JSON格式的连接信息
+ await websocket.send(json.dumps(connection_info, indent=2))
+ logger.info(f"已向客户端 {client_ip}:{client_port} 发送连接信息")
+ except Exception as e:
+ logger.error(f"发送连接信息失败: {e}")
+
async def process_message(self, websocket, message):
"""处理接收到的消息"""
if isinstance(message, bytes):
self.test_stats['total_received'] += 1
+ # 打印收到的二进制数据（十六进制）
+ try:
+ logger.info(f"收到二进制数据({len(data)} bytes): {data.hex()}")
+ except Exception:
+ logger.info(f"收到二进制数据({len(data)} bytes)")
+
# 回显数据
try:
await websocket.send(data)
self.test_stats['total_sent'] += 1
+ # 打印发回的二进制数据（十六进制）
+ try:
+ logger.info(f"发回二进制数据({len(data)} bytes): {data.hex()}")
+ except Exception:
+ logger.info(f"发回二进制数据({len(data)} bytes)")
+
if self.test_stats['total_received'] % 100 == 0:
logger.info(f"已处理 {self.test_stats['total_received']} 个数据包")
async def start_server(self):
"""启动WebSocket服务器"""
- logger.info(f"启动WebSocket测试服务器: ws://{self.host}:{self.port}")
+ logger.info("=" * 50)
+ logger.info("WebSocket 测试服务器")
+ logger.info("=" * 50)
+ logger.info(f"服务器地址: ws://{self.host}:{self.port}")
+ logger.info(f"测试数据大小: {TEST_DATA_SIZE} 字节")
+ logger.info(f"Ping间隔: 30 秒")
+ logger.info(f"Ping超时: 10 秒")
+ logger.info("功能特性:")
+ logger.info(" - 二进制数据回显测试")
+ logger.info(" - 文本消息处理")
+ logger.info(" - 连接信息自动发送")
+ logger.info(" - 实时统计信息")
+ logger.info(" - WebSocket升级请求模拟")
+ logger.info("=" * 50)
# 启动服务器
server = await serve(
)
self.test_stats['start_time'] = time.time()
- logger.info("WebSocket服务器启动成功")
+ logger.info("WebSocket服务器启动成功，等待客户端连接...")
# 保持服务器运行
await server.wait_closed()
--- /dev/null
+#!/usr/bin/env python3
+"""
+WebSocket测试服务器
+支持多客户端连接，进行数据回显测试
+"""
+
+import asyncio
+import logging
+import argparse
+import time
+from typing import Dict, Set
+import struct
+
+# 检查 SSL 模块
+try:
+ import ssl
+except ImportError:
+ print("警告: SSL 模块不可用，某些功能可能受限")
+ ssl = None
+
+# 导入 websockets
+try:
+ import websockets
+ from websockets.asyncio.server import serve
+ from websockets.exceptions import ConnectionClosedOK, ConnectionClosedError
+ # 避免导入可能导致兼容性问题的异常类
+except ImportError as e:
+ print(f"错误: 无法导入 websockets 模块: {e}")
+ print("请确保已安装 websockets: pip install websockets")
+ exit(1)
+
+# 配置日志
+logging.basicConfig(
+ level=logging.INFO,
+ format='%(asctime)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+# 全局变量
+TEST_DATA_SIZE = 228
+clients: Set = set() # 移除已弃用的类型注解
+test_stats = {
+ 'total_received': 0,
+ 'total_sent': 0,
+ 'lost_packets': 0,
+ 'start_time': None
+}
+
+class WebSocketTestServer:
+ def __init__(self, host: str, port: int):
+ self.host = host
+ self.port = port
+ self.clients = set()
+ self.test_stats = {
+ 'total_received': 0,
+ 'total_sent': 0,
+ 'lost_packets': 0,
+ 'start_time': None
+ }
+
+ async def handle_client(self, websocket, path=None):
+ """处理客户端连接
+ 兼容不同 websockets 版本: 旧版提供 path 参数，新版可从 websocket.path 获取
+ """
+ if path is None:
+ path = getattr(websocket, 'path', '')
+
+ client_id = id(websocket)
+
+ # 安全获取客户端地址信息，避免因 None 或非常规格式导致异常
+ remote_address = getattr(websocket, 'remote_address', None)
+ if isinstance(remote_address, tuple) and len(remote_address) >= 2:
+ client_ip, client_port = remote_address[0], remote_address[1]
+ else:
+ client_ip, client_port = 'unknown', -1
+
+ logger.info(f"客户端 {client_id} 连接: {client_ip}:{client_port}, 路径: {path}")
+
+ self.clients.add(websocket)
+
+ # 连接级别统计
+ conn_stats = {
+ 'total_received': 0,
+ 'total_sent': 0,
+ 'lost_packets': 0,
+ 'start_time': time.time()
+ }
+
+ # 外层 try/finally 确保无论如何关闭，均打印统计
+ try:
+ try:
+ async for message in websocket:
+ await self.process_message(websocket, message, conn_stats)
+ except ConnectionClosedOK:
+ logger.info(f"客户端 {client_id} 连接正常关闭")
+ except ConnectionClosedError as e:
+ logger.error(f"客户端 {client_id} 连接异常关闭: {e}")
+ except asyncio.CancelledError:
+ logger.info(f"客户端 {client_id} 任务被取消（可能是服务器关闭），将打印统计")
+ except Exception as e:
+ logger.error(f"处理客户端 {client_id} 消息时出错: {e}")
+ finally:
+ self.clients.discard(websocket)
+ self.print_connection_stats(client_id, client_ip, client_port, path, conn_stats)
+ logger.info(f"客户端 {client_id} 断开连接")
+
+ async def process_message(self, websocket, message, conn_stats):
+ """处理接收到的消息"""
+ if isinstance(message, bytes):
+ # 二进制消息
+ await self.handle_binary_message(websocket, message, conn_stats)
+ else:
+ # 文本消息
+ await self.handle_text_message(websocket, message)
+
+ async def handle_binary_message(self, websocket, data, conn_stats):
+ """处理二进制消息"""
+ if len(data) != TEST_DATA_SIZE:
+ logger.warning(f"收到错误大小的数据: {len(data)} bytes")
+ return
+
+ # 提取数据值
+ data_value = data[0]
+
+ # 验证数据一致性
+ expected_data = bytes([data_value] * TEST_DATA_SIZE)
+ if data != expected_data:
+ logger.warning(f"数据验证失败: 期望值={data_value}, 实际数据不一致")
+ self.test_stats['lost_packets'] += 1
+ conn_stats['lost_packets'] += 1
+ return
+
+ self.test_stats['total_received'] += 1
+ conn_stats['total_received'] += 1
+
+ # 打印收到的二进制数据（十六进制）
+ try:
+ logger.info(f"收到二进制数据({len(data)} bytes): {data.hex()}")
+ except Exception:
+ logger.info(f"收到二进制数据({len(data)} bytes)")
+
+ # 回显数据
+ try:
+ await websocket.send(data)
+ self.test_stats['total_sent'] += 1
+ conn_stats['total_sent'] += 1
+
+ # 打印发回的二进制数据（十六进制）
+ try:
+ logger.info(f"发回二进制数据({len(data)} bytes): {data.hex()}")
+ except Exception:
+ logger.info(f"发回二进制数据({len(data)} bytes)")
+
+ if self.test_stats['total_received'] % 100 == 0:
+ logger.info(f"已处理 {self.test_stats['total_received']} 个数据包")
+
+ except ConnectionClosedOK:
+ logger.info("发送数据时连接正常关闭")
+ except ConnectionClosedError as e:
+ logger.warning(f"发送数据时连接异常关闭: {e}")
+ except Exception as e:
+ logger.error(f"发送数据失败: {e}")
+
+ async def handle_text_message(self, websocket, message):
+ """处理文本消息"""
+ logger.info(f"收到文本消息: {message}")
+
+ # 发送响应
+ try:
+ response = f"Echo: {message}"
+ await websocket.send(response)
+ except ConnectionClosedOK:
+ logger.info("发送文本响应时连接正常关闭")
+ except ConnectionClosedError as e:
+ logger.warning(f"发送文本响应时连接异常关闭: {e}")
+ except Exception as e:
+ logger.error(f"发送文本响应失败: {e}")
+
+ async def start_server(self):
+ """启动WebSocket服务器"""
+ logger.info(f"启动WebSocket测试服务器: ws://{self.host}:{self.port}")
+
+ # 启动服务器
+ server = await serve(
+ self.handle_client,
+ self.host,
+ self.port,
+ ping_interval=30,
+ ping_timeout=10
+ )
+
+ self.test_stats['start_time'] = time.time()
+ logger.info("WebSocket服务器启动成功")
+
+ # 保持服务器运行
+ await server.wait_closed()
+
+ def print_connection_stats(self, client_id, client_ip, client_port, path, conn_stats):
+ """打印单连接统计信息"""
+ duration = time.time() - conn_stats['start_time'] if conn_stats.get('start_time') else 0.0
+ logger.info("=== 客户端连接传输报告 ===")
+ logger.info(f"客户端: {client_id} ({client_ip}:{client_port}) 路径: {path}")
+ logger.info(f"连接时长: {duration:.2f} 秒")
+ logger.info(f"接收数据包数: {conn_stats['total_received']}")
+ logger.info(f"发送数据包数: {conn_stats['total_sent']}")
+ logger.info(f"丢失数据包数: {conn_stats['lost_packets']}")
+ if conn_stats['total_received'] > 0:
+ loss_rate = (conn_stats['lost_packets'] / conn_stats['total_received']) * 100
+ logger.info(f"丢包率: {loss_rate:.2f}%")
+ if duration > 0:
+ receive_rate = conn_stats['total_received'] / duration
+ send_rate = conn_stats['total_sent'] / duration
+ logger.info(f"接收速率: {receive_rate:.2f} 包/秒")
+ logger.info(f"发送速率: {send_rate:.2f} 包/秒")
+ logger.info("========================")
+
+ def print_stats(self):
+ """打印测试统计信息"""
+ if self.test_stats['start_time'] is None:
+ return
+
+ duration = time.time() - self.test_stats['start_time']
+
+ logger.info("=== WebSocket测试服务器统计 ===")
+ logger.info(f"运行时间: {duration:.2f} 秒")
+ logger.info(f"当前连接数: {len(self.clients)}")
+ logger.info(f"接收数据包数: {self.test_stats['total_received']}")
+ logger.info(f"发送数据包数: {self.test_stats['total_sent']}")
+ logger.info(f"丢失数据包数: {self.test_stats['lost_packets']}")
+
+ if self.test_stats['total_received'] > 0:
+ loss_rate = (self.test_stats['lost_packets'] / self.test_stats['total_received']) * 100
+ logger.info(f"丢包率: {loss_rate:.2f}%")
+
+ if duration > 0:
+ receive_rate = self.test_stats['total_received'] / duration
+ send_rate = self.test_stats['total_sent'] / duration
+ logger.info(f"接收速率: {receive_rate:.2f} 包/秒")
+ logger.info(f"发送速率: {send_rate:.2f} 包/秒")
+
+ logger.info("================================")
+
+async def main():
+ """主函数"""
+ parser = argparse.ArgumentParser(description='WebSocket测试服务器')
+ parser.add_argument('--host', default='0.0.0.0', help='服务器地址 (默认: 0.0.0.0)')
+ parser.add_argument('--port', type=int, default=8910, help='服务器端口 (默认: 8910)')
+ parser.add_argument('--verbose', '-v', action='store_true', help='详细日志输出')
+
+ args = parser.parse_args()
+
+ if args.verbose:
+ logging.getLogger().setLevel(logging.DEBUG)
+
+ # 创建服务器实例
+ server = WebSocketTestServer(args.host, args.port)
+
+ # 启动服务器
+ try:
+ await server.start_server()
+ except KeyboardInterrupt:
+ logger.info("收到中断信号，正在关闭服务器...")
+ # 优雅关闭所有连接
+ for conn in list(server.clients):
+ try:
+ await conn.close()
+ except Exception as e:
+ logger.error(f"关闭连接时出错: {e}")
+ except Exception as e:
+ logger.error(f"服务器运行出错: {e}")
+ finally:
+ server.print_stats()
+
+if __name__ == "__main__":
+ asyncio.run(main())
projects / mirrors / FD230.git / commitdiff