From 6aaa47ecd6bb45dbc6a1c14b2c66dae67cce0411 Mon Sep 17 00:00:00 2001 From: =?utf8?q?=E5=AD=99=E6=98=A5=E9=98=B3?= Date: Sat, 9 Aug 2025 12:40:58 +0800 Subject: [PATCH] =?utf8?q?=E4=BF=AE=E6=94=B9cat1=5Fws.cpp=EF=BC=8Cwebsocke?= =?utf8?q?t=5Ftest.c=20=E6=8A=8Asocket=E6=93=8D=E4=BD=9C=E6=94=BE=E5=88=B0?= =?utf8?q?=E5=8D=95=E4=B8=80iothread=E6=89=A7=E8=A1=8C=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit --- code/sdk/drivers/peripheral/slip/asio.c | 12 +- .../slip/cat1Service/bk_websocket.h | 12 + .../peripheral/slip/cat1Service/cat1_ws.cpp | 328 ++++++++++++++++-- .../drivers/peripheral/slip/websocket_test.c | 88 +++-- 4 files changed, 371 insertions(+), 69 deletions(-) diff --git a/code/sdk/drivers/peripheral/slip/asio.c b/code/sdk/drivers/peripheral/slip/asio.c index 4b7719ba3..468c4d5dc 100644 --- a/code/sdk/drivers/peripheral/slip/asio.c +++ b/code/sdk/drivers/peripheral/slip/asio.c @@ -187,9 +187,9 @@ void sio_send(u8_t c, sio_fd_t fd) { slip_send_buffer.buffer[slip_send_buffer.len] = c; slip_send_buffer.len += 1; - // if( slip_send_buffer.len >= (SLIP_UART_SEND_BUFFER_MAX) ) - if( slip_send_buffer.len >= (SLIP_UART_SEND_BUFFER_MAX) || c == SLIP_END) { - //printf("send2 data len = %d\n", slip_send_buffer.len); + if( slip_send_buffer.len >= (SLIP_UART_SEND_BUFFER_MAX) ) + { + //printf("send2 data len = %d\n", slip_send_buffer.len); // uart_write_bytes(UART_NUM_0, (const char *)slip_send_buffer.buffer, slip_send_buffer.len); slip_uart_write((const char *)slip_send_buffer.buffer, slip_send_buffer.len); slip_send_buffer.len = 0; @@ -197,9 +197,9 @@ void sio_send(u8_t c, sio_fd_t fd) } void sio_sendall(sio_fd_t fd) { - // if (slip_send_buffer.len) - // slip_uart_write((const char *)slip_send_buffer.buffer, slip_send_buffer.len); - // slip_send_buffer.len = 0; + if (slip_send_buffer.len) + slip_uart_write((const char *)slip_send_buffer.buffer, slip_send_buffer.len); + slip_send_buffer.len = 0; } /** diff --git a/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h b/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h index a3380a9cb..0ccf6fc3d 100644 --- a/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h +++ b/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h @@ -39,6 +39,18 @@ void lwip_close_websocket_client(WebSocketClient *client) ; 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) ; +// === IO thread wrappers to ensure all socket operations are executed on a single thread === +// RX callback: invoked in IO thread context when data is read for a session +typedef void (*ws_rx_callback_t)(WebSocketClient *client, const uint8_t *data, int len, WebSocketMessageType type, void *user); + +void ws_io_init(void); +// Basic open (no RX callback; use ws_io_recv instead) +WebSocketClient *ws_io_open(const char *url, const char *headers, uint8_t use_dns); +// Extended open with RX callback (recommended) +WebSocketClient *ws_io_open_ex(const char *url, const char *headers, uint8_t use_dns, ws_rx_callback_t cb, void *user); +int ws_io_send(WebSocketClient *client, uint8_t *message, size_t message_len, WebSocketMessageType type); +int ws_io_recv(WebSocketClient *client, unsigned char *buf, int len, WebSocketMessageType *code); +void ws_io_close(WebSocketClient *client); #ifdef __cplusplus #if __cplusplus diff --git a/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ws.cpp b/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ws.cpp index c3bb8444e..d0443eab3 100644 --- a/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ws.cpp +++ b/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ws.cpp @@ -110,9 +110,7 @@ extern "C" { #include "mbedtls/sha1.h" -#define BUFFER_SIZE 1024 -#define WEBSOCKET_KEY_LEN 24 -#define WEBSOCKET_GUID "1234567890123456" + // #include "fs/fs.h" // #include "spiflash.h" @@ -128,6 +126,258 @@ extern "C" { #include #include +#include "msgqueue/osal_msgqueue.h" +#include "semaphore/osal_semaphore.h" +#include "bk_websocket.h" + + +#define BUFFER_SIZE 1024 +#define WEBSOCKET_KEY_LEN 25 +#define WEBSOCKET_GUID "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" + +// IO thread and command queue to serialize all socket operations +typedef enum { + WS_CMD_OPEN, + WS_CMD_SEND, + WS_CMD_RECV, + WS_CMD_CLOSE +} ws_cmd_type_t; + +typedef struct { + ws_cmd_type_t cmd; + // input args + const char *url; + const char *headers; + uint8_t use_dns; + WebSocketClient *client; + uint8_t *tx; + size_t tx_len; + WebSocketMessageType tx_type; + unsigned char *rx; + int rx_len; + WebSocketMessageType *rx_type; + // completion + int result_int; + WebSocketClient *result_client; + osal_semaphore sem; // completion semaphore +} ws_cmd_t; + +// Per-session RX callback registration +typedef struct { + WebSocketClient *client; + ws_rx_callback_t rx_cb; + void *user; +} ws_session_t; + +#define WS_MAX_SESSIONS 8 +static ws_session_t g_ws_sessions[WS_MAX_SESSIONS]; + +static unsigned long g_ws_queue_id = 0; +static osThreadId_t g_ws_io_thread = NULL; + +static void ws_cmd_init(ws_cmd_t *cmd) +{ + memset(cmd, 0, sizeof(*cmd)); + (void)osal_sem_init(&cmd->sem, 0); +} + +static void ws_cmd_deinit(ws_cmd_t *cmd) +{ + osal_sem_destroy(&cmd->sem); +} + +static void ws_sessions_init(void) +{ + memset(g_ws_sessions, 0, sizeof(g_ws_sessions)); +} + +static int ws_sessions_add(WebSocketClient *c, ws_rx_callback_t cb, void *user) +{ + for (int i = 0; i < WS_MAX_SESSIONS; ++i) { + if (g_ws_sessions[i].client == NULL) { + g_ws_sessions[i].client = c; + g_ws_sessions[i].rx_cb = cb; + g_ws_sessions[i].user = user; + return 0; + } + } + return -1; +} + +static void ws_sessions_remove(WebSocketClient *c) +{ + for (int i = 0; i < WS_MAX_SESSIONS; ++i) { + if (g_ws_sessions[i].client == c) { + memset(&g_ws_sessions[i], 0, sizeof(g_ws_sessions[i])); + return; + } + } +} + +static ws_session_t *ws_sessions_find(WebSocketClient *c) +{ + for (int i = 0; i < WS_MAX_SESSIONS; ++i) { + if (g_ws_sessions[i].client == c) { + return &g_ws_sessions[i]; + } + } + return NULL; +} + +static void ws_io_thread_entry(void *arg) +{ + (void)arg; + while (1) { + // Process next command if available; else try to poll RX on active sessions + ws_cmd_t *cmd_ptr = NULL; + unsigned int msg_size = sizeof(cmd_ptr); + if (osal_msg_queue_read_copy(g_ws_queue_id, &cmd_ptr, &msg_size, 10) == OSAL_SUCCESS && cmd_ptr != NULL) { + switch (cmd_ptr->cmd) { + case WS_CMD_OPEN: + cmd_ptr->result_client = init_websocket_client(cmd_ptr->url, cmd_ptr->headers, cmd_ptr->use_dns); + cmd_ptr->result_int = (cmd_ptr->result_client != NULL) ? 0 : -1; + break; + case WS_CMD_SEND: + cmd_ptr->result_int = send_websocket_message(cmd_ptr->client, cmd_ptr->tx, cmd_ptr->tx_len, cmd_ptr->tx_type); + break; + case WS_CMD_RECV: + cmd_ptr->result_int = receive_websocket_message(cmd_ptr->client, cmd_ptr->rx, cmd_ptr->rx_len, cmd_ptr->rx_type); + break; + case WS_CMD_CLOSE: + lwip_close_websocket_client(cmd_ptr->client); + ws_sessions_remove(cmd_ptr->client); + cmd_ptr->result_int = 0; + break; + default: + cmd_ptr->result_int = -1; + break; + } + osal_sem_up(&cmd_ptr->sem); + continue; + } + // No command -> build fd_set and wait for readable sockets + fd_set rfds; FD_ZERO(&rfds); + int maxfd = -1; int active = 0; + for (int i = 0; i < WS_MAX_SESSIONS; ++i) { + ws_session_t *s = &g_ws_sessions[i]; + if (s->client == NULL || s->rx_cb == NULL) continue; + FD_SET(s->client->sockfd, &rfds); + if (s->client->sockfd > maxfd) maxfd = s->client->sockfd; + active++; + } + if (active == 0 || maxfd < 0) { + // nothing to poll + continue; + } + struct timeval tv; tv.tv_sec = 0; tv.tv_usec = 20000; // 20ms + int ready = lwip_select(maxfd + 1, &rfds, NULL, NULL, &tv); + if (ready <= 0) { + continue; + } + for (int i = 0; i < WS_MAX_SESSIONS; ++i) { + ws_session_t *s = &g_ws_sessions[i]; + if (s->client == NULL || s->rx_cb == NULL) continue; + if (!FD_ISSET(s->client->sockfd, &rfds)) continue; + WebSocketMessageType type; + uint8_t buf[1024]; + int n = receive_websocket_message(s->client, buf, sizeof(buf), &type); + if (n > 0) { + s->rx_cb(s->client, buf, n, type, s->user); + } + } + } +} + +void ws_io_init(void) +{ + if (g_ws_queue_id == 0) { + (void)osal_msg_queue_create("ws_io_q", 16, &g_ws_queue_id, 0, sizeof(ws_cmd_t *)); + } + if (g_ws_io_thread == NULL) { + osThreadAttr_t attr; memset(&attr, 0, sizeof(attr)); attr.name = "ws_io"; attr.stack_size = 4096; attr.priority = osPriorityNormal; + ws_sessions_init(); + g_ws_io_thread = osThreadNew(ws_io_thread_entry, NULL, &attr); + } +} + +static int ws_enqueue_and_wait(ws_cmd_t *cmd) +{ + ws_cmd_t *ptr = cmd; + if (osal_msg_queue_write_copy(g_ws_queue_id, &ptr, sizeof(ptr), OSAL_MSGQ_WAIT_FOREVER) != OSAL_SUCCESS) { + return -1; + } + (void)osal_sem_down(&cmd->sem); + return 0; +} + +WebSocketClient *ws_io_open(const char *url, const char *headers, uint8_t use_dns) +{ + ws_cmd_t cmd; ws_cmd_init(&cmd); + cmd.cmd = WS_CMD_OPEN; + cmd.url = url; + cmd.headers = headers; + cmd.use_dns = use_dns; + if (ws_enqueue_and_wait(&cmd) != 0) { + ws_cmd_deinit(&cmd); + return NULL; + } + WebSocketClient *cli = cmd.result_client; + ws_cmd_deinit(&cmd); + return cli; +} + +WebSocketClient *ws_io_open_ex(const char *url, const char *headers, uint8_t use_dns, ws_rx_callback_t cb, void *user) +{ + WebSocketClient *cli = ws_io_open(url, headers, use_dns); + if (cli && cb) { + (void)ws_sessions_add(cli, cb, user); + } + return cli; +} + +int ws_io_send(WebSocketClient *client, uint8_t *message, size_t message_len, WebSocketMessageType type) +{ + ws_cmd_t cmd; ws_cmd_init(&cmd); + cmd.cmd = WS_CMD_SEND; + cmd.client = client; + cmd.tx = message; + cmd.tx_len = message_len; + cmd.tx_type = type; + if (ws_enqueue_and_wait(&cmd) != 0) { + ws_cmd_deinit(&cmd); + return -1; + } + int ret = cmd.result_int; + ws_cmd_deinit(&cmd); + return ret; +} + +int ws_io_recv(WebSocketClient *client, unsigned char *buf, int len, WebSocketMessageType *code) +{ + ws_cmd_t cmd; ws_cmd_init(&cmd); + cmd.cmd = WS_CMD_RECV; + cmd.client = client; + cmd.rx = buf; + cmd.rx_len = len; + cmd.rx_type = code; + if (ws_enqueue_and_wait(&cmd) != 0) { + ws_cmd_deinit(&cmd); + return -1; + } + int ret = cmd.result_int; + ws_cmd_deinit(&cmd); + return ret; +} + +void ws_io_close(WebSocketClient *client) +{ + ws_cmd_t cmd; ws_cmd_init(&cmd); + cmd.cmd = WS_CMD_CLOSE; + cmd.client = client; + (void)ws_enqueue_and_wait(&cmd); + ws_cmd_deinit(&cmd); +} + extern osStatus_t osDelay(uint32_t ticks); #define cat1_msleep(ms) LOS_Msleep(ms) /* osDelay(1) : 1ms */ @@ -370,23 +620,15 @@ static uint64_t ntohll(uint64_t netlonglong) { } // 生成随机的 WebSocket 密钥 void generate_websocket_key(char *key) { - static const char charset[] = - "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"; - for (int i = 0; i < WEBSOCKET_KEY_LEN - 1; ++i) { - key[i] = charset[rand() % (sizeof(charset) - 1)]; + unsigned char raw[16]; + for (int i = 0; i < 16; ++i) { + raw[i] = (unsigned char)(rand() & 0xFF); + } + size_t olen = 0; + mbedtls_base64_encode((unsigned char *)key, WEBSOCKET_KEY_LEN, &olen, raw, sizeof(raw)); + if (olen >= WEBSOCKET_KEY_LEN) { + olen = WEBSOCKET_KEY_LEN - 1; } - key[WEBSOCKET_KEY_LEN - 1] = '\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; - - strncpy(key, (const char *)encoded, olen); - key[olen] = '\0'; } @@ -401,7 +643,7 @@ void generate_accept_key(const char *key, char *accept_key) { unsigned char encoded[64]; size_t olen; - mbedtls_base64_encode(encoded, sizeof(encoded), &olen, sha1_result, 16);//sizeof(sha1_result)); + mbedtls_base64_encode(encoded, sizeof(encoded), &olen, sha1_result, sizeof(sha1_result)); strncpy(accept_key, (const char *)encoded, olen); accept_key[olen] = '\0'; } @@ -426,7 +668,7 @@ int send_handshake(WebSocketClient *client,const char *headers) { if(headers!= NULL) { snprintf(request, BUFFER_SIZE, - "GET /%s?%s HTTP/1.1\r\n" + "GET /%s%s%s HTTP/1.1\r\n" "Host: %s:%d\r\n" "Upgrade: websocket\r\n" "Connection: Upgrade\r\n" @@ -435,11 +677,11 @@ int send_handshake(WebSocketClient *client,const char *headers) { "Origin: http://%s:%d\r\n" "%s\r\n" "\r\n", - client->path,client->query ,client->host,client->port, accept_key,client->host,client->port,headers); + client->path ? client->path : "", (client->query?"?":""), (client->query?client->query:"") ,client->host,client->port, key,client->host,client->port,headers); }else { snprintf(request, BUFFER_SIZE, - "GET /%s?%s HTTP/1.1\r\n" + "GET /%s%s%s HTTP/1.1\r\n" "Host: %s:%d\r\n" "Upgrade: websocket\r\n" "Connection: Upgrade\r\n" @@ -447,7 +689,7 @@ int send_handshake(WebSocketClient *client,const char *headers) { "Sec-WebSocket-Version: 13\r\n" "Origin: http://%s:%d\r\n" "\r\n", - client->path,client->query , client->host,client->port, accept_key,client->host,client->port); + client->path ? client->path : "", (client->query?"?":""), (client->query?client->query:"") , client->host,client->port, key,client->host,client->port); } printf("send_handshake request:\r\n%s",request); @@ -719,8 +961,10 @@ static void dns_found(const char* hostname, const ip_addr_t *ipaddr, void *arg) { struct cat1_ws_init_t * ip_sem=(struct cat1_ws_init_t *)arg; - printf("hostname %s ipaddr%d \r\n",hostname,ipaddr->u_addr.ip4); - memcpy(&(ip_sem->ip),ipaddr,sizeof(ipaddr)); + printf("hostname %s\r\n", hostname); + if (ipaddr != NULL) { + memcpy(&(ip_sem->ip), ipaddr, sizeof(*ipaddr)); + } (void)osSemaphoreRelease(ip_sem->dns_done_sem); } @@ -773,20 +1017,28 @@ WebSocketClient *init_websocket_client(const char *url, const char *headers,uint // printf("init_websocket_client 3\n"); if(use_dns) { - ip_sem.dns_done_sem= osSemaphoreNew(1, 0, NULL); - if(ip_sem.dns_done_sem==0) - { - goto client_ERROR; - } + // If host is numeric IP, bypass DNS + if (inet_aton(parsed_url.host, &in_addr_t)) { + memcpy(&server_addr.sin_addr, &in_addr_t, sizeof(in_addr_t)); + } else { + ip_sem.dns_done_sem= osSemaphoreNew(1, 0, NULL); + if(ip_sem.dns_done_sem==0) + { + goto client_ERROR; + } - ret =dns_gethostbyname(parsed_url.host,&ip,dns_found,&ip_sem); - // if(ERR_OK==ret) - { - osSemaphoreAcquire(ip_sem.dns_done_sem, 1000*30); + ret = dns_gethostbyname(parsed_url.host, &ip, dns_found, &ip_sem); + if (ret == ERR_OK) { + memcpy(&server_addr.sin_addr.s_addr, &ip.u_addr.ip4, 4); + } else if (ret == ERR_INPROGRESS) { + osSemaphoreAcquire(ip_sem.dns_done_sem, 1000*30); + memcpy(&server_addr.sin_addr.s_addr, &ip_sem.ip.u_addr.ip4, 4); + } else { + osSemaphoreDelete(ip_sem.dns_done_sem); + goto client_ERROR; + } + osSemaphoreDelete(ip_sem.dns_done_sem); } - osSemaphoreDelete(ip_sem.dns_done_sem); - printf("dns_gethostbyname %d\r\n",ip_sem.ip.u_addr.ip4); - memcpy(&server_addr.sin_addr.s_addr, &ip_sem.ip.u_addr.ip4, 4); } else diff --git a/code/sdk/drivers/peripheral/slip/websocket_test.c b/code/sdk/drivers/peripheral/slip/websocket_test.c index d73e7cea9..dd9562317 100644 --- a/code/sdk/drivers/peripheral/slip/websocket_test.c +++ b/code/sdk/drivers/peripheral/slip/websocket_test.c @@ -23,10 +23,19 @@ extern "C" { // 测试配置 #define TEST_DATA_SIZE 228 -#define MAX_TEST_COUNT 1000 +#define MAX_TEST_COUNT 100000 #define THREAD_STACK_SIZE 4096 #define THREAD_PRIORITY 10 +// Queue message for received frames via IO thread callback +typedef struct { + WebSocketMessageType type; + int len; + uint8_t data[TEST_DATA_SIZE]; +} ws_msg_t; + +static osMessageQueueId_t g_ws_rx_queue = NULL; + // 全局变量 static osThreadId_t send_thread_id = NULL; static osThreadId_t recv_thread_id = NULL; @@ -61,6 +70,21 @@ static bool receive_test_data(uint8_t *data_value); static void print_test_results(void); static void check_memory_status(void); + +static void ws_on_rx(WebSocketClient *client, const uint8_t *data, int len, WebSocketMessageType type, void *user) +{ + (void)client; + (void)user; + if (g_ws_rx_queue == NULL || data == NULL || len <= 0) { + return; + } + ws_msg_t msg; + msg.type = type; + msg.len = (len > TEST_DATA_SIZE) ? TEST_DATA_SIZE : len; + memcpy(msg.data, data, (size_t)msg.len); + (void)osMessageQueuePut(g_ws_rx_queue, &msg, 0, 0); +} + /** * WebSocket发送线程 */ @@ -184,8 +208,20 @@ static bool init_websocket_test(const char *server_addr, uint16_t port) WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "连接WebSocket服务器: %s", url); - // 创建WebSocket客户端 - test_client = init_websocket_client(url, NULL, 1); // 启用DNS解析 + // 初始化单线程IO调度 + ws_io_init(); + + // 创建接收消息队列（供回调快速入队，接收线程取出处理） + if (g_ws_rx_queue == NULL) { + g_ws_rx_queue = osMessageQueueNew(32, sizeof(ws_msg_t), NULL); + if (g_ws_rx_queue == NULL) { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "创建接收队列失败"); + return false; + } + } + + // 创建WebSocket客户端（在IO线程中，并注册回调） + test_client = ws_io_open_ex(url, NULL, 1, ws_on_rx, NULL); if (test_client == NULL) { WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WebSocket客户端创建失败"); return false; @@ -195,7 +231,7 @@ static bool init_websocket_test(const char *server_addr, uint16_t port) test_mutex = osMutexNew(NULL); if (test_mutex == NULL) { WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "互斥锁创建失败"); - lwip_close_websocket_client(test_client); + ws_io_close(test_client); test_client = NULL; return false; } @@ -205,7 +241,7 @@ static bool init_websocket_test(const char *server_addr, uint16_t port) if (test_semaphore == NULL) { WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "信号量创建失败"); osMutexDelete(test_mutex); - lwip_close_websocket_client(test_client); + ws_io_close(test_client); test_client = NULL; return false; } @@ -273,9 +309,14 @@ static void cleanup_websocket_test(void) // 关闭WebSocket连接 - 在删除线程相关资源之前 if (test_client) { WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "关闭WebSocket连接"); - lwip_close_websocket_client(test_client); + ws_io_close(test_client); test_client = NULL; } + + if (g_ws_rx_queue) { + osMessageQueueDelete(g_ws_rx_queue); + g_ws_rx_queue = NULL; + } // 删除信号量 if (test_semaphore) { @@ -331,7 +372,7 @@ static bool send_test_data(uint8_t data_value) osDelay(5); // 发送数据 - int result = send_websocket_message(test_client, expected_data, TEST_DATA_SIZE, WS_BINARY); + int result = ws_io_send(test_client, expected_data, TEST_DATA_SIZE, WS_BINARY); if (result != 0) { WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "发送数据失败: %d", result); return false; @@ -350,30 +391,27 @@ static bool receive_test_data(uint8_t *data_value) 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); + if (g_ws_rx_queue == NULL) { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "接收数据失败: 队列未初始化"); return false; } - - if (type != WS_BINARY) { - WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "接收数据类型错误: type=%d", type); + ws_msg_t msg; + uint8_t prio = 0; + osStatus_t st = osMessageQueueGet(g_ws_rx_queue, &msg, &prio, 100); + if (st != osOK) { + return false; // 超时或无数据 + } + if (msg.type != WS_BINARY) { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "接收数据类型错误: type=%d", msg.type); return false; } - - // 检查数据大小 - if (result != TEST_DATA_SIZE) { - WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "接收数据大小错误: %d", result); + if (msg.len != TEST_DATA_SIZE) { + WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "接收数据大小错误: %d", msg.len); return false; } - - // 提取数据值 + memcpy(received_data, msg.data, TEST_DATA_SIZE); *data_value = received_data[0]; - - WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "接收数据成功: 值=%u, 大小=%d", *data_value, result); + WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "接收数据成功: 值=%u, 大小=%d", *data_value, msg.len); return true; } @@ -453,7 +491,7 @@ int at_test_websocket_handler(const char *cmd, const char *params) // 强制清理 test_running = false; if (test_client) { - lwip_close_websocket_client(test_client); + ws_io_close(test_client); test_client = NULL; } test_state = TEST_STATE_IDLE; -- 2.17.1