${CMAKE_CURRENT_SOURCE_DIR}/asio.c
${CMAKE_CURRENT_SOURCE_DIR}/slipif.c
${CMAKE_CURRENT_SOURCE_DIR}/uart_slip.c
+ ${CMAKE_CURRENT_SOURCE_DIR}/websocket_test.c
# ${CMAKE_CURRENT_SOURCE_DIR}/udp_server.c
# ${LWIP_PORT_SOURCES}
)
${ROOT_DIR}/open_source/openharmony/third_party/lwip/port
)
set(PUBLIC_DEFINES
+SUPPORT_WEBSOCKET_TEST
)
set(PRIVATE_HEADER
)
#include "app_at_process.h"
#include "cat1_ws.h"
+#include "../websocket_test.h"
+#include "cmsis_os2.h"
static int ConvertCmdToParam(uint8_t *para, uint32_t paraLen,
return 0;
}
+/**
+ * WebSocket测试AT命令处理函数
+ * 格式: AT^TESTWEBSOCKET=<server_addr>,<port1>
+ */
+static uint32_t test_websocket(uint8_t *para, uint32_t paraLen, char *strCommonCmd)
+{
+ UNUSED(strCommonCmd);
+
+ // 将参数转换为字符串
+ char params[256] = {0};
+ if (paraLen > 0 && paraLen < sizeof(params)) {
+ memcpy(params, para, paraLen);
+ params[paraLen] = '\0';
+ }
+
+ printf("WebSocket测试命令: AT^TESTWEBSOCKET=%s\n", params);
+
+ // 检查当前测试状态
+ int current_state = get_websocket_test_state();
+ printf("当前测试状态: %d\n", current_state);
+
+ if (current_state != 1) //TEST_STATE_RUNNING
+ {
+ int result = at_test_websocket_handler("AT^TESTWEBSOCKET", params);
+ if (result == 0) {
+ printf("WebSocket测试启动成功\n");
+ } else {
+ printf("WebSocket测试启动失败，错误码: %d\n", result);
+ }
+ }else{
+ printf("WebSocket测试已经运行\n");
+ }
+
+ return ERRCODE_SUCC;
+}
+
+
+
+/**
+ * WebSocket停止测试AT命令处理函数
+ * 格式: AT^STOPWEBSOCKET
+ */
+static uint32_t stop_websocket_test_cmd(uint8_t *para, uint32_t paraLen, char *strCommonCmd)
+{
+ UNUSED(strCommonCmd);
+ UNUSED(para);
+ UNUSED(paraLen);
+
+ printf("WebSocket停止测试命令: AT^STOPWEBSOCKET\n");
+
+ // 检查当前测试状态
+ int current_state = get_websocket_test_state();
+ printf("停止前测试状态: %d\n", current_state);
+
+ // 直接调用停止函数，减少中间变量
+ stop_websocket_test();
+
+ // 简化状态检查
+ printf("停止后测试状态: %d\n", get_websocket_test_state());
+ printf("WebSocket测试已停止\n");
+ return ERRCODE_SUCC;
+}
+
+/**
+ * WebSocket测试状态查询AT命令处理函数
+ * 格式: AT^WEBSOCKETSTATUS
+ */
+static uint32_t websocket_status(uint8_t *para, uint32_t paraLen, char *strCommonCmd)
+{
+ UNUSED(strCommonCmd);
+ UNUSED(para);
+ UNUSED(paraLen);
+
+ printf("WebSocket状态查询命令: AT^WEBSOCKETSTATUS\n");
+
+ // 获取测试状态
+ int state = get_websocket_test_state();
+ uint32_t send_count, recv_count, lost_count;
+ get_websocket_test_stats(&send_count, &recv_count, &lost_count);
+
+ const char *state_str[] = {"空闲", "运行中", "已完成", "错误"};
+ printf("测试状态: %s\n", state_str[state]);
+ printf("发送数据包: %u\n", send_count);
+ printf("接收数据包: %u\n", recv_count);
+ printf("丢失数据包: %u\n", lost_count);
+
+ return ERRCODE_SUCC;
+}
+
static const at_cmd_table_t g_cat1_cmd[] = {
{ (char *)"AT^SENDTTTS", sendttstxt, (char *)"1" },
{ (char *)"AT^STOPTEST", stopwstest, (char *)"1" },
{ (char *)"AT^OTAHTTP", sendfotaurl, (char *)"1" },
+ { (char *)"AT^TESTWEBSOCKET", test_websocket, (char *)"1" },
+ { (char *)"AT^STOPWEBSOCKET", stop_websocket_test_cmd, (char *)"1" },
+ { (char *)"AT^WEBSOCKETSTATUS", websocket_status, (char *)"1" },
};
--- /dev/null
+/*
+ * WebSocket测试程序
+ * 基于LiteOS-M系统
+ * 通过AT^TESTWEBSOCKET命令执行测试
+ */
+
+#ifdef SUPPORT_WEBSOCKET_TEST
+
+#ifdef __cplusplus
+extern "C" {
+#endif /* __cplusplus */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <stdint.h>
+#include <stdbool.h>
+#include "cmsis_os2.h"
+#include "bk_websocket.h"
+#include "cat1_ctl.h"
+#include "custom_runtime.h"
+#include "wearable_log.h"
+
+// 测试配置
+#define TEST_DATA_SIZE 228
+#define MAX_TEST_COUNT 1000
+#define THREAD_STACK_SIZE 4096
+#define THREAD_PRIORITY 10
+
+// 全局变量
+static osThreadId_t send_thread_id = NULL;
+static osThreadId_t recv_thread_id = NULL;
+static osMutexId_t test_mutex = NULL;
+static osSemaphoreId_t test_semaphore = NULL;
+
+static WebSocketClient *test_client = NULL;
+static bool test_running = false;
+static uint32_t send_count = 0;
+static uint32_t recv_count = 0;
+static uint32_t lost_count = 0;
+static uint8_t expected_data[TEST_DATA_SIZE];
+static uint8_t received_data[TEST_DATA_SIZE];
+
+// 测试状态
+typedef enum {
+ TEST_STATE_IDLE,
+ TEST_STATE_RUNNING,
+ TEST_STATE_COMPLETED,
+ TEST_STATE_ERROR
+} test_state_t;
+
+static test_state_t test_state = TEST_STATE_IDLE;
+
+// 函数声明
+static void websocket_send_thread(void *argument);
+static void websocket_recv_thread(void *argument);
+static bool init_websocket_test(const char *server_addr, uint16_t port);
+static void cleanup_websocket_test(void);
+static bool send_test_data(uint8_t data_value);
+static bool receive_test_data(uint8_t *data_value);
+static void print_test_results(void);
+static void check_memory_status(void);
+
+/**
+ * WebSocket发送线程
+ */
+static void websocket_send_thread(void *argument)
+{
+ uint8_t data_value = 0;
+ uint32_t send_interval = 100; // 100ms间隔
+ uint32_t error_count = 0;
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket发送线程启动");
+
+ while (test_running) {
+ // 检查客户端是否有效
+ if (!test_client || test_client->sockfd < 0) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "发送线程: 客户端无效");
+ break;
+ }
+
+ if (send_test_data(data_value)) {
+ send_count++;
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "发送数据包 %u, 值: %u", send_count, data_value);
+ data_value = (data_value + 1) % 256; // 0-255循环
+ error_count = 0; // 重置错误计数
+ } else {
+ error_count++;
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "发送测试数据失败, 错误计数: %u", error_count);
+
+ // 如果连续失败超过5次，退出循环
+ if (error_count > 5) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "发送线程: 连续失败次数过多，退出");
+ break;
+ }
+ }
+
+ // 检查是否需要停止
+ if (!test_running) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "发送线程: 收到停止信号");
+ break;
+ }
+
+ osDelay(send_interval);
+ }
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket发送线程结束");
+ return;
+}
+
+/**
+ * WebSocket接收线程
+ */
+static void websocket_recv_thread(void *argument)
+{
+ uint8_t data_value = 0;
+ uint32_t timeout_count = 0;
+ uint32_t error_count = 0;
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket接收线程启动");
+
+ while (test_running) {
+ // 检查客户端是否有效
+ if (!test_client || test_client->sockfd < 0) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "接收线程: 客户端无效");
+ break;
+ }
+
+ if (receive_test_data(&data_value)) {
+ recv_count++;
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "接收数据包 %u, 值: %u", recv_count, data_value);
+
+ // 检查数据是否正确
+ bool data_correct = true;
+ for (int i = 0; i < TEST_DATA_SIZE; i++) {
+ if (received_data[i] != data_value) {
+ data_correct = false;
+ break;
+ }
+ }
+
+ if (!data_correct) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "接收数据验证失败");
+ lost_count++;
+ }
+
+ timeout_count = 0;
+ error_count = 0; // 重置错误计数
+ } else {
+ timeout_count++;
+ error_count++;
+ if (timeout_count > 10) { // 1秒超时
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "接收数据超时");
+ break;
+ }
+
+ // 如果连续错误超过20次，退出循环
+ if (error_count > 20) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "接收线程: 连续错误次数过多，退出");
+ break;
+ }
+ }
+
+ // 检查是否需要停止
+ if (!test_running) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "接收线程: 收到停止信号");
+ break;
+ }
+
+ osDelay(100);
+ }
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket接收线程结束");
+ return;
+}
+
+/**
+ * 初始化WebSocket测试
+ */
+static bool init_websocket_test(const char *server_addr, uint16_t port)
+{
+ char url[256];
+ snprintf(url, sizeof(url), "ws://%s:%d/", server_addr, port);
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "连接WebSocket服务器: %s", url);
+
+ // 创建WebSocket客户端
+ test_client = init_websocket_client(url, NULL, 1); // 启用DNS解析
+ if (test_client == NULL) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WebSocket客户端创建失败");
+ return false;
+ }
+
+ // 创建互斥锁
+ test_mutex = osMutexNew(NULL);
+ if (test_mutex == NULL) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "互斥锁创建失败");
+ lwip_close_websocket_client(test_client);
+ test_client = NULL;
+ return false;
+ }
+
+ // 创建信号量
+ test_semaphore = osSemaphoreNew(1, 1, NULL);
+ if (test_semaphore == NULL) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "信号量创建失败");
+ osMutexDelete(test_mutex);
+ lwip_close_websocket_client(test_client);
+ test_client = NULL;
+ return false;
+ }
+
+ // 设置运行标志（在创建线程之前）
+ test_running = true;
+ test_state = TEST_STATE_RUNNING;
+
+ // 创建发送线程
+ osThreadAttr_t send_attr = {
+ .name = "websocket_send",
+ .stack_size = THREAD_STACK_SIZE,
+ .priority = osPriorityNormal,
+ };
+ send_thread_id = osThreadNew(websocket_send_thread, NULL, &send_attr);
+ if (send_thread_id == NULL) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "发送线程创建失败");
+ cleanup_websocket_test();
+ return false;
+ }
+
+ // 创建接收线程
+ osThreadAttr_t recv_attr = {
+ .name = "websocket_recv",
+ .stack_size = THREAD_STACK_SIZE,
+ .priority = osPriorityNormal,
+ };
+ recv_thread_id = osThreadNew(websocket_recv_thread, NULL, &recv_attr);
+ if (recv_thread_id == NULL) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "接收线程创建失败");
+ cleanup_websocket_test();
+ return false;
+ }
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket测试初始化成功");
+ return true;
+}
+
+/**
+ * 清理WebSocket测试
+ */
+static void cleanup_websocket_test(void)
+{
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "开始清理WebSocket测试");
+
+ // 首先停止运行标志
+ test_running = false;
+
+ // 等待一段时间让线程自然结束
+ osDelay(100);
+
+ // 安全地等待线程结束
+ if (send_thread_id) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "等待发送线程结束");
+ osThreadJoin(send_thread_id);
+ send_thread_id = NULL;
+ }
+
+ if (recv_thread_id) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "等待接收线程结束");
+ osThreadJoin(recv_thread_id);
+ recv_thread_id = NULL;
+ }
+
+ // 关闭WebSocket连接 - 在删除线程相关资源之前
+ if (test_client) {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "关闭WebSocket连接");
+ lwip_close_websocket_client(test_client);
+ test_client = NULL;
+ }
+
+ // 删除信号量
+ if (test_semaphore) {
+ osSemaphoreDelete(test_semaphore);
+ test_semaphore = NULL;
+ }
+
+ // 删除互斥锁 - 最后删除
+ if (test_mutex) {
+ osMutexDelete(test_mutex);
+ test_mutex = NULL;
+ }
+
+ test_state = TEST_STATE_IDLE;
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket测试清理完成");
+}
+
+/**
+ * 发送测试数据
+ */
+static bool send_test_data(uint8_t data_value)
+{
+ if (!test_client || !test_running) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "发送数据失败: 客户端或运行状态无效");
+ return false;
+ }
+
+ // 添加额外的安全检查
+ if (test_client->sockfd < 0) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "发送数据失败: 无效的socket");
+ return false;
+ }
+
+ // 验证测试数据缓冲区 - 数组地址总是为true，不需要检查
+ // if (!expected_data) {
+ // WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "发送数据失败: 测试数据缓冲区无效");
+ // return false;
+ // }
+
+ // 准备测试数据
+ memset(expected_data, data_value, TEST_DATA_SIZE);
+
+ // 添加内存检查
+ if (TEST_DATA_SIZE > 65536) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "发送数据失败: 数据大小过大");
+ return false;
+ }
+
+ // 检查内存状态
+ check_memory_status();
+
+ // 发送数据前增加延迟，避免过于频繁的发送
+ osDelay(5);
+
+ // 发送数据
+ int result = send_websocket_message(test_client, expected_data, TEST_DATA_SIZE, WS_BINARY);
+ if (result != 0) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "发送数据失败: %d", result);
+ return false;
+ }
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "发送数据成功: 值=%u, 大小=%d", data_value, TEST_DATA_SIZE);
+ return true;
+}
+
+/**
+ * 接收测试数据
+ */
+static bool receive_test_data(uint8_t *data_value)
+{
+ if (!test_client || !test_running) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "接收数据失败: 客户端或运行状态无效");
+ return false;
+ }
+
+ WebSocketMessageType type;
+ int result = receive_websocket_message(test_client, received_data, TEST_DATA_SIZE, &type);
+
+ if (result <= 0) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "接收数据失败: result=%d", result);
+ return false;
+ }
+
+ if (type != WS_BINARY) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "接收数据类型错误: type=%d", type);
+ return false;
+ }
+
+ // 检查数据大小
+ if (result != TEST_DATA_SIZE) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "接收数据大小错误: %d", result);
+ return false;
+ }
+
+ // 提取数据值
+ *data_value = received_data[0];
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "接收数据成功: 值=%u, 大小=%d", *data_value, result);
+ return true;
+}
+
+/**
+ * 打印测试结果
+ */
+/**
+ * 检查内存状态
+ */
+static void check_memory_status(void)
+{
+ // 检查内存状态
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket: 检查内存状态");
+
+ // 检查内存池状态 - 暂时注释掉，避免编译错误
+ // extern void ws_mem_pool_status(void);
+ // ws_mem_pool_status();
+
+ // 检查系统内存状态 - 暂时注释掉，避免编译错误
+ // extern size_t xPortGetFreeHeapSize(void);
+ // size_t free_heap = xPortGetFreeHeapSize();
+ // WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket: 系统空闲堆内存: %u 字节", (uint32_t)free_heap);
+
+ // 如果内存不足，增加延迟
+ // if (free_heap < 1024) {
+ // WEARABLE_LOGW(WEARABLE_LOG_MODULE_APP, "WebSocket: 系统内存不足，增加延迟");
+ // osDelay(50);
+ // } else {
+ osDelay(10);
+ // }
+}
+
+/**
+ * 打印测试结果
+ */
+static void print_test_results(void)
+{
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "=== WebSocket测试结果 ===");
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "发送数据包数: %u", send_count);
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "接收数据包数: %u", recv_count);
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "丢失数据包数: %u", lost_count);
+
+ if (send_count > 0) {
+ float loss_rate = (float)lost_count / send_count * 100.0f;
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "丢包率: %.2f%%", loss_rate);
+ }
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "========================");
+}
+
+/**
+ * AT命令处理函数
+ */
+int at_test_websocket_handler(const char *cmd, const char *params)
+{
+ if (strncmp(cmd, "AT^TESTWEBSOCKET", 16) != 0) {
+ return -1;
+ }
+
+ // 解析参数: AT^TESTWEBSOCKET=<server_addr>,<port1>
+ char server_addr[64] = {0};
+ uint16_t port1 = 0, port2 = 0;
+
+ if (sscanf(params, "%63[^,],%hu", server_addr, &port1) != 2) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "参数格式错误，正确格式: AT^TESTWEBSOCKET=<server_addr>,<port1>");
+ return -1;
+ }
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "开始WebSocket测试");
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "服务器地址: %s", server_addr);
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "端口1: %u", port1);
+
+ // 检查当前状态
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "当前测试状态: %d", test_state);
+ 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;
+ send_count = 0;
+ recv_count = 0;
+ lost_count = 0;
+ }
+
+ // 初始化测试
+ if (!init_websocket_test(server_addr, port1)) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WebSocket测试初始化失败");
+ return -1;
+ }
+
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket测试已启动，使用AT^STOPWEBSOCKET停止测试");
+ return 0;
+}
+
+/**
+ * 手动启动WebSocket测试
+ */
+int start_websocket_test(const char *server_addr, uint16_t port, uint32_t duration_ms)
+{
+ if (test_state != TEST_STATE_IDLE) {
+ WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "测试正在进行中");
+ return -1;
+ }
+
+ if (!init_websocket_test(server_addr, port)) {
+ return -1;
+ }
+
+ // 如果指定了持续时间，则运行指定时间后自动停止
+ if (duration_ms > 0) {
+ osDelay(duration_ms);
+ cleanup_websocket_test();
+ print_test_results();
+ }
+
+ return 0;
+}
+
+/**
+ * 停止WebSocket测试
+ */
+void stop_websocket_test(void)
+{
+ if (test_state == TEST_STATE_RUNNING) {
+ cleanup_websocket_test();
+ print_test_results();
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket测试已停止");
+ } else {
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket测试当前未运行");
+ }
+}
+
+/**
+ * 获取测试状态
+ */
+int get_websocket_test_state(void)
+{
+ return (int)test_state;
+}
+
+/**
+ * 获取测试统计信息
+ */
+void get_websocket_test_stats(uint32_t *send_count_out, uint32_t *recv_count_out, uint32_t *lost_count_out)
+{
+ if (send_count_out) *send_count_out = send_count;
+ if (recv_count_out) *recv_count_out = recv_count;
+ if (lost_count_out) *lost_count_out = lost_count;
+}
+
+/**
+ * 模块初始化
+ */
+void websocket_test_init(void)
+{
+ WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket测试模块初始化");
+
+ // 注册AT命令处理器
+ // 注意：这里需要根据实际的AT命令系统进行适配
+ // extern int register_at_command_handler(const char *cmd, at_command_handler_t handler);
+ // register_at_command_handler("TESTWEBSOCKET", at_test_websocket_handler);
+}
+
+#ifdef __cplusplus
+}
+#endif /* __cplusplus */
+
+#endif /* SUPPORT_WEBSOCKET_TEST */
\ No newline at end of file
--- /dev/null
+/*
+ * WebSocket测试模块头文件
+ * 基于LiteOS-M系统
+ */
+
+#ifndef __WEBSOCKET_TEST_H
+#define __WEBSOCKET_TEST_H
+
+#include <stdint.h>
+#include <stdbool.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/**
+ * WebSocket测试模块初始化
+ */
+void websocket_test_init(void);
+
+/**
+ * AT命令处理函数
+ * @param cmd AT命令字符串
+ * @param params 命令参数
+ * @return 0表示成功，-1表示失败
+ */
+int at_test_websocket_handler(const char *cmd, const char *params);
+
+/**
+ * 手动启动WebSocket测试
+ * @param server_addr 服务器地址
+ * @param port 服务器端口
+ * @param duration_ms 测试持续时间(毫秒)
+ * @return 0表示成功，-1表示失败
+ */
+int start_websocket_test(const char *server_addr, uint16_t port, uint32_t duration_ms);
+
+/**
+ * 停止WebSocket测试
+ */
+void stop_websocket_test(void);
+
+/**
+ * 获取测试状态
+ * @return 0-空闲, 1-运行中, 2-已完成, 3-错误
+ */
+int get_websocket_test_state(void);
+
+/**
+ * 获取测试统计信息
+ * @param send_count 发送数据包数
+ * @param recv_count 接收数据包数
+ * @param lost_count 丢失数据包数
+ */
+void get_websocket_test_stats(uint32_t *send_count, uint32_t *recv_count, uint32_t *lost_count);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* __WEBSOCKET_TEST_H */
\ No newline at end of file
// lwIP debug options, comment the ones you don't want
#define LWIP_DEBUG 1
#if LWIP_DEBUG
+
+#define DEBUG_TCP 0
+
#define ETHARP_DEBUG LWIP_DBG_OFF
#define NETIF_DEBUG LWIP_DBG_OFF
#define PBUF_DEBUG LWIP_DBG_OFF
#define TCP_ERR_DEBUG LWIP_DBG_OFF
#define TCP_INPUT_DEBUG LWIP_DBG_OFF
#define TCP_FR_DEBUG LWIP_DBG_OFF
-#define TCP_RTO_DEBUG LWIP_DBG_OFF
-#define TCP_CWND_DEBUG LWIP_DBG_OFF
+#if DEBUG_TCP
+#define TCP_RTO_DEBUG LWIP_DBG_ON
+#define TCP_CWND_DEBUG LWIP_DBG_ON
#define TCP_WND_DEBUG LWIP_DBG_OFF
-#define TCP_OUTPUT_DEBUG LWIP_DBG_OFF
+#define TCP_OUTPUT_DEBUG LWIP_DBG_ON
#define TCP_RST_DEBUG LWIP_DBG_OFF
-#define TCP_QLEN_DEBUG LWIP_DBG_OFF
+#define TCP_QLEN_DEBUG LWIP_DBG_ON
#define TCP_SACK_DEBUG LWIP_DBG_OFF
#define TCP_TLP_DEBUG LWIP_DBG_OFF
#define UDP_DEBUG LWIP_DBG_OFF
#define ALTCP_MBEDTLS_MEM_DEBUG LWIP_DBG_ON
#define ALTCP_MBEDTLS_DEBUG LWIP_DBG_ON
#define ALTCP_MBEDTLS_LIB_DEBUG LWIP_DBG_ON
+#else
+#define TCP_RTO_DEBUG LWIP_DBG_OFF
+#define TCP_CWND_DEBUG LWIP_DBG_OFF
+#define TCP_WND_DEBUG LWIP_DBG_OFF
+#define TCP_OUTPUT_DEBUG LWIP_DBG_OFF
+#define TCP_RST_DEBUG LWIP_DBG_OFF
+#define TCP_QLEN_DEBUG LWIP_DBG_OFF
+#define TCP_SACK_DEBUG LWIP_DBG_OFF
+#define TCP_TLP_DEBUG LWIP_DBG_OFF
+#define UDP_DEBUG LWIP_DBG_OFF
+#define TCPIP_DEBUG LWIP_DBG_OFF
+#define SLIP_DEBUG LWIP_DBG_OFF
+#define DHCP_DEBUG LWIP_DBG_OFF
+#define AUTOIP_DEBUG LWIP_DBG_OFF
+#define DNS_DEBUG LWIP_DBG_OFF
+#define TFTP_DEBUG LWIP_DBG_OFF
+#define SYS_ARCH_DEBUG LWIP_DBG_OFF
+#define SNTP_DEBUG LWIP_DBG_OFF
+#define IP6_DEBUG LWIP_DBG_OFF
+#define DHCP6_DEBUG LWIP_DBG_OFF
+#define DRV_STS_DEBUG LWIP_DBG_OFF
+#define HTTPC_DEBUG LWIP_DBG_OFF
+
+#define ALTCP_MBEDTLS_MEM_DEBUG LWIP_DBG_OFF
+#define ALTCP_MBEDTLS_DEBUG LWIP_DBG_OFF
+#define ALTCP_MBEDTLS_LIB_DEBUG LWIP_DBG_OFF
+#endif
#endif
//多线程安全支持
#define LWIP_NETCONN_SEM_PER_THREAD 1
+
+
+// Options in old lwipopts.h but kept in Defaults with new opt.h
+#define IP_FORWARD 0
+#define LWIP_DBG_TYPES_ON LWIP_DBG_ON
+#define LWIP_ICMP 1
+#define LWIP_NETCONN 1
+#define LWIP_SOCKET 1
+#define LWIP_STATS 1
+#define LWIP_TCP 1
+#define LWIP_UDP 1
+#define NO_SYS 0
+
+#define LWIP_ALTCP 1
+#define LWIP_ALTCP_TLS 1
+#define LWIP_ALTCP_TLS_MBEDTLS 1
+
// Options for old lwipopts.h
#define MEM_SIZE (512*1024) //(4*1024*1024) // (512*1024)
#define IP_FRAG_MAX_MTU 1500
#define LWIP_NETIF_HOSTNAME 1
#define LWIP_NETIF_TX_SINGLE_PBUF 1
-#define LWIP_NETCONN_FULLDUPLEX 1 // Caution
+#define LWIP_NETCONN_FULLDUPLEX 1 // 确保全双工支持
#define LWIP_POSIX_SOCKETS_IO_NAMES 0
#define LWIP_TCP_KEEPALIVE 1
#define RECV_BUFSIZE_DEFAULT 65535
#define LWIP_SO_SNDTIMEO 1
#define LWIP_STATS_DISPLAY 1
#define MEM_LIBC_MALLOC 1
-#define MEMP_NUM_ARP_QUEUE (65535 * LWIP_CONFIG_NUM_SOCKETS / (IP_FRAG_MAX_MTU - 20 - 8))
-#define MEMP_NUM_NETBUF (65535 * 3 * LWIP_CONFIG_NUM_SOCKETS / (IP_FRAG_MAX_MTU - 20 - 8))
+#define MEMP_NUM_ARP_QUEUE (65535UL * LWIP_CONFIG_NUM_SOCKETS / (IP_FRAG_MAX_MTU - 20 - 8))
+#define MEMP_NUM_NETBUF (65535UL * 3 * LWIP_CONFIG_NUM_SOCKETS / (IP_FRAG_MAX_MTU - 20 - 8))
#define MEMP_NUM_NETCONN LWIP_CONFIG_NUM_SOCKETS
#define MEMP_NUM_PBUF LWIP_CONFIG_NUM_SOCKETS*2
#define MEMP_NUM_RAW_PCB LWIP_CONFIG_NUM_SOCKETS
-#define MEMP_NUM_REASSDATA (IP_REASS_MAX_MEM_SIZE / 65535)
-#define MEMP_NUM_TCPIP_MSG_API 64
-#define MEMP_NUM_TCPIP_MSG_INPKT 512
+#define MEMP_NUM_REASSDATA (IP_REASS_MAX_MEM_SIZE / 65535UL)
+#define MEMP_NUM_TCPIP_MSG_API 64
+#define MEMP_NUM_TCPIP_MSG_INPKT 512
#define MEMP_NUM_TCP_PCB LWIP_CONFIG_NUM_SOCKETS
#define MEMP_NUM_TCP_PCB_LISTEN LWIP_CONFIG_NUM_SOCKETS
#define MEMP_NUM_UDP_PCB LWIP_CONFIG_NUM_SOCKETS
#define TCPIP_MBOX_SIZE 512
#define TCPIP_THREAD_PRIO 25 //osPriorityBelowNormal 16
#define TCPIP_THREAD_STACKSIZE 0x2000
-#define TCP_MAXRTX 12 //max 12
-#define LWIP_TCP_RTO_TIME 1500
+#define TCP_MAXRTX 6 // 从8减少到6，减少重传次数
+#define LWIP_TCP_RTO_TIME 1500 // 从1000减少到800，减少重传超时时间
+/* 优化TCP重传参数，减少AI音频传输延迟 */
+#define TCP_SYNMAXRTX 6 // 从3减少到2，减少SYN重传次数
+#define TCP_QUEUE_OOSEQ LWIP_TCP
#define TCP_MSS (IP_FRAG_MAX_MTU - 20 - 20)
-#define TCP_SND_BUF (10* TCP_MSS)
-#define TCP_SND_QUEUELEN (2*(TCP_SND_BUF/TCP_MSS))
-#define TCP_SNDLOWAT LWIP_MIN(LWIP_MAX(((TCP_SND_BUF)/2), (2 * TCP_MSS) + 1), (TCP_SND_BUF) - 1)
-#define TCP_SNDQUEUELOWAT LWIP_MAX(((TCP_SND_QUEUELEN)/2), 5)
+/* 针对AI音频传输优化TCP发送缓冲区大小 */
+#define TCP_SND_BUF (10* TCP_MSS) // 增加到16*，约11680字节，提高发送缓冲区
+/* 严格控制TCP发送队列长度，防止积压 */
+#define TCP_SND_QUEUELEN (2*(TCP_SND_BUF/TCP_MSS)) // 增加到8*，提高队列容量
+#define TCP_SNDLOWAT LWIP_MIN(LWIP_MAX(((TCP_SND_BUF)/2), (2 * TCP_MSS) + 1), (TCP_SND_BUF) - 1) /* 降低低水位阈值 */
+#define TCP_SNDQUEUELOWAT LWIP_MAX(((TCP_SND_QUEUELEN)/2), 5) /* 进一步降低低水位阈值 */
#define TCP_TTL 255
-#define TCP_WND (TCP_SND_BUF/2)
+/* 优化TCP窗口大小，匹配发送缓冲区 */
+#define TCP_WND (TCP_SND_BUF/2) // 保持SND_BUF/2，约5840字节，提高接收窗口
#define UDP_TTL 255
-#define MEMP_NUM_TCP_SEG (((TCP_SND_BUF * 3 / 2) + TCP_WND) * LWIP_CONFIG_NUM_SOCKETS / TCP_MSS)
+// 增加TCP段内存池大小
+#define MEMP_NUM_TCP_SEG (LWIP_CONFIG_NUM_SOCKETS * 16) // 从8增加到16，提高TCP段处理能力
+
-// Options in old lwipopts.h but kept in Defaults with new opt.h
-#define IP_FORWARD 0
-#define LWIP_DBG_TYPES_ON LWIP_DBG_ON
-#define LWIP_ICMP 1
-#define LWIP_NETCONN 1
-#define LWIP_SOCKET 1
-#define LWIP_STATS 1
-#define LWIP_TCP 1
-#define LWIP_UDP 1
-#define NO_SYS 0
-#define TCP_QUEUE_OOSEQ LWIP_TCP
-#define LWIP_ALTCP 1
-#define LWIP_ALTCP_TLS 1
-#define LWIP_ALTCP_TLS_MBEDTLS 1
#if defined LWIP_STATS_DISPLAY && !LWIP_STATS
// Options for enhancement code, same for old lwipopts.h
#define LWIP_NETIF_PROMISC 0
+// 修复后
+#ifdef LOSCFG_NET_LWIP_SACK_TFTP
#define LWIP_TFTP LOSCFG_NET_LWIP_SACK_TFTP
+#else
+#define LWIP_TFTP 0
+#endif
#define LWIP_DHCPS 1
#define LWIP_ENABLE_NET_CAPABILITY 1
#define LWIP_ENABLE_CAP_NET_BROADCAST 0
#define MEMP_MEM_MALLOC 1
-
-// #undef LWIP_SOCKET_SELECT
-// #define LWIP_SOCKET_SELECT 1
-
-
/* MIB2 stats are required to check IPv4 reassembly results */
#define MIB2_STATS 1
#define SLIP_RX_FROM_ISR 1
--- /dev/null
+# WebSocket AT命令测试脚本
+# 使用方法：在串口终端中依次输入以下命令
+
+# 1. 启动WebSocket测试
+AT^TESTWEBSOCKET=snicker.com.cn,8910
+
+# 2. 查询测试状态
+AT^WEBSOCKETSTATUS
+
+# 3. 停止WebSocket测试
+AT^STOPWEBSOCKET
+
+# 4. 再次查询状态确认已停止
+AT^WEBSOCKETSTATUS
+
+# 预期输出：
+# 启动命令：WebSocket测试已启动，使用AT^STOPWEBSOCKET停止测试
+# 状态查询：显示当前测试状态和统计数据
+# 停止命令：WebSocket测试已停止，并显示最终测试结果
\ No newline at end of file
--- /dev/null
+#!/usr/bin/env python3
+"""
+WebSocket客户端测试程序（原生版本）
+使用Python标准库socket实现，不依赖websockets库
+"""
+
+import socket
+import ssl
+import base64
+import hashlib
+import struct
+import logging
+import argparse
+import time
+import threading
+import sys
+import random
+
+# 配置日志
+logging.basicConfig(
+ level=logging.INFO,
+ format='%(asctime)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+# 测试配置
+TEST_DATA_SIZE = 228
+SEND_INTERVAL = 0.1 # 100ms间隔
+
+class WebSocketClient:
+ """原生WebSocket客户端实现"""
+
+ def __init__(self, host, port):
+ self.host = host
+ self.port = port
+ self.socket = None
+ self.connected = False
+ self.last_pong_time = None
+
+ def connect(self):
+ """连接到WebSocket服务器"""
+ try:
+ # 创建TCP连接
+ self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ self.socket.connect((self.host, self.port))
+
+ # 生成WebSocket握手密钥
+ key = base64.b64encode(bytes(random.getrandbits(8) for _ in range(16))).decode()
+
+ # 发送WebSocket握手请求
+ handshake = (
+ f"GET / HTTP/1.1\r\n"
+ f"Host: {self.host}:{self.port}\r\n"
+ f"Upgrade: websocket\r\n"
+ f"Connection: Upgrade\r\n"
+ f"Sec-WebSocket-Key: {key}\r\n"
+ f"Sec-WebSocket-Version: 13\r\n"
+ f"\r\n"
+ )
+
+ self.socket.send(handshake.encode())
+
+ # 接收握手响应
+ response = self.socket.recv(1024).decode()
+
+ if "101 Switching Protocols" in response:
+ logger.info("WebSocket握手成功")
+ self.connected = True
+ return True
+ else:
+ logger.error(f"WebSocket握手失败: {response}")
+ return False
+
+ except Exception as e:
+ logger.error(f"连接失败: {e}")
+ return False
+
+ def disconnect(self):
+ """断开连接"""
+ if self.socket:
+ try:
+ # 发送关闭帧
+ close_frame = struct.pack('!BB', 0x88, 0x00)
+ self.socket.send(close_frame)
+ self.socket.close()
+ except:
+ pass
+ self.socket = None
+ self.connected = False
+ logger.info("WebSocket连接已断开")
+
+ def send_frame(self, data, opcode=0x02): # 0x02 = binary frame
+ """发送WebSocket帧"""
+ if not self.connected:
+ return False
+
+ try:
+ length = len(data)
+ if length < 126:
+ frame = struct.pack('!BB', 0x80 | opcode, 0x80 | length)
+ elif length < 65536:
+ frame = struct.pack('!BBH', 0x80 | opcode, 0x80 | 126, length)
+ else:
+ frame = struct.pack('!BBQ', 0x80 | opcode, 0x80 | 127, length)
+
+ # 添加掩码
+ mask = bytes(random.getrandbits(8) for _ in range(4))
+ frame += mask
+
+ # 掩码数据
+ masked_data = bytes(data[i] ^ mask[i % 4] for i in range(len(data)))
+ frame += masked_data
+
+ self.socket.send(frame)
+ return True
+ except Exception as e:
+ logger.error(f"发送帧失败: {e}")
+ return False
+
+ def send_ping(self):
+ """发送ping帧"""
+ try:
+ # ping帧: FIN=1, RSV=0, Opcode=9, Mask=1, Payload Length=0
+ ping_frame = struct.pack('!BB', 0x89, 0x80) # 0x89 = 10001001 (FIN=1, Opcode=9)
+
+ # 添加掩码
+ mask = bytes(random.getrandbits(8) for _ in range(4))
+ ping_frame += mask
+
+ self.socket.send(ping_frame)
+ logger.debug("发送ping帧")
+ return True
+ except Exception as e:
+ logger.error(f"发送ping帧失败: {e}")
+ return False
+
+ def send_pong(self, payload=b''):
+ """发送pong帧"""
+ try:
+ length = len(payload)
+ if length == 0:
+ pong_frame = struct.pack('!BB', 0x8A, 0x80) # 0x8A = 10001010 (FIN=1, Opcode=10)
+ else:
+ pong_frame = struct.pack('!BB', 0x8A, 0x80 | length)
+
+ # 添加掩码
+ mask = bytes(random.getrandbits(8) for _ in range(4))
+ pong_frame += mask
+
+ # 掩码数据
+ masked_data = bytes(payload[i] ^ mask[i % 4] for i in range(len(payload)))
+ pong_frame += masked_data
+
+ self.socket.send(pong_frame)
+ logger.debug("发送pong帧")
+ return True
+ except Exception as e:
+ logger.error(f"发送pong帧失败: {e}")
+ return False
+
+ def receive_frame(self):
+ """接收WebSocket帧"""
+ if not self.connected:
+ return None
+
+ try:
+ # 读取帧头
+ header = self.socket.recv(2)
+ if len(header) < 2:
+ return None
+
+ fin = (header[0] & 0x80) != 0
+ opcode = header[0] & 0x0F
+ masked = (header[1] & 0x80) != 0
+ payload_length = header[1] & 0x7F
+
+ # 读取扩展长度
+ if payload_length == 126:
+ length_bytes = self.socket.recv(2)
+ payload_length = struct.unpack('!H', length_bytes)[0]
+ elif payload_length == 127:
+ length_bytes = self.socket.recv(8)
+ payload_length = struct.unpack('!Q', length_bytes)[0]
+
+ # 读取掩码
+ mask = None
+ if masked:
+ mask = self.socket.recv(4)
+
+ # 读取数据
+ data = b''
+ remaining = payload_length
+ while remaining > 0:
+ chunk = self.socket.recv(min(remaining, 4096))
+ if not chunk:
+ break
+ data += chunk
+ remaining -= len(chunk)
+
+ # 解掩码
+ if masked and mask:
+ data = bytes(data[i] ^ mask[i % 4] for i in range(len(data)))
+
+ # 处理不同类型的帧
+ if opcode == 0x09: # ping帧
+ logger.debug("收到ping帧，发送pong响应")
+ self.send_pong(data) # 将ping的payload作为pong的payload返回
+ return None
+ elif opcode == 0x0A: # pong帧
+ logger.debug("收到pong帧")
+ self.last_pong_time = time.time()
+ return None
+ elif opcode == 0x08: # close帧
+ logger.info("收到关闭帧")
+ self.connected = False
+ return None
+ elif opcode == 0x02: # binary帧
+ return data
+ else:
+ logger.debug(f"收到未知帧类型: {opcode}")
+ return None
+
+ except Exception as e:
+ logger.error(f"接收帧失败: {e}")
+ return None
+
+class WebSocketTestClient:
+ def __init__(self, server_addr, port):
+ self.server_addr = server_addr
+ self.port = port
+ self.websocket = None
+ self.test_running = False
+ self.send_count = 0
+ self.recv_count = 0
+ self.lost_count = 0
+ self.expected_data = bytearray(TEST_DATA_SIZE)
+ self.received_data = bytearray(TEST_DATA_SIZE)
+
+ # 测试状态
+ self.test_state = 'IDLE' # IDLE, RUNNING, COMPLETED, ERROR
+
+ # ping/pong相关
+ self.last_ping_time = None
+ self.ping_interval = 30 # 30秒发送一次ping
+
+ def connect(self):
+ """连接到WebSocket服务器"""
+ logger.info(f"连接WebSocket服务器: ws://{self.server_addr}:{self.port}/")
+
+ try:
+ self.websocket = WebSocketClient(self.server_addr, self.port)
+ if self.websocket.connect():
+ logger.info("WebSocket连接成功")
+ return True
+ else:
+ logger.error("WebSocket连接失败")
+ return False
+ except Exception as e:
+ logger.error(f"WebSocket连接失败: {e}")
+ return False
+
+ def disconnect(self):
+ """断开WebSocket连接"""
+ if self.websocket:
+ self.websocket.disconnect()
+ self.websocket = None
+ logger.info("WebSocket连接已断开")
+
+ def send_test_data(self, data_value):
+ """发送测试数据"""
+ if not self.websocket or not self.test_running:
+ return False
+
+ try:
+ # 准备测试数据
+ test_data = bytes([data_value] * TEST_DATA_SIZE)
+
+ # 发送数据
+ if self.websocket.send_frame(test_data):
+ return True
+ else:
+ return False
+ except Exception as e:
+ logger.error(f"发送数据失败: {e}")
+ return False
+
+ def receive_test_data(self):
+ """接收测试数据"""
+ if not self.websocket or not self.test_running:
+ return None
+
+ try:
+ # 接收数据
+ data = self.websocket.receive_frame()
+
+ if data and len(data) == TEST_DATA_SIZE:
+ self.received_data = data
+ return data[0] # 返回第一个字节作为数据值
+ else:
+ return None
+
+ except Exception as e:
+ logger.error(f"接收数据失败: {e}")
+ return None
+
+ def send_thread_func(self):
+ """发送线程函数"""
+ data_value = 0
+ logger.info("WebSocket发送线程启动")
+
+ while self.test_running:
+ # 发送测试数据
+ if self.send_test_data(data_value):
+ self.send_count += 1
+ data_value = (data_value + 1) % 256 # 0-255循环
+ else:
+ logger.error("发送测试数据失败")
+ break
+
+ # 发送ping帧
+ current_time = time.time()
+ if self.last_ping_time is None or (current_time - self.last_ping_time) >= self.ping_interval:
+ if self.websocket and self.websocket.connected:
+ self.websocket.send_ping()
+ self.last_ping_time = current_time
+
+ time.sleep(SEND_INTERVAL)
+
+ logger.info("WebSocket发送线程结束")
+
+ def recv_thread_func(self):
+ """接收线程函数"""
+ timeout_count = 0
+ logger.info("WebSocket接收线程启动")
+
+ while self.test_running:
+ data_value = self.receive_test_data()
+
+ if data_value is not None:
+ self.recv_count += 1
+
+ # 检查数据是否正确
+ data_correct = True
+ for i in range(TEST_DATA_SIZE):
+ if self.received_data[i] != data_value:
+ data_correct = False
+ break
+
+ if not data_correct:
+ logger.error("接收数据验证失败")
+ self.lost_count += 1
+
+ timeout_count = 0
+ else:
+ timeout_count += 1
+ if timeout_count > 10: # 1秒超时
+ logger.error("接收数据超时")
+ break
+
+ time.sleep(0.1)
+
+ logger.info("WebSocket接收线程结束")
+
+ def status_thread_func(self, duration_seconds):
+ """状态打印线程函数"""
+ logger.info("状态监控线程启动")
+
+ start_time = time.time()
+ last_status_time = start_time
+
+ while self.test_running:
+ current_time = time.time()
+ elapsed_time = current_time - start_time
+
+ # 每10秒打印一次状态
+ if current_time - last_status_time >= 10:
+ self.print_status(elapsed_time, duration_seconds)
+ last_status_time = current_time
+
+ time.sleep(1) # 每秒检查一次
+
+ logger.info("状态监控线程结束")
+
+ def print_status(self, elapsed_time, total_duration):
+ """打印当前状态"""
+ progress = (elapsed_time / total_duration) * 100 if total_duration > 0 else 0
+
+ logger.info("=== 测试状态 ===")
+ logger.info(f"运行时间: {elapsed_time:.1f}秒 / {total_duration}秒 ({progress:.1f}%)")
+ logger.info(f"发送数据包数: {self.send_count}")
+ logger.info(f"接收数据包数: {self.recv_count}")
+ logger.info(f"丢失数据包数: {self.lost_count}")
+
+ if self.send_count > 0:
+ loss_rate = (self.lost_count / self.send_count) * 100.0
+ logger.info(f"丢包率: {loss_rate:.2f}%")
+
+ if elapsed_time > 0:
+ send_rate = self.send_count / elapsed_time
+ recv_rate = self.recv_count / elapsed_time
+ logger.info(f"发送速率: {send_rate:.2f} 包/秒")
+ logger.info(f"接收速率: {recv_rate:.2f} 包/秒")
+
+ # ping/pong状态
+ if self.websocket and self.websocket.last_pong_time:
+ pong_delay = time.time() - self.websocket.last_pong_time
+ logger.info(f"最后pong时间: {pong_delay:.1f}秒前")
+
+ logger.info("================")
+
+ def start_test(self, duration_seconds=30):
+ """启动测试"""
+ if self.test_state != 'IDLE':
+ logger.error("测试正在进行中，请等待完成")
+ return False
+
+ # 连接服务器
+ if not self.connect():
+ return False
+
+ self.test_running = True
+ self.test_state = 'RUNNING'
+
+ logger.info("WebSocket测试初始化成功")
+
+ # 启动发送和接收线程
+ send_thread = threading.Thread(target=self.send_thread_func)
+ recv_thread = threading.Thread(target=self.recv_thread_func)
+ status_thread = threading.Thread(target=self.status_thread_func, args=(duration_seconds,))
+
+ send_thread.start()
+ recv_thread.start()
+ status_thread.start()
+
+ # 运行指定时间
+ time.sleep(duration_seconds)
+
+ # 停止测试
+ self.test_running = False
+
+ # 等待线程完成
+ send_thread.join()
+ recv_thread.join()
+ status_thread.join()
+
+ # 断开连接
+ self.disconnect()
+
+ self.test_state = 'COMPLETED'
+
+ # 打印结果
+ self.print_test_results()
+
+ logger.info("WebSocket测试完成")
+ return True
+
+ def stop_test(self):
+ """停止测试"""
+ if self.test_state == 'RUNNING':
+ self.test_running = False
+ logger.info("WebSocket测试已停止")
+
+ def get_test_state(self):
+ """获取测试状态"""
+ return self.test_state
+
+ def get_test_stats(self):
+ """获取测试统计信息"""
+ return self.send_count, self.recv_count, self.lost_count
+
+ def print_test_results(self):
+ """打印测试结果"""
+ logger.info("=== WebSocket测试结果 ===")
+ logger.info(f"发送数据包数: {self.send_count}")
+ logger.info(f"接收数据包数: {self.recv_count}")
+ logger.info(f"丢失数据包数: {self.lost_count}")
+
+ if self.send_count > 0:
+ loss_rate = (self.lost_count / self.send_count) * 100.0
+ logger.info(f"丢包率: {loss_rate:.2f}%")
+
+ logger.info("========================")
+
+def main():
+ """主函数"""
+ parser = argparse.ArgumentParser(description='WebSocket客户端测试程序（原生版本）')
+ parser.add_argument('--server', default='localhost', help='服务器地址 (默认: localhost)')
+ parser.add_argument('--port', type=int, default=8080, help='服务器端口 (默认: 8080)')
+ parser.add_argument('--duration', type=int, default=30, help='测试持续时间(秒) (默认: 30)')
+ parser.add_argument('--verbose', '-v', action='store_true', help='详细日志输出')
+
+ args = parser.parse_args()
+
+ if args.verbose:
+ logging.getLogger().setLevel(logging.DEBUG)
+
+ logger.info("开始WebSocket客户端测试")
+ logger.info(f"服务器地址: {args.server}")
+ logger.info(f"服务器端口: {args.port}")
+ logger.info(f"测试持续时间: {args.duration}秒")
+
+ # 创建客户端实例
+ client = WebSocketTestClient(args.server, args.port)
+
+ try:
+ # 启动测试
+ success = client.start_test(args.duration)
+
+ if success:
+ logger.info("🎉 测试完成")
+ else:
+ logger.error("💥 测试失败")
+
+ except KeyboardInterrupt:
+ logger.info("收到中断信号，正在停止测试...")
+ client.stop_test()
+ except Exception as e:
+ logger.error(f"测试运行出错: {e}")
+
+if __name__ == "__main__":
+ main()
\ No newline at end of file
--- /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
+ # 避免导入可能导致兼容性问题的异常类
+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):
+ """处理客户端连接"""
+ client_id = id(websocket)
+ client_ip, client_port = websocket.remote_address
+ logger.info(f"客户端 {client_id} 连接: {client_ip}:{client_port}, 路径: {path}")
+
+ self.clients.add(websocket)
+
+ try:
+ async for message in websocket:
+ await self.process_message(websocket, message)
+ except ConnectionClosedOK:
+ logger.info(f"客户端 {client_id} 连接正常关闭")
+ except ConnectionClosedError as e:
+ logger.error(f"客户端 {client_id} 连接异常关闭: {e}")
+ except Exception as e:
+ logger.error(f"处理客户端 {client_id} 消息时出错: {e}")
+ finally:
+ self.clients.discard(websocket)
+ logger.info(f"客户端 {client_id} 断开连接")
+
+ async def process_message(self, websocket, message):
+ """处理接收到的消息"""
+ if isinstance(message, bytes):
+ # 二进制消息
+ await self.handle_binary_message(websocket, message)
+ else:
+ # 文本消息
+ await self.handle_text_message(websocket, message)
+
+ async def handle_binary_message(self, websocket, data):
+ """处理二进制消息"""
+ 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
+ return
+
+ self.test_stats['total_received'] += 1
+
+ # 回显数据
+ try:
+ await websocket.send(data)
+ self.test_stats['total_sent'] += 1
+
+ if self.test_stats['total_received'] % 100 == 0:
+ logger.info(f"已处理 {self.test_stats['total_received']} 个数据包")
+
+ except Exception as e:
+ if "connection closed" in str(e).lower():
+ logger.info("发送数据时连接关闭")
+ else:
+ 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 Exception as e:
+ if "connection closed" in str(e).lower():
+ logger.info("发送文本响应时连接关闭")
+ else:
+ 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_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