From 9b79e6690dce556b18754075f065508e2c22bbb8 Mon Sep 17 00:00:00 2001 From: xRain Date: Mon, 24 Aug 2026 23:52:15 +0800 Subject: [PATCH] =?UTF-8?q?=E9=87=8D=E6=9E=84=E6=9C=8D=E5=8A=A1=E7=AB=AF?= =?UTF-8?q?=E4=B8=BA=E6=8E=A5=E5=8F=A3=E5=BC=8F(WebSocketServer(port)+addH?= =?UTF-8?q?andler=20=E5=B7=A5=E5=8E=82,=E6=AF=8F=E8=BF=9E=E6=8E=A5?= =?UTF-8?q?=E7=8B=AC=E7=AB=8B=20handler),=E7=BB=84=E7=AE=A1=E7=90=86?= =?UTF-8?q?=E4=B8=8E=E5=8F=91=E9=80=81=E6=94=B6=E6=95=9B=E4=B8=BA=E9=93=BE?= =?UTF-8?q?=E5=BC=8F=E9=97=A8=E9=9D=A2=20group(name)/conn(connId),?= =?UTF-8?q?=E7=BB=9F=E4=B8=80=20send=20=E4=B8=89=E9=87=8D=E9=87=8D?= =?UTF-8?q?=E8=BD=BD,=E5=8D=87=E7=BA=A7=E7=89=88=E6=9C=AC=201.1.0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 148 +++++++++++----- cjpm.toml | 4 +- src/client/WebSocketClient.cj | 8 +- src/server/IWebsocketHandler.cj | 26 +++ src/server/WebSocketConnection.cj | 127 ++++++++------ src/server/WebSocketSenders.cj | 188 ++++++++++++++++++++ src/server/WebSocketServer.cj | 196 +++++++++++---------- src/tests/websocket_test.cj | 277 ++++++++++++++++++++++-------- 8 files changed, 701 insertions(+), 273 deletions(-) create mode 100644 src/server/IWebsocketHandler.cj create mode 100644 src/server/WebSocketSenders.cj diff --git a/README.md b/README.md index 13af6b3..0ee6688 100644 --- a/README.md +++ b/README.md @@ -1,18 +1,20 @@ # simcu::websocket — 仓颉 WebSocket 库（RFC 6455） -事件回调式 WebSocket 客户端与服务端，纯仓颉实现，仅依赖标准库 + stdx（无第三方库）。 +事件回调式 WebSocket 客户端 + 接口式服务端，纯仓颉实现，仅依赖标准库 + stdx（无第三方库）。 - 组织：`simcu` - 包名：`websocket` - 子包：`simcu::websocket.client` / `simcu::websocket.server` / `simcu::websocket.common` -- 构建：`cjpm build`；测试：`cjpm test`（8 个端到端用例全通过） +- 构建：`cjpm build`；测试：`cjpm test`（10 个端到端用例全通过） ## 功能特性 -- **客户端**：`connect()` / `send()` / `close()` / `terminate()` + 事件回调 `onOpen` / `onMessage` / `onError` / `onClose` -- **服务端**：`on('connection')` / `on('message')` / `send()` / `close()` / `terminate()` + 监听 `on('close')` / `on('error')` +- **客户端**：`connect()` / `send()`（三重重载）/ `ping()` / `close()` / `terminate()` + 事件回调 `onOpen` / `onMessage` / `onError` / `onClose` +- **服务端**：`WebSocketServer(port)` + `addHandler(path, factory)` 按路径路由，每连接一个独立 handler 实例；`start()` / `close()` +- **链式门面**：组管理与发送统一收敛到 `ws.group(name)` / `ws.conn(connId)`，同一个功能只有一种入口 +- **组管理**：入组 / 出组 / 组计数 / 组判空 / 组内连接快照 / 连接所属组 / 全部组名，连接断开自动清组 +- **发送（三重重载）**：`send(message)` / `send(text)` / `send(data)` 覆盖消息对象、文本、二进制；客户端 / 连接 / 组播 / 单播 / 广播统一 - **RFC 6455 完整帧层**：掩码、分片重组、Ping/Pong、Close 握手、`maxPayload` 上限（超限 1009） -- **群发广播**：`broadcastText` / `broadcastBinary` - **wss（TLS）**：`FrameStream` 基于 `StreamingSocket`，可直接包裹 `TlsSocket`（已在 BotRoleHelper 实战验证） ## 依赖引入 @@ -27,6 +29,7 @@ ```cangjie import simcu::websocket.client.WebSocketClient import simcu::websocket.server.WebSocketServer +import simcu::websocket.server.IWebsocketHandler import simcu::websocket.server.WebSocketConnection import simcu::websocket.common.WebSocketMessage ``` @@ -42,7 +45,9 @@ src/ ├── client/ │ └── WebSocketClient.cj # 客户端 ├── server/ -│ ├── WebSocketServer.cj # 服务端监听 / 握手 / 广播 +│ ├── IWebsocketHandler.cj # 服务端 handler 接口（onConnect/onMessage/onClose/onError） +│ ├── WebSocketServer.cj # 监听 / 握手 / 按路径路由 / 广播 / 组存储 +│ ├── WebSocketSenders.cj # 链式门面（组管理 + 组播/单播发送的实现） │ └── WebSocketConnection.cj # 服务端单条连接 └── tests/ └── websocket_test.cj # 端到端集成测试 @@ -61,8 +66,8 @@ client.onClose = { code: Int64, reason: String => ... } // 关闭，携带关 client.connect(timeout: Duration.second * 30) // 建立连接；握手失败抛 WebSocketException client.send(WebSocketMessage) // 发送消息对象 -client.sendText("hello") // 发送文本 -client.sendBinary(byteArray) // 发送二进制 +client.send("hello") // 发送文本（String 重载） +client.send(byteArray) // 发送二进制（Array 重载） client.ping() // 心跳（服务端自动回 Pong） client.close(code: 1000, reason: "bye") // 关闭握手，5 秒兜底强制断开 client.terminate() // 立即断开（对端收 1006） @@ -83,43 +88,98 @@ client.isOpen() ## 服务端 API ```cangjie -let server = WebSocketServer(bindAt: 8080, path: Some("/ws"), maxPayload: 65536) +let server = WebSocketServer(8080) // 端口，0 表示随机空闲端口 +server.addHandler("/ws") { => MyHandler() } // 路径 -> 工厂，每次握手创建独立 handler 实例 +server.start() // 开始监听（异步 accept） +... +server.close() // 停止监听并断开所有连接 -// 服务端事件 -server.on("connection", { conn: WebSocketConnection => ... }) // 新连接完成握手 -server.on("message", { conn, msg => ... }) // 任意连接收到消息 -server.on("close", { conn, code, reason => ... }) // 任意连接关闭 -server.on("error", { e => ... }) // 监听 / 连接错误 - -server.listen() // 开始监听（异步 accept） -server.close() // 停止监听并关闭全部连接 -server.broadcastText("hi-all") // 群发文本 -server.broadcastBinary(bytes) // 群发二进制 -server.localPort // 实际监听端口（bindAt=0 时随机） -server.connectionCount // 当前连接数 +server.localPort // 实际监听端口（端口 0 时用） +server.connectionCount // 当前在线连接数 +server.broadcast("hi-all") // 广播文本 +server.broadcast(bytes) // 广播二进制 +server.groups() // 所有组名（快照） ``` 构造参数： | 参数 | 默认 | 说明 | |---|---|---| -| `bindAt: UInt16` | `0` | 监听端口，`0` 表示随机空闲端口（`listen` 后读 `localPort`） | -| `path: ?String` | `None` | 仅接受该路径的握手请求，`None` 表示不限制 | +| `port: Int64` | — | 监听端口，`0` 表示随机空闲端口（`start` 后读 `localPort`） | | `maxPayload: Int64` | `65536` | 单条消息最大字节数 | +### Handler 接口 + +`IWebsocketHandler` 只关心连接生命周期回调；端点路径由 `addHandler(path, factory)` 注册时指定，handler 自身不感知。每个连接在握手成功时由工厂创建一个独立的 handler 实例，因此 handler 内可持有该连接的会话状态（如 connectionId / 组信息），但不要持有会跨连接共享的 scoped 服务。 + +```cangjie +public interface IWebsocketHandler { + func onConnect(conn: WebSocketConnection): Unit + func onMessage(conn: WebSocketConnection, msg: WebSocketMessage): Unit + func onClose(conn: WebSocketConnection, code: Int64, reason: String): Unit + func onError(conn: WebSocketConnection, e: Exception): Unit +} +``` + +实现示例： + +```cangjie +class MyHandler <: IWebsocketHandler { + public override func onConnect(conn: WebSocketConnection): Unit { + conn.joinGroup("authed") // 握手成功即入组 + } + + public override func onMessage(conn: WebSocketConnection, msg: WebSocketMessage): Unit { + conn.send("echo: ${msg.text.getOrThrow()}") // send 三重重载：String / Array / WebSocketMessage + } + + public override func onClose(conn: WebSocketConnection, code: Int64, reason: String): Unit { + () // 连接断开由 server 自动清组，无需手动处理 + } + + public override func onError(conn: WebSocketConnection, e: Exception): Unit { + () + } +} +``` + +### 链式门面（组管理 + 发送） + +组管理与发送的全部实现收敛在链式门面中，`WebSocketServer` 上不再暴露散落的组管理方法： + +```cangjie +// 组管理 +server.group("room").join(conn) // 入组（组不存在自动创建），也支持 join(connId) +server.group("room").leave(conn) // 出组（组空自动删除），也支持 leave(connId) +server.group("room").count() // 组内连接数 +server.group("room").isEmpty() // 组是否为空 +server.group("room").connIds() // 组内连接Id快照 + +// 组播 / 单播 / 广播（send 三重重载：WebSocketMessage / String / Array） +server.group("room").send("hi").send(bytes) +server.conn(connId).send("hi") // 连接不存在则静默忽略 +server.broadcast("hi-all") + +// 查询 +server.conn(connId).groups() // 连接所属的所有组名 +server.groups() // 所有组名 +``` + +组为空时自动删除；连接断开由 server 自动从所有组移除，无需手动清理。 + ### 连接级 API ```cangjie -// 连接级事件（conn.on 重载按监听器参数区分，也可直接赋 conn.onMessage 等属性） -conn.on("message", { m: WebSocketMessage => ... }) -conn.on("close", { code: Int64, reason: String => ... }) -conn.on("error", { e: Exception => ... }) - -conn.send(message) / sendText(text) / sendBinary(bytes) / ping() -conn.close(code: 4000, reason: "svr") // 关闭握手 -conn.terminate() // 立即断开 +conn.send(message) / conn.send(text) / conn.send(bytes) // 三重重载 +conn.ping() // 心跳 +conn.close(code: 4000, reason: "svr") // 关闭握手 +conn.terminate() // 立即断开 +conn.joinGroup("room") / conn.leaveGroup("room") // 连接自己入组/出组（经门面实现） +conn.connectionId // 唯一连接Id conn.readyState / conn.isOpen() -conn.remoteAddress // "ip:port" +conn.remoteAddress // "ip:port" +conn.requestPath // 握手请求路径 +conn.requestQuery // 握手请求 query ``` ## 底层帧 API（common 子包） @@ -181,9 +241,6 @@ WebSocketMessage.fromText("hi") / fromBinary(bytes) // WebSocketException（异常含 RFC 6455 关闭码） ex.code // 1002 协议错误 / 1009 消息过大 / 1006 对端异常断开 ... CloseCodes.messageTooBig // 常用关闭码常量 - -// 服务端事件名（on(event, ...) 使用） -WsEvents.connection / WsEvents.message / WsEvents.close / WsEvents.error ``` ## 测试覆盖 @@ -191,25 +248,28 @@ WsEvents.connection / WsEvents.message / WsEvents.close / WsEvents.error | 用例 | 验证点 | |---|---| | echoText | 客户端发送 → 服务端回声 → 客户端收到 | -| broadcast | 双客户端群发均收到 | +| broadcast | 双客户端广播均收到 | | closeHandshake | client.close(1000, "bye") 双端 onClose 收到 (1000, "bye") | | serverClose | conn.close(4000, "svr") → 客户端收 (4000, "svr") | | terminate | client.terminate() → 服务端收 (1006, "") | | ping | 心跳后连接仍可收发 | -| pathFilter | path 过滤：/ws 通过，/other 拒绝 | +| pathRoute | 按路径路由：/ws 命中 handler，其他路径拒绝 | | maxPayload | 超限触发 onError(code=1009) | +| groupMulticast | 组播两连接均收到；连接断开自动退组、组计数减一 | +| unicast | conn(connId).send 只发到指定连接，另一连接收不到 | -## 已补齐的 API 缺口 +## API 演进记录 -对照最初需求（connect/send/close + onOpen/onMessage/onError/onClose），补齐了： +对照最初需求（connect/send/close + onOpen/onMessage/onError/onClose），已补齐： -- **sendText / sendBinary**：按文本 / 二进制发送的重载（`send(WebSocketMessage)` 之外） +- **接口式服务端**：`WebSocketServer(port)` + `addHandler(path, factory)`，按路径路由，每连接独立 handler 实例 +- **send 三重重载**：`send(WebSocketMessage)` / `send(String)` / `send(Array)` 全库统一（客户端 / 连接 / 门面 / 广播），不再有 sendText / sendBinary / broadcastText 等散落命名 +- **链式门面**：`ws.group(name)` / `ws.conn(connId)` 收敛组管理与发送，同一个功能只有一种入口 +- **组管理**：join / leave / count / isEmpty / connIds / groupsOf / groups + 连接断开自动清组 - **ping()**：心跳保活 - **close(code, reason)**：携带关闭码与原因的关闭握手；onClose 回调带 (code, reason) -- **readyState / isOpen()**：连接状态查询 -- **broadcastText / broadcastBinary**：服务端群发 -- **localPort / connectionCount / remoteAddress**：服务端与连接元信息 -- **服务端 path 过滤**：仅接受指定路径的握手 +- **readyState / isOpen() / connectionId / remoteAddress / requestPath / requestQuery**：连接状态与元信息 +- **localPort / connectionCount**：服务端元信息 - **maxPayload**：单条消息上限，超限报 1009 ## 尚未实现（后续方向） diff --git a/cjpm.toml b/cjpm.toml index 13854aa..e9d21f3 100644 --- a/cjpm.toml +++ b/cjpm.toml @@ -2,8 +2,8 @@ cjc-version = "1.1.3" name = "websocket" organization = "simcu" -description = "SimApi WebSocket 库（RFC 6455）：事件回调式客户端与服务端。客户端 connect/send/close + onOpen/onMessage/onError/onClose；服务端 on('connection'/'message'/'close'/'error') + Connection send/close/terminate，支持广播与心跳" -version = "1.0.0" +description = "SimApi WebSocket 库（RFC 6455）：事件回调式客户端与接口式服务端。客户端 connect/send(三重重载)/ping/close/terminate + onOpen/onMessage/onError/onClose；服务端 WebSocketServer(port) + addHandler(path, factory) 按路径路由、每连接独立 handler，链式门面 group(name)/conn(connId) 支持组管理/组播/单播/广播，连接断开自动清组" +version = "1.1.0" target-dir = "" output-type = "static" diff --git a/src/client/WebSocketClient.cj b/src/client/WebSocketClient.cj index 4eb60bb..26b57df 100644 --- a/src/client/WebSocketClient.cj +++ b/src/client/WebSocketClient.cj @@ -7,7 +7,7 @@ * client.onMessage = { (m) => println(m.text) } * client.onClose = { (code, reason) => println("关闭: ${code} ${reason}") } * client.connect() - * client.sendText("hello") + * client.send("hello") * client.close() // 或 client.terminate() 立即断开 * * 说明： @@ -117,13 +117,13 @@ public class WebSocketClient { } } - /// 发送文本消息。 - public func sendText(text: String): Unit { + /// 发送文本消息（三重重载：WebSocketMessage / String / Array）。 + public func send(text: String): Unit { frame.getOrThrow().writeText(text) } /// 发送二进制消息。 - public func sendBinary(data: Array): Unit { + public func send(data: Array): Unit { frame.getOrThrow().writeBinary(data) } diff --git a/src/server/IWebsocketHandler.cj b/src/server/IWebsocketHandler.cj new file mode 100644 index 0000000..0c09bf7 --- /dev/null +++ b/src/server/IWebsocketHandler.cj @@ -0,0 +1,26 @@ +/* + * simcu::websocket.server —— WebSocket 服务端 handler 接口。 + * + * handler 只关心连接生命周期回调:onConnect / onMessage / onClose / onError。 + * 端点路径由 WebSocketServer.addHandler(path, factory) 注册时指定,handler 自身不感知。 + * 每个连接在握手成功时由工厂创建一个独立的 handler 实例,因此 handler 内可持有该 + * 连接的会话状态(如 connectionId / 组信息),但不要持有会跨连接共享的 scoped 服务。 + */ +package simcu::websocket.server + +import simcu::websocket.common.WebSocketMessage + +/// WebSocket 服务端 handler 接口(多 handler 按注册路径路由)。 +public interface IWebsocketHandler { + /// 握手成功、连接建立后触发。 + func onConnect(conn: WebSocketConnection): Unit + + /// 连接收到消息时触发。 + func onMessage(conn: WebSocketConnection, msg: WebSocketMessage): Unit + + /// 连接关闭时触发(参数为关闭码、关闭原因)。 + func onClose(conn: WebSocketConnection, code: Int64, reason: String): Unit + + /// 连接运行时错误时触发。 + func onError(conn: WebSocketConnection, e: Exception): Unit +} diff --git a/src/server/WebSocketConnection.cj b/src/server/WebSocketConnection.cj index cb7ec05..67f635e 100644 --- a/src/server/WebSocketConnection.cj +++ b/src/server/WebSocketConnection.cj @@ -1,11 +1,9 @@ /* * simcu::websocket.server —— 服务端单条 WebSocket 连接。 * - * 由 WebSocketServer 在握手成功后创建并暴露给 on('connection') 回调。 - * 提供 send/sendText/sendBinary/ping/close/terminate，以及连接级事件： - * conn.on("message", { (msg) => ... }) - * conn.on("close", { (code, reason) => ... }) - * conn.on("error", { (e) => ... }) + * 由 WebSocketServer 在握手成功后创建并交给匹配的 IWebsocketHandler。 + * 提供 send（三重重载：WebSocketMessage / String / Array）/ ping / close / terminate，以及只读属性 + * connectionId / readyState / remoteAddress / requestPath / requestQuery。 */ package simcu::websocket.server @@ -19,11 +17,11 @@ import simcu::websocket.common.MessageType import simcu::websocket.common.ReadyState import simcu::websocket.common.WebSocketMessage import simcu::websocket.common.WebSocketException -import simcu::websocket.common.WsEvents -/// 服务端单条 WebSocket 连接（事件回调式）。 +/// 服务端单条 WebSocket 连接（由对应 handler 处理回调）。 public class WebSocketConnection { private let parent: ?WebSocketServer + private let handler: IWebsocketHandler private let frame: FrameStream private let stateLock = Mutex() private var state: ReadyState = ReadyState.Open @@ -31,17 +29,54 @@ public class WebSocketConnection { /// 连接唯一编号（由服务端分配，用于连接集合的引用比较）。 internal let connId: Int64 - /// 连接级事件：收到消息（不含父级 connection 参数）。 - public var onMessage: (WebSocketMessage) -> Unit = { m => () } - /// 连接级事件：连接关闭，参数 (关闭码, 关闭原因)。 - public var onClose: (Int64, String) -> Unit = { c, r => () } - /// 连接级事件：运行时错误。 - public var onError: (Exception) -> Unit = { e => () } + /// 连接唯一编号（对外只读，业务层用于关联连接状态）。 + public prop connectionId: Int64 { + get() { connId } + } - internal init(parent: ?WebSocketServer, frame: FrameStream, connId: Int64) { + /// 握手请求路径（不含 query，如 "/ws/server"）。 + public let requestPath: String + /// 握手请求 query 字符串（原始形式，不含 '?'；无 query 时为空字符串，如 "serverId=srv-1"）。 + public let requestQuery: String + + /// 回调归属的 handler（由 WebSocketServer 按路径匹配后注入）。 + internal prop wsHandler: IWebsocketHandler { + get() { handler } + } + + /// 归属的服务端（用于组管理 / 单播 / 组播委托，None 表示已脱离服务端）。 + public prop server: ?WebSocketServer { + get() { parent } + } + + internal init(parent: ?WebSocketServer, handler: IWebsocketHandler, frame: FrameStream, + connId: Int64, requestPath: String, requestQuery: String) { this.parent = parent + this.handler = handler this.frame = frame this.connId = connId + this.requestPath = requestPath + this.requestQuery = requestQuery + } + + /// 取 query 参数值（如 "?serverId=srv-1" → queryParam("serverId") = Some("srv-1")）。 + /// 参数不存在返回 None；参数存在但无值时返回 Some("")。 + public func queryParam(name: String): ?String { + if (requestQuery.isEmpty()) { + return None + } + for (pair in requestQuery.split("&")) { + if (let Some(e) <- pair.indexOf("=")) { + if (pair[0..e] == name) { + return Some(pair[e + 1..]) + } + } else { + if (pair == name) { + return Some("") + } + } + } + None } /// 当前连接状态。 @@ -66,34 +101,7 @@ public class WebSocketConnection { } } - /// 注册连接级事件。支持 "message" / "close" / "error"。 - public func on(event: String, listener: (WebSocketMessage) -> Unit): Unit { - if (event == WsEvents.message) { - onMessage = listener - } else { - throw WebSocketException("不支持的事件: ${event}") - } - } - - /// 注册连接级事件。支持 "close" / "error"。 - public func on(event: String, listener: (Int64, String) -> Unit): Unit { - if (event == WsEvents.close) { - onClose = listener - } else { - throw WebSocketException("不支持的事件: ${event}") - } - } - - /// 注册连接级事件。支持 "error"。 - public func on(event: String, listener: (Exception) -> Unit): Unit { - if (event == WsEvents.error) { - onError = listener - } else { - throw WebSocketException("不支持的事件: ${event}") - } - } - - /// 发送消息。 + /// 发送消息（三重重载：WebSocketMessage / String 文本 / Array 二进制）。 public func send(message: WebSocketMessage): Unit { ensureOpen() if (message.`type` == MessageType.Text) { @@ -104,17 +112,33 @@ public class WebSocketConnection { } /// 发送文本消息。 - public func sendText(text: String): Unit { + public func send(text: String): Unit { ensureOpen() frame.writeText(text) } /// 发送二进制消息。 - public func sendBinary(data: Array): Unit { + public func send(data: Array): Unit { ensureOpen() frame.writeBinary(data) } + // ===== 组管理（经 server 转发） ===== + + /// 将当前连接加入组（组不存在则自动创建）。 + public func joinGroup(group: String): Unit { + if (let Some(s) <- parent) { + s.group(group).join(this) + } + } + + /// 将当前连接移出组。 + public func leaveGroup(group: String): Unit { + if (let Some(s) <- parent) { + s.group(group).leave(this) + } + } + /// 发送 Ping（心跳保活）。 public func ping(): Unit { ensureOpen() @@ -193,20 +217,14 @@ public class WebSocketConnection { } else { WebSocketMessage(MessageType.Binary, f.payload) } - onMessage(msg) - if (let Some(p) <- parent) { - p.fireMessage(this, msg) - } + handler.onMessage(this, msg) } } } } catch (e: Exception) { code = 1006 if (state != ReadyState.Closing) { - onError(e) - if (let Some(p) <- parent) { - p.fireError(e) - } + handler.onError(this, e) } } synchronized(stateLock) { @@ -220,9 +238,6 @@ public class WebSocketConnection { if (let Some(p) <- parent) { p.removeConnection(this) } - onClose(code, reason) - if (let Some(p) <- parent) { - p.fireClose(this, code, reason) - } + handler.onClose(this, code, reason) } } diff --git a/src/server/WebSocketSenders.cj b/src/server/WebSocketSenders.cj new file mode 100644 index 0000000..52e4957 --- /dev/null +++ b/src/server/WebSocketSenders.cj @@ -0,0 +1,188 @@ +/* + * simcu::websocket.server —— 链式门面（组管理 + 发送）。 + * + * 组管理与发送的实现都收敛在这里，由 WebSocketServer.group(name)/conn(connId) 返回： + * ws.group("authed").join(conn).send("hello").send(data) // 入组 + 组播（send 三重重载） + * ws.conn(connId).send("hi").send(data) // 单播 + * ws.group("authed").count() / isEmpty() / connIds() // 组信息 + * ws.conn(connId).groups() // 连接所属组 + * + * 组存储（server.connGroups）由 server 持有并负责连接断开自动清组； + * 门面在 server.connLock 保护下直接操作。连接不存在/发送失败静默忽略，单个失败不影响其余。 + */ + +package simcu::websocket.server + +import std.collection.* +import simcu::websocket.common.WebSocketMessage + +/// 判断连接Id是否在列表中（组管理辅助）。 +internal func listHasConnId(list: ArrayList, id: Int64): Bool { + for (x in list) { + if (x == id) { + return true + } + } + false +} + +/// 从列表中移除所有匹配的连接Id（组管理辅助）。 +internal func removeConnId(list: ArrayList, id: Int64): Unit { + var i: Int64 = 0 + while (i < list.size) { + if (list[i] == id) { + list.remove(i..(i + 1)) + } else { + i += 1 + } + } +} + +/// 组门面:由 WebSocketServer.group(name) 返回,支持链式 join/leave/send（send 三重重载：WebSocketMessage / String / Array）与 count/isEmpty/connIds。 +public class WebSocketGroupSender { + private let server: WebSocketServer + private let group: String + + internal init(server: WebSocketServer, group: String) { + this.server = server + this.group = group + } + + /// 向组内所有连接发送消息(链式;组/连接不存在或发送失败均静默忽略)。 + public func send(message: WebSocketMessage): WebSocketGroupSender { + for (id in connIds()) { + if (let Some(c) <- server.getConn(id)) { + try { + c.send(message) + } catch (_) { + () + } + } + } + this + } + + /// 向组内所有连接发送文本(链式)。 + public func send(text: String): WebSocketGroupSender { + send(WebSocketMessage.fromText(text)) + } + + /// 向组内所有连接发送二进制(链式)。 + public func send(data: Array): WebSocketGroupSender { + send(WebSocketMessage.fromBinary(data)) + } + + /// 将连接加入组（组不存在则自动创建；链式）。 + public func join(conn: WebSocketConnection): WebSocketGroupSender { + join(conn.connId) + } + + /// 将连接加入组（按连接Id；链式）。 + public func join(connId: Int64): WebSocketGroupSender { + synchronized(server.connLock) { + if (let Some(l) <- server.connGroups.get(group)) { + if (!listHasConnId(l, connId)) { + l.add(connId) + } + } else { + let l = ArrayList() + l.add(connId) + server.connGroups[group] = l + } + } + this + } + + /// 将连接移出组（组为空则删除组；链式）。 + public func leave(conn: WebSocketConnection): WebSocketGroupSender { + leave(conn.connId) + } + + /// 将连接移出组（按连接Id；链式）。 + public func leave(connId: Int64): WebSocketGroupSender { + synchronized(server.connLock) { + if (let Some(l) <- server.connGroups.get(group)) { + removeConnId(l, connId) + if (l.size == 0) { + server.connGroups.remove(group) + } + } + } + this + } + + /// 组内连接数。 + public func count(): Int64 { + synchronized(server.connLock) { + if (let Some(l) <- server.connGroups.get(group)) { + l.size + } else { + 0 + } + } + } + + /// 组是否为空。 + public func isEmpty(): Bool { + count() == 0 + } + + /// 组内所有连接Id（快照）。 + public func connIds(): ArrayList { + synchronized(server.connLock) { + let copy = ArrayList() + if (let Some(l) <- server.connGroups.get(group)) { + for (x in l) { + copy.add(x) + } + } + copy + } + } +} + +/// 连接门面:由 WebSocketServer.conn(connId) 返回,支持链式 send（三重重载：WebSocketMessage / String / Array）与 groups()。 +public class WebSocketConnSender { + private let server: WebSocketServer + private let connId: Int64 + + internal init(server: WebSocketServer, connId: Int64) { + this.server = server + this.connId = connId + } + + /// 向指定连接发送消息(链式;连接不存在或发送失败均静默忽略)。 + public func send(message: WebSocketMessage): WebSocketConnSender { + if (let Some(c) <- server.getConn(connId)) { + try { + c.send(message) + } catch (_) { + () + } + } + this + } + + /// 向指定连接发送文本(链式)。 + public func send(text: String): WebSocketConnSender { + send(WebSocketMessage.fromText(text)) + } + + /// 向指定连接发送二进制(链式)。 + public func send(data: Array): WebSocketConnSender { + send(WebSocketMessage.fromBinary(data)) + } + + /// 连接所属的所有组名（快照）。 + public func groups(): ArrayList { + synchronized(server.connLock) { + let res = ArrayList() + for ((k, v) in server.connGroups) { + if (listHasConnId(v, connId)) { + res.add(k) + } + } + res + } + } +} diff --git a/src/server/WebSocketServer.cj b/src/server/WebSocketServer.cj index 9f9d66f..5bfc120 100644 --- a/src/server/WebSocketServer.cj +++ b/src/server/WebSocketServer.cj @@ -1,23 +1,23 @@ /* - * simcu::websocket.server —— 事件回调式 WebSocket 服务端（RFC 6455）。 + * simcu::websocket.server —— WebSocket 服务端（RFC 6455），handler 接口式。 * * 用法： - * let server = WebSocketServer(bindAt: 8080) - * server.on("connection", { (conn) => - * println("新连接: ${conn.remoteAddress}") - * conn.on("message", { (msg) => conn.sendText("echo: ${msg.text}") }) - * }) - * server.listen() + * let server = WebSocketServer(8080) + * server.addHandler("/ws") { => MyHandler() } // 路径 -> 工厂，每次握手创建独立 handler 实例 + * server.start() * ... * server.close() // 停止监听并断开所有连接 * - * 服务端事件：on('connection') / on('message') / on('close') / on('error')。 - * 连接方法：send / sendText / sendBinary / ping / close / terminate。 - * 集群方法：broadcast / broadcastText / broadcastBinary / connectionCount。 + * 多 handler 按注册路径路由：握手时按请求路径匹配 addHandler 的 path，匹配则调用工厂 + * 创建一个新的 handler 实例（每连接一个），无匹配返回 404。 + * 连接方法：send / ping / close / terminate（send 三重重载：WebSocketMessage / String / Array）。 + * 组管理（链式）：ws.group(name).join/leave/count/isEmpty/connIds，ws.conn(connId).groups()，ws.groups()。 + * 发送门面（链式）：group(name).send / conn(connId).send（均为三重重载）。 + * 广播：broadcast（三重重载）/ connectionCount。 */ package simcu::websocket.server -import std.collection.ArrayList +import std.collection.{ArrayList, HashMap} import std.net.IPSocketAddress import std.net.SocketException import std.net.TcpServerSocket @@ -25,78 +25,37 @@ import std.net.TcpSocket import std.sync.Mutex import simcu::websocket.common.FrameStream -import simcu::websocket.common.MessageType import simcu::websocket.common.WebSocketMessage import simcu::websocket.common.WebSocketException -import simcu::websocket.common.WsEvents -/// 事件回调式 WebSocket 服务端。 +/// WebSocket 服务端（按路径工厂路由：每连接创建独立 handler 实例，支持组管理/单播/组播/广播）。 public class WebSocketServer { private let bindPort: UInt16 - private let pathFilter: ?String private let maxPayload: Int64 + private var pathFactories = HashMap IWebsocketHandler>() private var serverSock: ?TcpServerSocket = None private var running = false - private var connections = ArrayList() + private var connections = HashMap() + internal var connGroups = HashMap>() private var nextConnId: Int64 = 1 - private let connLock = Mutex() + internal let connLock = Mutex() - /// 服务端事件：有连接完成握手时触发（参数为连接对象）。 - public var onConnection: (WebSocketConnection) -> Unit = { conn => () } - /// 服务端事件：任意连接收到消息时触发（参数为连接对象与消息）。 - public var onMessage: (WebSocketConnection, WebSocketMessage) -> Unit = { conn, msg => () } - /// 服务端事件：任意连接关闭时触发（参数为连接对象、关闭码、原因）。 - public var onClose: (WebSocketConnection, Int64, String) -> Unit = { conn, code, reason => () } - /// 服务端事件：监听或连接运行时错误时触发（参数为异常）。 - public var onError: (Exception) -> Unit = { e => () } - - /// @param bindAt 监听端口，0 表示随机空闲端口（listen 后通过 localPort 读取）。 - /// @param path 仅接受该路径的握手请求（None 表示不限制）。 + /// @param port 监听端口，0 表示随机空闲端口（start 后通过 localPort 读取）。 /// @param maxPayload 单条消息最大字节数，默认 64KB。 - public init(bindAt!: UInt16 = 0, path!: ?String = None, maxPayload!: Int64 = 65536) { - this.bindPort = bindAt - this.pathFilter = path + public init(port: Int64, maxPayload!: Int64 = 65536) { + this.bindPort = UInt16(port) this.maxPayload = maxPayload } - /// 注册服务端事件。支持 "connection"。 - public func on(event: String, listener: (WebSocketConnection) -> Unit): Unit { - if (event == WsEvents.connection) { - onConnection = listener - } else { - throw WebSocketException("不支持的事件: ${event}") - } - } - - /// 注册服务端事件。支持 "message"。 - public func on(event: String, listener: (WebSocketConnection, WebSocketMessage) -> Unit): Unit { - if (event == WsEvents.message) { - onMessage = listener - } else { - throw WebSocketException("不支持的事件: ${event}") - } - } - - /// 注册服务端事件。支持 "close"。 - public func on(event: String, listener: (WebSocketConnection, Int64, String) -> Unit): Unit { - if (event == WsEvents.close) { - onClose = listener - } else { - throw WebSocketException("不支持的事件: ${event}") - } - } - - /// 注册服务端事件。支持 "error"。 - public func on(event: String, listener: (Exception) -> Unit): Unit { - if (event == WsEvents.error) { - onError = listener - } else { - throw WebSocketException("不支持的事件: ${event}") - } + /// 注册一个 handler 工厂并绑定到端点路径（多 handler 按路径路由）。 + /// 每次握手成功都会调用工厂创建一个新的 handler 实例（每连接一个）， + /// 因此不要在工厂里复用有状态对象（如 scoped 服务），应让各回调按需现场解析。 + public func addHandler(path: String, factory: () -> IWebsocketHandler): Unit { + pathFactories[path] = factory } /// 启动监听：bind + accept 循环，每连接一个协程处理握手与读循环。 - public func listen(backlog!: Int64 = 128): Unit { + public func start(backlog!: Int64 = 128): Unit { if (running) { throw WebSocketException("服务端已在监听") } @@ -129,7 +88,7 @@ public class WebSocketServer { } } - /// 停止监听并强制断开所有连接（每连接会触发自身 close 事件）。 + /// 停止监听并强制断开所有连接（每连接会触发自身 close 回调）。 public func close(): Unit { running = false if (let Some(s) <- serverSock) { @@ -158,33 +117,44 @@ public class WebSocketServer { } } - /// 广播文本消息。 - public func broadcastText(text: String): Unit { + /// 广播文本消息（三重重载：WebSocketMessage / String / Array）。 + public func broadcast(text: String): Unit { broadcast(WebSocketMessage.fromText(text)) } /// 广播二进制消息。 - public func broadcastBinary(data: Array): Unit { + public func broadcast(data: Array): Unit { broadcast(WebSocketMessage.fromBinary(data)) } - // ===== 内部回调转发（供 WebSocketConnection 调用）===== - - internal func fireMessage(conn: WebSocketConnection, msg: WebSocketMessage): Unit { - onMessage(conn, msg) + /// 所有组名（快照）。 + public func groups(): ArrayList { + synchronized(connLock) { + let copy = ArrayList() + for ((k, _) in connGroups) { + copy.add(k) + } + copy + } } - internal func fireClose(conn: WebSocketConnection, code: Int64, reason: String): Unit { - onClose(conn, code, reason) + // ===== 链式发送门面 ===== + + /// 组门面(链式): ws.group("authed").join(conn).send("hi").send(data)；count/isEmpty/connIds 读组信息。 + public func group(name: String): WebSocketGroupSender { + WebSocketGroupSender(this, name) } - internal func fireError(e: Exception): Unit { - onError(e) + /// 连接门面(链式): ws.conn(connId).send("hi").send(data)；groups() 查连接所属组。 + public func conn(connId: Int64): WebSocketConnSender { + WebSocketConnSender(this, connId) } + // ===== 内部连接管理 ===== + internal func addConnection(conn: WebSocketConnection): Unit { synchronized(connLock) { - connections.add(conn) + connections[conn.connId] = conn } } @@ -198,23 +168,36 @@ public class WebSocketServer { internal func removeConnection(conn: WebSocketConnection): Unit { synchronized(connLock) { - var idx: Int64 = -1 - for (i in 0..connections.size) { - if (connections[i].connId == conn.connId) { - idx = i - break + connections.remove(conn.connId) + removeFromAllGroups(conn.connId) + } + } + + internal func getConn(connId: Int64): ?WebSocketConnection { + synchronized(connLock) { + connections.get(connId) + } + } + + private func removeFromAllGroups(connId: Int64): Unit { + var toRemove = ArrayList() + for ((g, list) in connGroups) { + if (listHasConnId(list, connId)) { + removeConnId(list, connId) + if (list.size == 0) { + toRemove.add(g) } } - if (idx >= 0) { - connections.remove(at: idx) - } + } + for (g in toRemove) { + connGroups.remove(g) } } private func snapshotConnections(): ArrayList { synchronized(connLock) { let copy = ArrayList() - for (c in connections) { + for (c in connections.values()) { copy.add(c) } copy @@ -223,6 +206,11 @@ public class WebSocketServer { // ===== 监听与握手 ===== + /// 按请求路径匹配 handler 工厂。无匹配返回 None。 + private func findFactory(path: String): ?() -> IWebsocketHandler { + pathFactories.get(path) + } + private func acceptLoop(): Unit { while (running) { try { @@ -237,12 +225,12 @@ public class WebSocketServer { if (!running) { return } - onError(e) + println("[websocket] accept 失败: ${e}") } catch (e: Exception) { if (!running) { return } - onError(e) + println("[websocket] accept 失败: ${e}") } } } @@ -261,11 +249,19 @@ public class WebSocketServer { reject(sock, 400, "Bad Request") return } - if (let Some(p) <- pathFilter) { - if (parts[1] != p) { - reject(sock, 404, "Not Found") - return - } + // 拆出路径与 query（如 "/ws/server?serverId=srv-1" → path="/ws/server", query="serverId=srv-1"） + let target = parts[1] + var requestPath = target + var requestQuery = "" + if (let Some(qi) <- target.indexOf("?")) { + requestPath = target[0..qi] + requestQuery = target[qi + 1..] + } + // 按 WsPath 路由到 handler 工厂，匹配则调用工厂创建新的 handler 实例（每连接一个） + let factory = findFactory(requestPath) + if (let None <- factory) { + reject(sock, 404, "Not Found") + return } var upgradeOk = false var connectionOk = false @@ -302,13 +298,15 @@ public class WebSocketServer { sock.write(resp.toString().toArray()) sock.flush() - let conn = WebSocketConnection(Some(this), fs, allocateConnId()) + let fac = factory.getOrThrow() + let h = fac() + let conn = WebSocketConnection(Some(this), h, fs, allocateConnId(), requestPath, requestQuery) addConnection(conn) - onConnection(conn) + h.onConnect(conn) conn.runReadLoop() } catch (e: Exception) { - // 握手阶段失败：上报服务端 error 事件 - onError(e) + // 握手阶段失败：记录日志（连接尚未建立，无 handler 可回调） + println("[websocket] 握手失败: ${e}") } } diff --git a/src/tests/websocket_test.cj b/src/tests/websocket_test.cj index 1f50443..52a1c88 100644 --- a/src/tests/websocket_test.cj +++ b/src/tests/websocket_test.cj @@ -1,7 +1,8 @@ /* * simcu::websocket.tests —— 端到端集成测试（cjpm test）。 * 覆盖：文本回声 / 广播 / 关闭握手（客户端主动、服务端主动）/ terminate / - * ping 心跳后连接仍可用 / 路径过滤 / maxPayload 超限(1009)。 + * ping 心跳后连接仍可用 / 路径路由（多 handler 按注册路径）/ maxPayload 超限(1009) / + * 组播（group.send）/ 单播（conn.send）。 * * 运行：在 websocket-cj 目录执行 cjpm test */ @@ -17,6 +18,7 @@ import simcu::websocket.client.WebSocketClient import simcu::websocket.common.MessageType import simcu::websocket.common.WebSocketException import simcu::websocket.common.WebSocketMessage +import simcu::websocket.server.IWebsocketHandler import simcu::websocket.server.WebSocketConnection import simcu::websocket.server.WebSocketServer @@ -45,27 +47,115 @@ private func waitUntil(timeoutMs: Int64, condition: () -> Bool): Bool { condition() } +/// 测试用 handler：把事件回调逻辑收进接口方法，行为写入 harness。 +class TestWsHandler <: IWebsocketHandler { + let harness: WsTestHarness + + init(harness: WsTestHarness) { + this.harness = harness + } + + public override func onConnect(conn: WebSocketConnection): Unit { + harness.serverConn = Some(conn) + } + + public override func onMessage(conn: WebSocketConnection, msg: WebSocketMessage): Unit { + harness.connMessage = msg.text.getOrThrow() + conn.send("echo: ${msg.text.getOrThrow()}") + } + + public override func onClose(conn: WebSocketConnection, code: Int64, reason: String): Unit { + harness.connCloseCode = code + harness.connCloseReason = reason + } + + public override func onError(conn: WebSocketConnection, e: Exception): Unit { + if (let we: WebSocketException <- e) { + harness.connErrorCode = we.code + } + } +} + +/// 广播场景：harness1 收到广播后转发文本。 +class EchoHandler <: IWebsocketHandler { + let harness: WsTestHarness + + init(harness: WsTestHarness) { + this.harness = harness + } + + public override func onConnect(conn: WebSocketConnection): Unit { + () + } + + public override func onMessage(conn: WebSocketConnection, msg: WebSocketMessage): Unit { + harness.connMessage = msg.text.getOrThrow() + } + + public override func onClose(conn: WebSocketConnection, code: Int64, reason: String): Unit { + () + } + + public override func onError(conn: WebSocketConnection, e: Exception): Unit { + () + } +} + +/// 服务端主动关闭场景：连接建立后立即 close。 +class CloseOnConnectHandler <: IWebsocketHandler { + init() {} + + public override func onConnect(conn: WebSocketConnection): Unit { + conn.close(code: 4000, reason: "svr") + } + + public override func onMessage(conn: WebSocketConnection, msg: WebSocketMessage): Unit { + () + } + + public override func onClose(conn: WebSocketConnection, code: Int64, reason: String): Unit { + () + } + + public override func onError(conn: WebSocketConnection, e: Exception): Unit { + () + } +} + +/// 分组场景：连接建立后加入 "room" 组，便于验证组播/单播。 +class GroupJoinHandler <: IWebsocketHandler { + public override func onConnect(conn: WebSocketConnection): Unit { + conn.joinGroup("room") + } + + public override func onMessage(conn: WebSocketConnection, msg: WebSocketMessage): Unit { + () + } + + public override func onClose(conn: WebSocketConnection, code: Int64, reason: String): Unit { + () + } + + public override func onError(conn: WebSocketConnection, e: Exception): Unit { + () + } +} + @Test public class WebSocketTest { - /// 1. 文本回声：client.sendText → server 收到后回发 → client 收到 + /// 1. 文本回声：client.send → server 收到后回发 → client 收到 @TestCase public func echoText(): Unit { let h = WsTestHarness() - let server = WebSocketServer(bindAt: 0) - server.on("connection", { conn: WebSocketConnection => - h.serverConn = Some(conn) - conn.on("message", { m: WebSocketMessage => - h.connMessage = m.text.getOrThrow() - conn.sendText("echo: ${m.text.getOrThrow()}") - }) - }) - server.listen() - let client = WebSocketClient("127.0.0.1", server.localPort) + let server = WebSocketServer(0) + server.addHandler("/ws") { => TestWsHandler(h) } + server.start() + let client = WebSocketClient("127.0.0.1", server.localPort, path: "/ws") client.onMessage = { m: WebSocketMessage => h.clientMessage = m.text.getOrThrow() } client.connect() - client.sendText("hello") + client.send("hello") @Expect(waitUntil(5000) { => h.connMessage == Some("hello") }, true) @Expect(waitUntil(5000) { => h.clientMessage == Some("echo: hello") }, true) @@ -73,27 +163,28 @@ public class WebSocketTest { server.close() } - /// 2. 广播：server.broadcastText 两个客户端都收到 + /// 2. 广播：server.broadcast 两个客户端都收到 @TestCase public func broadcast(): Unit { let h1 = WsTestHarness() let h2 = WsTestHarness() - let server = WebSocketServer(bindAt: 0) - server.listen() + let server = WebSocketServer(0) + server.addHandler("/ws") { => EchoHandler(h1) } + server.start() - let c1 = WebSocketClient("127.0.0.1", server.localPort) + let c1 = WebSocketClient("127.0.0.1", server.localPort, path: "/ws") c1.onMessage = { m: WebSocketMessage => h1.clientMessage = m.text.getOrThrow() } c1.connect() - let c2 = WebSocketClient("127.0.0.1", server.localPort) + let c2 = WebSocketClient("127.0.0.1", server.localPort, path: "/ws") c2.onMessage = { m: WebSocketMessage => h2.clientMessage = m.text.getOrThrow() } c2.connect() @Expect(waitUntil(5000) { => server.connectionCount == 2 }, true) - server.broadcastText("hi-all") + server.broadcast("hi-all") @Expect(waitUntil(5000) { => h1.clientMessage == Some("hi-all") }, true) @Expect(waitUntil(5000) { => h2.clientMessage == Some("hi-all") }, true) c1.close() @@ -105,15 +196,10 @@ public class WebSocketTest { @TestCase public func closeHandshake(): Unit { let h = WsTestHarness() - let server = WebSocketServer(bindAt: 0) - server.on("connection", { conn: WebSocketConnection => - conn.on("close", { code: Int64, reason: String => - h.connCloseCode = code - h.connCloseReason = reason - }) - }) - server.listen() - let client = WebSocketClient("127.0.0.1", server.localPort) + let server = WebSocketServer(0) + server.addHandler("/ws") { => TestWsHandler(h) } + server.start() + let client = WebSocketClient("127.0.0.1", server.localPort, path: "/ws") client.onClose = { code: Int64, reason: String => h.clientCloseCode = code h.clientCloseReason = reason @@ -130,13 +216,10 @@ public class WebSocketTest { @TestCase public func serverClose(): Unit { let h = WsTestHarness() - let server = WebSocketServer(bindAt: 0) - server.on("connection", { conn: WebSocketConnection => - h.serverConn = Some(conn) - conn.close(code: 4000, reason: "svr") - }) - server.listen() - let client = WebSocketClient("127.0.0.1", server.localPort) + let server = WebSocketServer(0) + server.addHandler("/ws") { => CloseOnConnectHandler() } + server.start() + let client = WebSocketClient("127.0.0.1", server.localPort, path: "/ws") client.onClose = { code: Int64, reason: String => h.clientCloseCode = code h.clientCloseReason = reason @@ -151,15 +234,10 @@ public class WebSocketTest { @TestCase public func terminate(): Unit { let h = WsTestHarness() - let server = WebSocketServer(bindAt: 0) - server.on("connection", { conn: WebSocketConnection => - conn.on("close", { code: Int64, reason: String => - h.connCloseCode = code - h.connCloseReason = reason - }) - }) - server.listen() - let client = WebSocketClient("127.0.0.1", server.localPort) + let server = WebSocketServer(0) + server.addHandler("/ws") { => TestWsHandler(h) } + server.start() + let client = WebSocketClient("127.0.0.1", server.localPort, path: "/ws") client.connect() client.terminate() @@ -171,29 +249,27 @@ public class WebSocketTest { @TestCase public func ping(): Unit { let h = WsTestHarness() - let server = WebSocketServer(bindAt: 0) - server.on("connection", { conn: WebSocketConnection => - conn.on("message", { m: WebSocketMessage => - h.connMessage = m.text.getOrThrow() - }) - }) - server.listen() - let client = WebSocketClient("127.0.0.1", server.localPort) + let server = WebSocketServer(0) + server.addHandler("/ws") { => TestWsHandler(h) } + server.start() + let client = WebSocketClient("127.0.0.1", server.localPort, path: "/ws") client.connect() client.ping() sleep(Duration.millisecond * 100) - client.sendText("after-ping") + client.send("after-ping") @Expect(waitUntil(5000) { => h.connMessage == Some("after-ping") }, true) client.close() server.close() } - /// 7. 路径过滤：path=Some("/ws") 时 /ws 握手成功，/other 被拒（connect 抛异常） + /// 7. 路径路由：注册 /ws 与 /admin 两个 handler，/ws 握手成功，/other 被拒（connect 抛异常） @TestCase - public func pathFilter(): Unit { - let server = WebSocketServer(bindAt: 0, path: Some("/ws")) - server.listen() + public func pathRoute(): Unit { + let server = WebSocketServer(0) + server.addHandler("/ws") { => TestWsHandler(WsTestHarness()) } + server.addHandler("/admin") { => TestWsHandler(WsTestHarness()) } + server.start() let ok = WebSocketClient("127.0.0.1", server.localPort, path: "/ws") ok.connect() @@ -214,24 +290,89 @@ public class WebSocketTest { @TestCase public func maxPayload(): Unit { let h = WsTestHarness() - let server = WebSocketServer(bindAt: 0, maxPayload: 10) - server.on("connection", { conn: WebSocketConnection => - conn.on("error", { e: Exception => - if (let we: WebSocketException <- e) { - h.connErrorCode = we.code - } - }) - }) - server.listen() - let client = WebSocketClient("127.0.0.1", server.localPort) + let server = WebSocketServer(0, maxPayload: 10) + server.addHandler("/ws") { => TestWsHandler(h) } + server.start() + let client = WebSocketClient("127.0.0.1", server.localPort, path: "/ws") client.onError = { e: Exception => h.clientErrorCount += 1 } client.connect() - client.sendText("this message is way too long") + client.send("this message is way too long") @Expect(waitUntil(5000) { => h.connErrorCode == 1009 }, true) client.terminate() server.close() } + + /// 9. 组播：两个连接加入同组（"room"），group("room").send 两连接都收到；关闭一个后组计数减一 + @TestCase + public func groupMulticast(): Unit { + let h1 = WsTestHarness() + let h2 = WsTestHarness() + let server = WebSocketServer(0) + server.addHandler("/ws") { => GroupJoinHandler() } + server.start() + + let c1 = WebSocketClient("127.0.0.1", server.localPort, path: "/ws") + c1.onMessage = { m: WebSocketMessage => + h1.clientMessage = m.text.getOrThrow() + } + c1.connect() + let c2 = WebSocketClient("127.0.0.1", server.localPort, path: "/ws") + c2.onMessage = { m: WebSocketMessage => + h2.clientMessage = m.text.getOrThrow() + } + c2.connect() + @Expect(waitUntil(5000) { => server.group("room").count() == 2 }, true) + + server.group("room").send("team-msg") + @Expect(waitUntil(5000) { => h1.clientMessage == Some("team-msg") }, true) + @Expect(waitUntil(5000) { => h2.clientMessage == Some("team-msg") }, true) + + // 关闭一个连接后自动退组，组计数减一 + c1.close() + @Expect(waitUntil(5000) { => server.group("room").count() == 1 }, true) + server.group("room").send("after-leave") + @Expect(waitUntil(5000) { => h2.clientMessage == Some("after-leave") }, true) + c2.close() + server.close() + } + + /// 10. 单播：conn(connId).send 只发到指定连接Id，另一连接收不到 + @TestCase + public func unicast(): Unit { + let h1 = WsTestHarness() + let h2 = WsTestHarness() + let server = WebSocketServer(0) + server.addHandler("/ws") { => GroupJoinHandler() } + server.start() + + let c1 = WebSocketClient("127.0.0.1", server.localPort, path: "/ws") + c1.onMessage = { m: WebSocketMessage => + h1.clientMessage = m.text.getOrThrow() + } + c1.connect() + let c2 = WebSocketClient("127.0.0.1", server.localPort, path: "/ws") + c2.onMessage = { m: WebSocketMessage => + h2.clientMessage = m.text.getOrThrow() + } + c2.connect() + @Expect(waitUntil(5000) { => server.group("room").connIds().size == 2 }, true) + + h1.clientMessage = None + h2.clientMessage = None + let ids = server.group("room").connIds() + let target = ids[0] + server.conn(target).send("only-you") + @Expect(waitUntil(5000) { => h1.clientMessage == Some("only-you") || h2.clientMessage == Some("only-you") }, true) + // 稍等片刻后断言：仅目标接收，另一连接未收到 + sleep(Duration.millisecond * 200) + let exactlyOne = (h1.clientMessage == Some("only-you") && h2.clientMessage != Some("only-you")) || + (h2.clientMessage == Some("only-you") && h1.clientMessage != Some("only-you")) + @Expect(exactlyOne, true) + c1.close() + c2.close() + server.close() + } }