#include "mbedtls/sha1.h"
-#define BUFFER_SIZE 1024
-#define WEBSOCKET_KEY_LEN 24
-#define WEBSOCKET_GUID "1234567890123456"
+
// #include "fs/fs.h"
// #include "spiflash.h"
#include <stdint.h>
#include <asm/delay.h>
+#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 */
}
// 生成随机的 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';
}
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';
}
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"
"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"
"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);
{
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);
}
// 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
// 测试配置
#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;
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发送线程
*/
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;
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;
}
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;
}
// 关闭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) {
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;
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;
}
// 强制清理
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;
projects / mirrors / FD230.git / commitdiff