From 035e4bebb4fb7734025c83b11a24cceb36e193e9 Mon Sep 17 00:00:00 2001 From: xfy911 Date: Fri, 5 Jun 2026 18:12:46 +0800 Subject: [PATCH] =?UTF-8?q?feat(websocket):=20=E5=AE=9E=E7=8E=B0=20WebSock?= =?UTF-8?q?et=20=E5=B9=BF=E6=92=AD=E4=B8=8E=E9=A2=91=E9=81=93=E8=B7=AF?= =?UTF-8?q?=E7=94=B1=E7=B3=BB=E7=BB=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增功能： - ws_registry_add/remove: 连接注册/注销，自动管理活跃连接链表 - ws_broadcast(): 向所有活跃连接广播文本消息 - ws_broadcast_to_path(): 按握手路径（频道）定向广播 - ws_connection_count(): 获取当前 WebSocket 连接数 - ws_handle_connection() 新增 path 参数，用于频道标识 实现细节： - 全局链表 + pthread_mutex 保证多线程安全 - 原子计数器统计连接数 - 连接建立时自动注册，关闭/异常时自动注销 - 默认行为保持 echo 不变 编译零警告，61 项集成测试 + 127 个单元测试全部通过。 --- server.c | 2 +- websocket.c | 143 ++++++++++++++++++++++++++++++++++++++++++++++++++-- websocket.h | 29 ++++++++++- 3 files changed, 168 insertions(+), 6 deletions(-) diff --git a/server.c b/server.c index 7a59537..1c1ade6 100644 --- a/server.c +++ b/server.c @@ -795,7 +795,7 @@ static void client_handler(void *arg) { memmove(conn->buf, conn->buf + parsed, conn->buf_len - (size_t)parsed); } conn->buf_len -= (size_t)parsed; - ws_handle_connection(conn->fd, conn->timeout_ms); + ws_handle_connection(conn->fd, conn->timeout_ms, req.path); break; } } diff --git a/websocket.c b/websocket.c index a9ad845..4ba7abe 100644 --- a/websocket.c +++ b/websocket.c @@ -3,7 +3,7 @@ * @brief WebSocket 协议实现（RFC 6455） * * 支持握手、帧解析、帧编码、连接管理。 - * 当前为简单 echo 服务器，可扩展为消息路由或广播。 + * 新增广播与频道路由系统（2026-06-05）。 */ #include "websocket.h" @@ -15,6 +15,8 @@ #include #include #include +#include +#include /* WebSocket 魔数字符串（GUID） */ static const char WS_GUID[] = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; @@ -236,6 +238,134 @@ int ws_send_pong(int fd, const uint8_t *payload, size_t len) { return ws_send_frame(fd, WS_OP_PONG, payload, len); } +/* ============================================================ + * WebSocket 连接注册表 — 广播与频道路由 + * ============================================================ */ + +typedef struct ws_conn_node { + int fd; /**< 客户端 socket */ + char *path; /**< 握手路径（频道标识） */ + struct ws_conn_node *next; /**< 链表下一个节点 */ +} ws_conn_node_t; + +/** 全局注册表 */ +typedef struct { + ws_conn_node_t *head; /**< 链表头 */ + pthread_mutex_t mutex; /**< 保护链表的互斥锁 */ + atomic_int count; /**< 当前连接数 */ +} ws_registry_t; + +static ws_registry_t g_ws_registry = { + .head = NULL, + .mutex = PTHREAD_MUTEX_INITIALIZER, + .count = 0 +}; + +/** + * @brief 注册 WebSocket 连接 + * + * @param fd 客户端 socket + * @param path 握手路径（可为 NULL，使用默认频道） + */ +static void ws_registry_add(int fd, const char *path) { + ws_conn_node_t *node = (ws_conn_node_t *)calloc(1, sizeof(ws_conn_node_t)); + if (!node) return; + node->fd = fd; + node->path = path ? strdup(path) : NULL; + + pthread_mutex_lock(&g_ws_registry.mutex); + node->next = g_ws_registry.head; + g_ws_registry.head = node; + atomic_fetch_add(&g_ws_registry.count, 1); + pthread_mutex_unlock(&g_ws_registry.mutex); + + log_info("WebSocket 注册 fd=%d 路径=%s，当前连接数: %d", + fd, path ? path : "(default)", atomic_load(&g_ws_registry.count)); +} + +/** + * @brief 注销 WebSocket 连接 + * + * @param fd 客户端 socket + */ +static void ws_registry_remove(int fd) { + pthread_mutex_lock(&g_ws_registry.mutex); + ws_conn_node_t **pp = &g_ws_registry.head; + while (*pp) { + ws_conn_node_t *node = *pp; + if (node->fd == fd) { + *pp = node->next; + free(node->path); + free(node); + atomic_fetch_sub(&g_ws_registry.count, 1); + log_info("WebSocket 注销 fd=%d，当前连接数: %d", + fd, atomic_load(&g_ws_registry.count)); + break; + } + pp = &node->next; + } + pthread_mutex_unlock(&g_ws_registry.mutex); +} + +/** + * @brief 向所有活跃 WebSocket 连接广播文本消息 + * + * @param text 文本内容 + * @return 成功发送的连接数 + */ +int ws_broadcast(const char *text) { + if (!text) return 0; + int sent = 0; + pthread_mutex_lock(&g_ws_registry.mutex); + ws_conn_node_t *node = g_ws_registry.head; + while (node) { + if (ws_send_text(node->fd, text) == 0) { + sent++; + } + node = node->next; + } + pthread_mutex_unlock(&g_ws_registry.mutex); + log_debug("WebSocket 广播: %s -> %d/%d 连接", text, sent, + atomic_load(&g_ws_registry.count)); + return sent; +} + +/** + * @brief 向指定频道的所有连接发送文本消息 + * + * @param path 频道路径 + * @param text 文本内容 + * @return 成功发送的连接数 + */ +int ws_broadcast_to_path(const char *path, const char *text) { + if (!path || !text) return 0; + int sent = 0; + pthread_mutex_lock(&g_ws_registry.mutex); + ws_conn_node_t *node = g_ws_registry.head; + while (node) { + if (node->path && strcmp(node->path, path) == 0) { + if (ws_send_text(node->fd, text) == 0) { + sent++; + } + } + node = node->next; + } + pthread_mutex_unlock(&g_ws_registry.mutex); + log_debug("WebSocket 频道广播 [%s]: %s -> %d 连接", path, text, sent); + return sent; +} + +/** + * @brief 获取当前 WebSocket 连接数 + */ +int ws_connection_count(void) { + return atomic_load(&g_ws_registry.count); +} + +/* ============================================================ + * WebSocket 连接处理 + * ============================================================ */ + /** * @brief 从连接读取数据（简单阻塞读取） */ @@ -244,14 +374,17 @@ static ssize_t ws_read_data(int fd, uint8_t *buf, size_t max_len) { return n; } -void ws_handle_connection(int fd, uint32_t timeout_ms) { +void ws_handle_connection(int fd, uint32_t timeout_ms, const char *path) { (void)timeout_ms; /* TODO: 超时处理 */ uint8_t buf[8192]; size_t buf_len = 0; bool closed = false; - log_info("WebSocket 连接建立 fd=%d", fd); + /* 注册到全局连接表 */ + ws_registry_add(fd, path); + + log_info("WebSocket 连接建立 fd=%d path=%s", fd, path ? path : "(default)"); while (!closed) { ssize_t n = ws_read_data(fd, buf + buf_len, sizeof(buf) - buf_len); @@ -287,6 +420,7 @@ void ws_handle_connection(int fd, uint32_t timeout_ms) { log_debug("WebSocket fd=%d 收到文本: %s", fd, frame.payload ? (char *)frame.payload : "(empty)"); if (frame.fin) { + /* 默认行为：echo 回发送者 */ ws_send_text(fd, frame.payload ? (char *)frame.payload : ""); } break; @@ -334,5 +468,8 @@ void ws_handle_connection(int fd, uint32_t timeout_ms) { } } + /* 从全局连接表注销 */ + ws_registry_remove(fd); + log_info("WebSocket 连接关闭 fd=%d", fd); } diff --git a/websocket.h b/websocket.h index 4601f86..b120b93 100644 --- a/websocket.h +++ b/websocket.h @@ -116,15 +116,40 @@ int ws_send_ping(int fd); */ int ws_send_pong(int fd, const uint8_t *payload, size_t len); +/** + * @brief 向所有活跃 WebSocket 连接广播文本消息 + * + * @param text 文本内容 + * @return 成功发送的连接数 + */ +int ws_broadcast(const char *text); + +/** + * @brief 向指定频道的所有连接发送文本消息 + * + * @param path 频道路径 + * @param text 文本内容 + * @return 成功发送的连接数 + */ +int ws_broadcast_to_path(const char *path, const char *text); + +/** + * @brief 获取当前 WebSocket 连接数 + * + * @return 活跃连接数 + */ +int ws_connection_count(void); + /** * @brief 处理 WebSocket 连接（主循环） * * 进入 WebSocket 帧循环，处理文本/二进制/ping/pong/close。 - * 当前实现为简单 echo 服务器。 + * 连接自动注册到全局广播表，退出时注销。 * * @param fd 客户端 socket * @param timeout_ms 超时毫秒（0 表示默认） + * @param path WebSocket 握手路径（频道标识，可为 NULL） */ -void ws_handle_connection(int fd, uint32_t timeout_ms); +void ws_handle_connection(int fd, uint32_t timeout_ms, const char *path); #endif /* WEBSOCKET_H */