From 3ca5aff5169df2e1f69839388d84db2cf7a56594 Mon Sep 17 00:00:00 2001 From: xRain Date: Sat, 22 Aug 2026 06:14:48 +0800 Subject: [PATCH] =?UTF-8?q?=E5=88=9D=E5=A7=8B=E7=89=88=E6=9C=AC=EF=BC=9A?= =?UTF-8?q?=E4=BA=8B=E4=BB=B6=E5=9B=9E=E8=B0=83=E5=BC=8F=20WebSocket=20?= =?UTF-8?q?=E5=AE=A2=E6=88=B7=E7=AB=AF=E4=B8=8E=E6=9C=8D=E5=8A=A1=E7=AB=AF?= =?UTF-8?q?=EF=BC=88RFC=206455=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 10 + README.md | 220 ++++++++++++++++++++ cjpm.lock | 3 + cjpm.toml | 24 +++ src/client/WebSocketClient.cj | 258 +++++++++++++++++++++++ src/common/FrameStream.cj | 331 ++++++++++++++++++++++++++++++ src/common/WsTypes.cj | 145 +++++++++++++ src/server/WebSocketConnection.cj | 228 ++++++++++++++++++++ src/server/WebSocketServer.cj | 329 +++++++++++++++++++++++++++++ src/tests/websocket_test.cj | 237 +++++++++++++++++++++ src/websocket.cj | 14 ++ 11 files changed, 1799 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 cjpm.lock create mode 100644 cjpm.toml create mode 100644 src/client/WebSocketClient.cj create mode 100644 src/common/FrameStream.cj create mode 100644 src/common/WsTypes.cj create mode 100644 src/server/WebSocketConnection.cj create mode 100644 src/server/WebSocketServer.cj create mode 100644 src/tests/websocket_test.cj create mode 100644 src/websocket.cj diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..02fe63a --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +# 构建产物 +target/ +.cache/ + +# IDE / 系统 +.idea/ +.vscode/ +*.iml +.DS_Store +Thumbs.db diff --git a/README.md b/README.md new file mode 100644 index 0000000..13af6b3 --- /dev/null +++ b/README.md @@ -0,0 +1,220 @@ +# simcu::websocket — 仓颉 WebSocket 库（RFC 6455） + +事件回调式 WebSocket 客户端与服务端，纯仓颉实现，仅依赖标准库 + stdx（无第三方库）。 + +- 组织：`simcu` +- 包名：`websocket` +- 子包：`simcu::websocket.client` / `simcu::websocket.server` / `simcu::websocket.common` +- 构建：`cjpm build`；测试：`cjpm test`（8 个端到端用例全通过） + +## 功能特性 + +- **客户端**：`connect()` / `send()` / `close()` / `terminate()` + 事件回调 `onOpen` / `onMessage` / `onError` / `onClose` +- **服务端**：`on('connection')` / `on('message')` / `send()` / `close()` / `terminate()` + 监听 `on('close')` / `on('error')` +- **RFC 6455 完整帧层**：掩码、分片重组、Ping/Pong、Close 握手、`maxPayload` 上限（超限 1009） +- **群发广播**：`broadcastText` / `broadcastBinary` +- **wss（TLS）**：`FrameStream` 基于 `StreamingSocket`，可直接包裹 `TlsSocket`（已在 BotRoleHelper 实战验证） + +## 依赖引入 + +在项目 `cjpm.toml` 中添加： + +```toml +[dependencies] +"simcu::websocket" = { path = "../websocket-cj" } +``` + +```cangjie +import simcu::websocket.client.WebSocketClient +import simcu::websocket.server.WebSocketServer +import simcu::websocket.server.WebSocketConnection +import simcu::websocket.common.WebSocketMessage +``` + +## 目录结构 + +``` +src/ +├── websocket.cj # 模块锚点 +├── common/ +│ ├── WsTypes.cj # WebSocketMessage / ReadyState / WebSocketException / CloseCodes / WsEvents +│ └── FrameStream.cj # RFC 6455 帧层（握手工具、掩码、分片重组、maxPayload） +├── client/ +│ └── WebSocketClient.cj # 客户端 +├── server/ +│ ├── WebSocketServer.cj # 服务端监听 / 握手 / 广播 +│ └── WebSocketConnection.cj # 服务端单条连接 +└── tests/ + └── websocket_test.cj # 端到端集成测试 +``` + +## 客户端 API + +```cangjie +let client = WebSocketClient("127.0.0.1", 8080, path: "/ws", maxPayload: 65536) + +// 事件回调（连接建立后触发） +client.onOpen = { => ... } // 握手完成 +client.onMessage = { m: WebSocketMessage => ... } // 收到消息（m.text / m.bytes / m.type） +client.onError = { e: Exception => ... } // 运行时错误（WebSocketException 含 .code） +client.onClose = { code: Int64, reason: String => ... } // 关闭，携带关闭码与原因 + +client.connect(timeout: Duration.second * 30) // 建立连接；握手失败抛 WebSocketException +client.send(WebSocketMessage) // 发送消息对象 +client.sendText("hello") // 发送文本 +client.sendBinary(byteArray) // 发送二进制 +client.ping() // 心跳（服务端自动回 Pong） +client.close(code: 1000, reason: "bye") // 关闭握手，5 秒兜底强制断开 +client.terminate() // 立即断开（对端收 1006） + +client.readyState // Connecting / Open / Closing / Closed +client.isOpen() +``` + +构造参数： + +| 参数 | 默认 | 说明 | +|---|---|---| +| `host: String` | — | 服务端主机（IP 或域名） | +| `port: UInt16` | — | 服务端端口 | +| `path: String` | `"/"` | 请求路径 | +| `maxPayload: Int64` | `65536` | 单条消息最大字节数，超出触发 `onError(code=1009)` | + +## 服务端 API + +```cangjie +let server = WebSocketServer(bindAt: 8080, path: Some("/ws"), maxPayload: 65536) + +// 服务端事件 +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 // 当前连接数 +``` + +构造参数： + +| 参数 | 默认 | 说明 | +|---|---|---| +| `bindAt: UInt16` | `0` | 监听端口，`0` 表示随机空闲端口（`listen` 后读 `localPort`） | +| `path: ?String` | `None` | 仅接受该路径的握手请求，`None` 表示不限制 | +| `maxPayload: Int64` | `65536` | 单条消息最大字节数 | + +### 连接级 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.readyState / conn.isOpen() +conn.remoteAddress // "ip:port" +``` + +## 底层帧 API（common 子包） + +高层 client/server 已封装全部帧逻辑；需要自定义握手、代理或 wss 包裹时，可直接使用 `FrameStream`（lurmix 的 BotRoleHelper 即基于它实现）： + +```cangjie +import simcu::websocket.common.FrameStream + +// 构造：包裹 TCP（或 TLS）流；客户端必须 maskOutgoing=true，服务端 false +let fs = FrameStream(conn, maxPayload, true) + +// 握手工具 +FrameStream.generateWebSocketKey() // 生成 Sec-WebSocket-Key +FrameStream.computeAccept(key) // 计算 Sec-WebSocket-Accept +FrameStream.parseClosePayload(payload) // 解析 close 帧 -> (code, reason) + +// 读写 +fs.readHttpHeader() // 读 HTTP 头（到 \r\n\r\n），帧字节保留在缓冲 +fs.readMessage() // 读一条完整消息：分片重组、自动回 Pong；返回 ?WsFrame +fs.writeText(text) / fs.writeBinary(data) +fs.writeFrame(opcode, payload) // 发送任意 opcode 帧 +fs.writePing(payload) / fs.writePong(payload) +fs.writeClose(code, reason) +fs.close() // 关闭底层流 +fs.remoteAddress() // 远端 SocketAddress +``` + +`WsFrame` 字段：`fin: Bool`、`opcode: Int64`（1=text，2=binary，8=close）、`payload: Array`。 +`readMessage()` 返回 `None` 表示对端关闭/EOF；收到 close 帧时已自动回写 close 并关闭流。 + +### wss（TLS）示例 + +```cangjie +import stdx.net.tls.* +import stdx.net.tls.common.* + +let tcp = TcpSocket(host, port) +tcp.connect(timeout: Duration.second * 30) +var tls = TlsClientConfig() +tls.verifyMode = CertificateVerifyMode.TrustAll +tls.serverName = Some(host) +let tlsSocket = TlsSocket.client(tcp, session: None, clientConfig: tls) +tlsSocket.handshake(timeout: Duration.second * 30) +let fs = FrameStream(tlsSocket, maxPayload, true) // 客户端掩码 + TLS 包裹 = wss +``` + +完整实战示例见 `lemon-lurmix-cj/src/helpers/BotRoleHelper.cj`。 + +## 消息与错误码 + +```cangjie +// WebSocketMessage +m.type // MessageType.Text / Binary +m.bytes // 原始 payload +m.text // 文本内容（二进制消息返回 None） +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 +``` + +## 测试覆盖 + +| 用例 | 验证点 | +|---|---| +| echoText | 客户端发送 → 服务端回声 → 客户端收到 | +| 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 拒绝 | +| maxPayload | 超限触发 onError(code=1009) | + +## 已补齐的 API 缺口 + +对照最初需求（connect/send/close + onOpen/onMessage/onError/onClose），补齐了： + +- **sendText / sendBinary**：按文本 / 二进制发送的重载（`send(WebSocketMessage)` 之外） +- **ping()**：心跳保活 +- **close(code, reason)**：携带关闭码与原因的关闭握手；onClose 回调带 (code, reason) +- **readyState / isOpen()**：连接状态查询 +- **broadcastText / broadcastBinary**：服务端群发 +- **localPort / connectionCount / remoteAddress**：服务端与连接元信息 +- **服务端 path 过滤**：仅接受指定路径的握手 +- **maxPayload**：单条消息上限，超限报 1009 + +## 尚未实现（后续方向） + +- **子协议协商（Sec-WebSocket-Protocol）**：握手阶段携带子协议并校验 +- **自动定时心跳**：目前为手动 `ping()`，未内置保活定时器 +- **分片发送**：接收侧已支持分片重组；发送侧未提供手动分片 API +- **压缩扩展（permessage-deflate）** diff --git a/cjpm.lock b/cjpm.lock new file mode 100644 index 0000000..c42311c --- /dev/null +++ b/cjpm.lock @@ -0,0 +1,3 @@ +version = 0 + +[requires] diff --git a/cjpm.toml b/cjpm.toml new file mode 100644 index 0000000..13854aa --- /dev/null +++ b/cjpm.toml @@ -0,0 +1,24 @@ +[package] +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" +target-dir = "" +output-type = "static" + +[dependencies] + +[target] +[target.x86_64-w64-mingw32] +compile-option = "-Woff unused --diagnostic-format=noColor -Woff deprecated" +[target.x86_64-w64-mingw32.bin-dependencies] +path-option = ["${CANGJIE_STDX_PATH}"] +[target.x86_64-unknown-linux-gnu] +compile-option = "-Woff unused --diagnostic-format=noColor -Woff deprecated" +[target.x86_64-unknown-linux-gnu.bin-dependencies] +path-option = ["${CANGJIE_STDX_PATH}"] +[target.aarch64-unknown-linux-gnu] +compile-option = "-Woff unused --diagnostic-format=noColor -Woff deprecated" +[target.aarch64-unknown-linux-gnu.bin-dependencies] +path-option = ["${CANGJIE_STDX_PATH}"] diff --git a/src/client/WebSocketClient.cj b/src/client/WebSocketClient.cj new file mode 100644 index 0000000..4eb60bb --- /dev/null +++ b/src/client/WebSocketClient.cj @@ -0,0 +1,258 @@ +/* + * simcu::websocket.client —— 事件回调式 WebSocket 客户端（RFC 6455）。 + * + * 用法： + * let client = WebSocketClient("127.0.0.1", 8080) + * client.onOpen = { println("已连接") } + * client.onMessage = { (m) => println(m.text) } + * client.onClose = { (code, reason) => println("关闭: ${code} ${reason}") } + * client.connect() + * client.sendText("hello") + * client.close() // 或 client.terminate() 立即断开 + * + * 说明： + * - connect() 抛出的异常表示握手失败；连接建立后的错误走 onError。 + * - close(code, reason) 发起 RFC 6455 关闭握手（等待对端回 close 帧，5 秒兜底）； + * terminate() 直接断开底层连接。 + * - 回调在后台读线程触发，禁止在回调中直接做耗时操作。 + */ +package simcu::websocket.client + +import std.net.TcpSocket +import std.time.* +import std.sync.Mutex + +import simcu::websocket.common.FrameStream +import simcu::websocket.common.MessageType +import simcu::websocket.common.ReadyState +import simcu::websocket.common.WebSocketMessage +import simcu::websocket.common.WebSocketException + +/// 事件回调式 WebSocket 客户端。 +public class WebSocketClient { + private let host: String + private let port: UInt16 + private let path: String + private let maxPayload: Int64 + private var frame: ?FrameStream = None + private var state: ReadyState = ReadyState.Closed + private let stateLock = Mutex() + + /// 连接建立（握手完成）后触发，无参数。 + public var onOpen: () -> Unit = { => () } + /// 收到文本/二进制消息时触发，参数为消息。 + public var onMessage: (WebSocketMessage) -> Unit = { m => () } + /// 连接建立后的运行时错误触发，参数为异常。 + public var onError: (Exception) -> Unit = { e => () } + /// 连接关闭时触发，参数为 (关闭码, 关闭原因)。 + public var onClose: (Int64, String) -> Unit = { c, r => () } + + /// @param host 服务端主机（IP 或域名）。 + /// @param port 服务端端口。 + /// @param path 请求路径，默认 "/"。 + /// @param maxPayload 单条消息最大字节数，默认 64KB，超出触发 onError(code=1009)。 + public init(host: String, port: UInt16, path!: String = "/", maxPayload!: Int64 = 65536) { + this.host = host + this.port = port + this.path = path + this.maxPayload = maxPayload + } + + /// 当前连接状态（Connecting/Open/Closing/Closed）。 + public prop readyState: ReadyState { + get() { state } + } + + /// 连接是否处于 Open。 + public func isOpen(): Bool { + state == ReadyState.Open + } + + /// 建立连接：TCP + WebSocket 握手。 + /// 成功 → 状态 Open 并触发 onOpen，随后后台线程持续读取消息。 + /// 失败（网络/握手被拒）→ 抛出 WebSocketException。 + public func connect(timeout!: ?Duration = None): Unit { + if (state != ReadyState.Closed) { + throw WebSocketException("重复 connect：当前 readyState=${state}") + } + state = ReadyState.Connecting + let sock = TcpSocket(host, port) + sock.connect(timeout: timeout) + let fs = FrameStream(sock, maxPayload, true) + + // 发送握手请求 + let key = FrameStream.generateWebSocketKey() + let req = StringBuilder() + req.append("GET ${path} HTTP/1.1\r\n") + req.append("Host: ${host}:${port}\r\n") + req.append("Upgrade: websocket\r\n") + req.append("Connection: Upgrade\r\n") + req.append("Sec-WebSocket-Key: ${key}\r\n") + req.append("Sec-WebSocket-Version: 13\r\n\r\n") + sock.write(req.toString().toArray()) + sock.flush() + + // 读取并校验握手响应 + let response = fs.readHttpHeader() + if (!validateHandshake(response, key)) { + fs.close() + sock.close() + state = ReadyState.Closed + throw WebSocketException("WebSocket 握手失败（服务端拒绝升级）") + } + + frame = Some(fs) + state = ReadyState.Open + onOpen() + spawn { => readLoop() } + } + + /// 发送消息（按消息类型自动选择 text/binary 帧）。 + public func send(message: WebSocketMessage): Unit { + let f = frame.getOrThrow() + if (message.`type` == MessageType.Text) { + f.writeFrame(1, message.bytes) + } else { + f.writeFrame(2, message.bytes) + } + } + + /// 发送文本消息。 + public func sendText(text: String): Unit { + frame.getOrThrow().writeText(text) + } + + /// 发送二进制消息。 + public func sendBinary(data: Array): Unit { + frame.getOrThrow().writeBinary(data) + } + + /// 发送 Ping（心跳保活），对端 Pong 由帧层自动忽略。 + public func ping(): Unit { + frame.getOrThrow().writePing(Array(0, repeat: 0)) + } + + /// 发送 Ping（携带 payload）。 + public func ping(payload: Array): Unit { + frame.getOrThrow().writePing(payload) + } + + /// 发起关闭握手：发送 close 帧（code + reason），等待对端回执后触发 onClose。 + /// 对端 5 秒内不回执则强制断开。 + public func close(code!: Int64 = 1000, reason!: String = ""): Unit { + synchronized(stateLock) { + if (state != ReadyState.Open) { + return + } + state = ReadyState.Closing + } + if (let Some(f) <- frame) { + try { + f.writeClose(code, reason) + } catch (_) { + terminate() + return + } + } + spawn { => + sleep(Duration.second * 5) + if (state == ReadyState.Closing) { + terminate() + } + } + } + + /// 立即断开底层连接（不发送 close 帧），读线程会触发 onClose(1006, "")。 + public func terminate(): Unit { + synchronized(stateLock) { + if (state == ReadyState.Closed) { + return + } + state = ReadyState.Closed + } + if (let Some(f) <- frame) { + try { + f.close() + } catch (_) { + () + } + } + } + + /// 后台读循环：数据帧 → onMessage；close 帧 → 回执并 onClose；EOF/异常 → onError(仅异常)+onClose。 + private func readLoop(): Unit { + var code: Int64 = 1006 + var reason: String = "" + try { + while (true) { + let m = frame.getOrThrow().readMessage() + match (m) { + case None => + break // EOF：对端直接断连 + case Some(f) => + if (f.opcode == 8) { + let (c, r) = FrameStream.parseClosePayload(f.payload) + code = c + reason = r + if (state == ReadyState.Open) { + try { + frame.getOrThrow().writeClose(c, r) + } catch (_) { + () + } + } + break + } else { + let msg = if (f.opcode == 1) { + WebSocketMessage(MessageType.Text, f.payload) + } else { + WebSocketMessage(MessageType.Binary, f.payload) + } + onMessage(msg) + } + } + } + } catch (e: Exception) { + code = 1006 + if (state != ReadyState.Closing) { + onError(e) + } + } + synchronized(stateLock) { + state = ReadyState.Closed + } + if (let Some(f) <- frame) { + try { + f.close() + } catch (_) { + () + } + } + onClose(code, reason) + } + + /// 校验服务端握手响应：状态行 101 + Upgrade/Connection/Accept 头。 + private func validateHandshake(response: String, key: String): Bool { + if (!response.startsWith("HTTP/1.1 101")) { + return false + } + let expected = FrameStream.computeAccept(key) + var foundUpgrade = false + var foundAccept = false + let lines = response.split("\r\n") + for (line in lines) { + let ci = line.indexOf(":") + if (let Some(c) <- ci) { + let name = line[0..c].toAsciiLower() + let value = line[c + 1..].trimAscii() + if (name == "upgrade" && value.toAsciiLower().contains("websocket")) { + foundUpgrade = true + } + if (name == "sec-websocket-accept" && value == expected) { + foundAccept = true + } + } + } + foundUpgrade && foundAccept + } +} diff --git a/src/common/FrameStream.cj b/src/common/FrameStream.cj new file mode 100644 index 0000000..1f91233 --- /dev/null +++ b/src/common/FrameStream.cj @@ -0,0 +1,331 @@ +/* + * simcu::websocket.common —— RFC 6455 帧层。 + * FrameStream 同时服务客户端与服务端： + * - 缓冲式读取（readExact / readHttpHeader，握手后多余字节保留在缓冲中） + * - 帧解析（FIN/RSV/opcode/扩展长度 126/127/掩码解包，控制帧 ≤125、maxPayload 限制） + * - 帧发送（maskOutgoing 决定是否掩码：客户端必须掩码、服务端不掩码，Mutex 保护写） + * - readMessage()：自动回 Pong、忽略 Pong、分片重组（continuation） + * - 握手工具：Sec-WebSocket-Key 生成 / Sec-WebSocket-Accept 计算 + */ +package simcu::websocket.common + +import std.collection.ArrayList +import std.net.SocketAddress +import std.net.StreamingSocket +import std.random.Random +import std.sync.Mutex +import stdx.crypto.digest.SHA1 +import stdx.encoding.base64.toBase64String + +/// 单帧结构（readFrame 返回值）。 +public class WsFrame { + public let fin: Bool + public let opcode: Int64 + public let payload: Array + + public init(fin: Bool, opcode: Int64, payload: Array) { + this.fin = fin + this.opcode = opcode + this.payload = payload + } +} + +/// 帧与连接层（客户端 / 服务端连接共用）。 +public class FrameStream { + public static let wsGuid: String = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" // RFC 6455 握手 GUID + + private static let opcodeContinuation: Int64 = 0 + private static let opcodeText: Int64 = 1 + private static let opcodeBinary: Int64 = 2 + private static let opcodeClose: Int64 = 8 + private static let opcodePing: Int64 = 9 + private static let opcodePong: Int64 = 10 + + private let sock: StreamingSocket + private let maxPayload: Int64 + private let maskOutgoing: Bool + private let writeLock = Mutex() + private var pending = ArrayList() + private var pendingPos: Int64 = 0 + private let readBuf = Array(8192, repeat: 0) + + /// @param sock 底层 TCP（或 TLS）套接字。 + /// @param maxPayload 单条消息最大字节数（超出抛 WebSocketException(code=1009)）。 + /// @param maskOutgoing 发送帧是否掩码：客户端 true、服务端 false。 + public init(sock: StreamingSocket, maxPayload: Int64, maskOutgoing: Bool) { + this.sock = sock + this.maxPayload = maxPayload + this.maskOutgoing = maskOutgoing + } + + // ===== 握手工具 ===== + + /// 生成 Sec-WebSocket-Key：16 字节随机数 Base64。 + public static func generateWebSocketKey(): String { + let rnd = Random() + var bytes = Array(16, repeat: 0) + for (i in 0..16) { + bytes[i] = UInt8(rnd.nextUInt64() & 0xFFu64) + } + toBase64String(bytes) + } + + /// 计算 Sec-WebSocket-Accept = base64(SHA1(key + GUID))。 + public static func computeAccept(key: String): String { + let md = SHA1() + md.write((key + wsGuid).toArray()) + toBase64String(md.finish()) + } + + /// 解析 close 帧 payload：返回 (关闭码, 关闭原因)。空 payload → (1005, "")。 + public static func parseClosePayload(payload: Array): (Int64, String) { + if (payload.size == 0) { + return (1005, "") + } + if (payload.size == 1) { + throw WebSocketException("close 帧 payload 长度非法", code: 1002) + } + let code = (Int64(payload[0]) << 8) | Int64(payload[1]) + (code, String.fromUtf8(payload[2..])) + } + + // ===== 读取 ===== + + /// 读取 HTTP 头部（到 \r\n\r\n），握手后的帧字节保留在缓冲中。 + public func readHttpHeader(): String { + let sb = StringBuilder() + var tail = ArrayList() + while (true) { + let b = readByte() + tail.add(b) + sb.append(Rune(UInt32(b))) + if (tail.size > 4) { + tail.remove(0..1) + } + if (tail.size == 4 && tail[0] == 13u8 && tail[1] == 10u8 && tail[2] == 13u8 && tail[3] == 10u8) { + break + } + } + sb.toString() + } + + /// 读取一帧。 + public func readFrame(): WsFrame { + let h = readExact(2) + let b0 = h[0] + let b1 = h[1] + if ((b0 & 0x70u8) != 0u8) { + throw WebSocketException("帧 RSV 位不为 0", code: 1002) + } + let fin = (b0 & 0x80u8) != 0u8 + let opcode = Int64(b0 & 0x0Fu8) + let masked = (b1 & 0x80u8) != 0u8 + var len = Int64(b1 & 0x7Fu8) + if (len == 126) { + let ext = readExact(2) + len = (Int64(ext[0]) << 8) | Int64(ext[1]) + } else if (len == 127) { + let ext = readExact(8) + len = 0 + for (i in 0..8) { + len = (len << 8) | Int64(ext[i]) + } + } + if (len > maxPayload) { + throw WebSocketException("消息超过 maxPayload(${maxPayload})", code: 1009) + } + // 控制帧：必须 FIN，payload ≤ 125 + if (opcode >= 8 && (len > 125 || !fin)) { + throw WebSocketException("非法控制帧", code: 1002) + } + var maskKey = Array(0, repeat: 0) + if (masked) { + maskKey = readExact(4) + } + var data = readExact(len) + if (masked) { + for (i in 0..data.size) { + data[i] = data[i] ^ maskKey[i % 4] + } + } + WsFrame(fin, opcode, data) + } + + /** + * 读取一条完整消息（自动分片重组）。 + * 控制帧处理：Ping → 自动回 Pong；Pong → 忽略。 + * @return 数据帧或 close 帧；连接已关闭（EOF）返回 None。 + * @throws WebSocketException 协议错误 / 超过 maxPayload。 + */ + public func readMessage(): ?WsFrame { + var messageOpcode: Int64 = -1 + var messagePayload = ArrayList() + while (true) { + let frame = readFrame() + if (frame.opcode == 9) { + // Ping → 自动回 Pong（保持原 payload） + writeFrame(10, frame.payload) + } else if (frame.opcode == 10) { + // Pong → 忽略 + () + } else if (frame.opcode == 8) { + // Close → 交给调用方处理 + if (messageOpcode != -1) { + throw WebSocketException("分片期间收到 close 帧", code: 1002) + } + return Some(frame) + } else if (frame.opcode == 0) { + // Continuation + if (messageOpcode == -1) { + throw WebSocketException("非预期的 continuation 帧", code: 1002) + } + for (b in frame.payload) { + messagePayload.add(b) + } + if (frame.fin) { + return Some(WsFrame(true, messageOpcode, messagePayload.toArray())) + } + } else { + // 新数据帧（text/binary） + if (messageOpcode != -1) { + throw WebSocketException("分片期间收到新的数据帧", code: 1002) + } + messageOpcode = frame.opcode + for (b in frame.payload) { + messagePayload.add(b) + } + if (frame.fin) { + return Some(WsFrame(true, frame.opcode, messagePayload.toArray())) + } + } + } + // 不可达：仅满足编译器对返回类型的检查 + return None + } + + // ===== 发送 ===== + + /// 发送一帧（客户端掩码 / 服务端不掩码），Mutex 串行化并发写。 + public func writeFrame(opcode: Int64, payload: Array): Unit { + synchronized(writeLock) { + if (opcode >= 8 && payload.size > 125) { + throw WebSocketException("控制帧 payload 不得超过 125 字节", code: 1002) + } + var head = ArrayList() + head.add(0x80u8 | UInt8(opcode)) + let len = Int64(payload.size) + if (len < 126) { + head.add(if (maskOutgoing) { 0x80u8 | UInt8(len) } else { UInt8(len) }) + } else if (len <= 0xFFFF) { + head.add(if (maskOutgoing) { 0x80u8 | 126u8 } else { 126u8 }) + head.add(UInt8((len >> 8) & 0xFF)) + head.add(UInt8(len & 0xFF)) + } else { + head.add(if (maskOutgoing) { 0x80u8 | 127u8 } else { 127u8 }) + for (i in 0..8) { + head.add(UInt8((len >> ((7 - i) * 8)) & 0xFF)) + } + } + if (maskOutgoing) { + let rnd = Random() + var mask = Array(4, repeat: 0) + for (i in 0..4) { + mask[i] = UInt8(rnd.nextUInt64() & 0xFFu64) + } + for (m in mask) { + head.add(m) + } + sock.write(head.toArray()) + var masked = Array(payload.size, repeat: 0) + for (i in 0..payload.size) { + masked[i] = payload[i] ^ mask[i % 4] + } + sock.write(masked) + } else { + sock.write(head.toArray()) + sock.write(payload) + } + sock.flush() + } + } + + /// 发送文本帧。 + public func writeText(text: String): Unit { + writeFrame(1, text.toArray()) + } + + /// 发送二进制帧。 + public func writeBinary(data: Array): Unit { + writeFrame(2, data) + } + + /// 发送 Ping 帧。 + public func writePing(payload: Array): Unit { + writeFrame(9, payload) + } + + /// 发送 Pong 帧。 + public func writePong(payload: Array): Unit { + writeFrame(10, payload) + } + + /// 发送 close 帧（2 字节大端关闭码 + UTF-8 原因）。 + public func writeClose(code: Int64, reason: String): Unit { + var payload = ArrayList() + payload.add(UInt8((code >> 8) & 0xFF)) + payload.add(UInt8(code & 0xFF)) + for (b in reason.toArray()) { + payload.add(b) + } + writeFrame(8, payload.toArray()) + } + + /// 关闭底层套接字。 + public func close(): Unit { + sock.close() + } + + /// 远端地址（SocketAddress）。 + public func remoteAddress(): SocketAddress { + sock.remoteAddress + } + + // ===== 内部：缓冲读取 ===== + + private func readByte(): Byte { + if (pendingPos < pending.size) { + let v = pending[pendingPos] + pendingPos++ + return v + } + fillPending() + let v = pending[pendingPos] + pendingPos++ + v + } + + private func readExact(n: Int64): Array { + var result = ArrayList() + while (result.size < n) { + if (pendingPos < pending.size) { + result.add(pending[pendingPos]) + pendingPos++ + } else { + fillPending() + } + } + result.toArray() + } + + private func fillPending(): Unit { + let r = sock.read(readBuf) + if (r <= 0) { + throw WebSocketException("连接已关闭") + } + pending.clear() + pendingPos = 0 + for (i in 0..r) { + pending.add(readBuf[i]) + } + } +} diff --git a/src/common/WsTypes.cj b/src/common/WsTypes.cj new file mode 100644 index 0000000..fe10cb3 --- /dev/null +++ b/src/common/WsTypes.cj @@ -0,0 +1,145 @@ +/* + * simcu::websocket.common —— 共享类型定义。 + * WebSocketMessage / MessageType / ReadyState / WebSocketException / CloseCodes / 事件名常量。 + */ +package simcu::websocket.common + +/// 消息类型：文本或二进制。 +public enum MessageType { + | Text + | Binary + + public operator func ==(right: MessageType): Bool { + match (this) { + case Text => match (right) { + case Text => true + case _ => false + } + case Binary => match (right) { + case Binary => true + case _ => false + } + } + } + + public operator func !=(right: MessageType): Bool { + !(this == right) + } +} + +/// 连接就绪状态（对齐浏览器 WebSocket readyState 语义）。 +public enum ReadyState <: ToString { + | Connecting + | Open + | Closing + | Closed + + public operator func ==(right: ReadyState): Bool { + match (this) { + case Connecting => match (right) { + case Connecting => true + case _ => false + } + case Open => match (right) { + case Open => true + case _ => false + } + case Closing => match (right) { + case Closing => true + case _ => false + } + case Closed => match (right) { + case Closed => true + case _ => false + } + } + } + + public operator func !=(right: ReadyState): Bool { + !(this == right) + } + + public func toString(): String { + match (this) { + case Connecting => "Connecting" + case Open => "Open" + case Closing => "Closing" + case Closed => "Closed" + } + } +} + +/** + * WebSocket 消息：携带类型 + 原始 payload 字节。 + * 文本消息的 UTF-8 解码通过 text 属性按需进行。 + */ +public class WebSocketMessage { + public let `type`: MessageType + public let bytes: Array + + public init(`type`: MessageType, bytes: Array) { + this.`type` = `type` + this.bytes = bytes + } + + /// 构造文本消息。 + public static func fromText(value: String): WebSocketMessage { + WebSocketMessage(MessageType.Text, value.toArray()) + } + + /// 构造二进制消息。 + public static func fromBinary(data: Array): WebSocketMessage { + WebSocketMessage(MessageType.Binary, data) + } + /// 文本消息内容（二进制消息返回 None）。 + public prop text: ?String { + get() { + if (`type` == MessageType.Text) { + Some(String.fromUtf8(bytes)) + } else { + None + } + } + } + + /// 友好描述（调试用）。 + public func describe(): String { + match (`type`) { + case MessageType.Text => "[text] ${text.getOrThrow()}" + case MessageType.Binary => "[binary] ${bytes.size} bytes" + } + } +} + +/** + * WebSocket 异常：code 为 RFC 6455 关闭码（0 表示无关闭码，如握手失败、协议错误）。 + */ +public class WebSocketException <: Exception { + public let code: Int64 + + public init(message: String, code!: Int64 = 0) { + super(message) + this.code = code + } +} + +/// RFC 6455 常见关闭码（发送/接收均按 2 字节大端编码在 close 帧 payload 中）。 +public class CloseCodes { + public static let normalClosure: Int64 = 1000 + public static let goingAway: Int64 = 1001 + public static let protocolError: Int64 = 1002 + public static let unsupportedData: Int64 = 1003 + public static let abnormalClosure: Int64 = 1006 // 不发送到对端，仅本地事件使用 + public static let invalidFramePayloadData: Int64 = 1007 + public static let policyViolation: Int64 = 1008 + public static let messageTooBig: Int64 = 1009 + public static let internalServerError: Int64 = 1011 +} + +/// 服务端事件名常量（on(event, listener) 使用）。 +public class WsEvents { + public static let connection: String = "connection" + public static let message: String = "message" + public static let close: String = "close" + public static let error: String = "error" +} diff --git a/src/server/WebSocketConnection.cj b/src/server/WebSocketConnection.cj new file mode 100644 index 0000000..cb7ec05 --- /dev/null +++ b/src/server/WebSocketConnection.cj @@ -0,0 +1,228 @@ +/* + * 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) => ... }) + */ +package simcu::websocket.server + +import std.net.IPSocketAddress +import std.net.TcpSocket +import std.sync.Mutex +import std.time.* + +import simcu::websocket.common.FrameStream +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 连接（事件回调式）。 +public class WebSocketConnection { + private let parent: ?WebSocketServer + private let frame: FrameStream + private let stateLock = Mutex() + private var state: ReadyState = ReadyState.Open + + /// 连接唯一编号（由服务端分配，用于连接集合的引用比较）。 + 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 => () } + + internal init(parent: ?WebSocketServer, frame: FrameStream, connId: Int64) { + this.parent = parent + this.frame = frame + this.connId = connId + } + + /// 当前连接状态。 + public prop readyState: ReadyState { + get() { state } + } + + /// 连接是否处于 Open。 + public func isOpen(): Bool { + state == ReadyState.Open + } + + /// 远端地址（ip:port 或套接字地址字符串）。 + public prop remoteAddress: String { + get() { + let sa = frame.remoteAddress() + if (let Some(ip) <- (sa as IPSocketAddress)) { + "${ip.address}:${ip.port}" + } else { + sa.toString() + } + } + } + + /// 注册连接级事件。支持 "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}") + } + } + + /// 发送消息。 + public func send(message: WebSocketMessage): Unit { + ensureOpen() + if (message.`type` == MessageType.Text) { + frame.writeFrame(1, message.bytes) + } else { + frame.writeFrame(2, message.bytes) + } + } + + /// 发送文本消息。 + public func sendText(text: String): Unit { + ensureOpen() + frame.writeText(text) + } + + /// 发送二进制消息。 + public func sendBinary(data: Array): Unit { + ensureOpen() + frame.writeBinary(data) + } + + /// 发送 Ping（心跳保活）。 + public func ping(): Unit { + ensureOpen() + frame.writePing(Array(0, repeat: 0)) + } + + /// 发起关闭握手：发送 close 帧，等待对端回执；5 秒兜底强制断开。 + public func close(code!: Int64 = 1000, reason!: String = ""): Unit { + synchronized(stateLock) { + if (state != ReadyState.Open) { + return + } + state = ReadyState.Closing + } + try { + frame.writeClose(code, reason) + } catch (_) { + terminate() + return + } + spawn { => + sleep(Duration.second * 5) + if (state == ReadyState.Closing) { + terminate() + } + } + } + + /// 立即断开底层连接（不发送 close 帧）。 + public func terminate(): Unit { + synchronized(stateLock) { + if (state == ReadyState.Closed) { + return + } + state = ReadyState.Closed + } + try { + frame.close() + } catch (_) { + () + } + } + + private func ensureOpen(): Unit { + if (state != ReadyState.Open) { + throw WebSocketException("连接未打开（readyState=${state}）") + } + } + + /// 读循环（由 WebSocketServer 的连接处理协程调用）。 + internal func runReadLoop(): Unit { + var code: Int64 = 1006 + var reason: String = "" + try { + while (true) { + let m = frame.readMessage() + match (m) { + case None => + break + case Some(f) => + if (f.opcode == 8) { + let (c, r) = FrameStream.parseClosePayload(f.payload) + code = c + reason = r + if (state == ReadyState.Open) { + try { + frame.writeClose(c, r) + } catch (_) { + () + } + } + break + } else { + let msg = if (f.opcode == 1) { + WebSocketMessage(MessageType.Text, f.payload) + } else { + WebSocketMessage(MessageType.Binary, f.payload) + } + onMessage(msg) + if (let Some(p) <- parent) { + p.fireMessage(this, msg) + } + } + } + } + } catch (e: Exception) { + code = 1006 + if (state != ReadyState.Closing) { + onError(e) + if (let Some(p) <- parent) { + p.fireError(e) + } + } + } + synchronized(stateLock) { + state = ReadyState.Closed + } + try { + frame.close() + } catch (_) { + () + } + if (let Some(p) <- parent) { + p.removeConnection(this) + } + onClose(code, reason) + if (let Some(p) <- parent) { + p.fireClose(this, code, reason) + } + } +} diff --git a/src/server/WebSocketServer.cj b/src/server/WebSocketServer.cj new file mode 100644 index 0000000..9f9d66f --- /dev/null +++ b/src/server/WebSocketServer.cj @@ -0,0 +1,329 @@ +/* + * simcu::websocket.server —— 事件回调式 WebSocket 服务端（RFC 6455）。 + * + * 用法： + * let server = WebSocketServer(bindAt: 8080) + * server.on("connection", { (conn) => + * println("新连接: ${conn.remoteAddress}") + * conn.on("message", { (msg) => conn.sendText("echo: ${msg.text}") }) + * }) + * server.listen() + * ... + * server.close() // 停止监听并断开所有连接 + * + * 服务端事件：on('connection') / on('message') / on('close') / on('error')。 + * 连接方法：send / sendText / sendBinary / ping / close / terminate。 + * 集群方法：broadcast / broadcastText / broadcastBinary / connectionCount。 + */ +package simcu::websocket.server + +import std.collection.ArrayList +import std.net.IPSocketAddress +import std.net.SocketException +import std.net.TcpServerSocket +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 服务端。 +public class WebSocketServer { + private let bindPort: UInt16 + private let pathFilter: ?String + private let maxPayload: Int64 + private var serverSock: ?TcpServerSocket = None + private var running = false + private var connections = ArrayList() + private var nextConnId: Int64 = 1 + private 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 maxPayload 单条消息最大字节数，默认 64KB。 + public init(bindAt!: UInt16 = 0, path!: ?String = None, maxPayload!: Int64 = 65536) { + this.bindPort = bindAt + this.pathFilter = path + 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}") + } + } + + /// 启动监听：bind + accept 循环，每连接一个协程处理握手与读循环。 + public func listen(backlog!: Int64 = 128): Unit { + if (running) { + throw WebSocketException("服务端已在监听") + } + let s = TcpServerSocket(bindAt: bindPort) + s.backlogSize = backlog + s.bind() + serverSock = Some(s) + running = true + spawn { => acceptLoop() } + } + + /// 实际监听端口（bindAt=0 时用于发现随机端口）。 + public prop localPort: UInt16 { + get() { + let sa = serverSock.getOrThrow().localAddress + if (let Some(ip) <- (sa as IPSocketAddress)) { + ip.port + } else { + throw WebSocketException("无法解析本地监听端口") + } + } + } + + /// 当前在线连接数。 + public prop connectionCount: Int64 { + get() { + synchronized(connLock) { + connections.size + } + } + } + + /// 停止监听并强制断开所有连接（每连接会触发自身 close 事件）。 + public func close(): Unit { + running = false + if (let Some(s) <- serverSock) { + try { + s.close() + } catch (_) { + () + } + serverSock = None + } + let snapshot = snapshotConnections() + for (c in snapshot) { + c.terminate() + } + } + + /// 向所有在线连接广播消息（单个连接发送失败不影响其余连接）。 + public func broadcast(message: WebSocketMessage): Unit { + let snapshot = snapshotConnections() + for (c in snapshot) { + try { + c.send(message) + } catch (_) { + () + } + } + } + + /// 广播文本消息。 + public func broadcastText(text: String): Unit { + broadcast(WebSocketMessage.fromText(text)) + } + + /// 广播二进制消息。 + public func broadcastBinary(data: Array): Unit { + broadcast(WebSocketMessage.fromBinary(data)) + } + + // ===== 内部回调转发（供 WebSocketConnection 调用）===== + + internal func fireMessage(conn: WebSocketConnection, msg: WebSocketMessage): Unit { + onMessage(conn, msg) + } + + internal func fireClose(conn: WebSocketConnection, code: Int64, reason: String): Unit { + onClose(conn, code, reason) + } + + internal func fireError(e: Exception): Unit { + onError(e) + } + + internal func addConnection(conn: WebSocketConnection): Unit { + synchronized(connLock) { + connections.add(conn) + } + } + + internal func allocateConnId(): Int64 { + synchronized(connLock) { + let id = nextConnId + nextConnId += 1 + id + } + } + + 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 + } + } + if (idx >= 0) { + connections.remove(at: idx) + } + } + } + + private func snapshotConnections(): ArrayList { + synchronized(connLock) { + let copy = ArrayList() + for (c in connections) { + copy.add(c) + } + copy + } + } + + // ===== 监听与握手 ===== + + private func acceptLoop(): Unit { + while (running) { + try { + let s = serverSock + if (let Some(sock) <- s) { + let client = sock.accept() + spawn { => handleConnection(client) } + } else { + return + } + } catch (e: SocketException) { + if (!running) { + return + } + onError(e) + } catch (e: Exception) { + if (!running) { + return + } + onError(e) + } + } + } + + private func handleConnection(sock: TcpSocket): Unit { + let fs = FrameStream(sock, maxPayload, false) + try { + let request = fs.readHttpHeader() + let lines = request.split("\r\n") + if (lines.size == 0 || !lines[0].startsWith("GET ")) { + reject(sock, 400, "Bad Request") + return + } + let parts = lines[0].split(" ") + if (parts.size < 3) { + reject(sock, 400, "Bad Request") + return + } + if (let Some(p) <- pathFilter) { + if (parts[1] != p) { + reject(sock, 404, "Not Found") + return + } + } + var upgradeOk = false + var connectionOk = false + var key = "" + for (i in 1..lines.size) { + let line = lines[i] + let ci = line.indexOf(":") + if (let Some(c) <- ci) { + let name = line[0..c].toAsciiLower() + let value = line[c + 1..].trimAscii() + if (name == "upgrade") { + upgradeOk = value.toAsciiLower().contains("websocket") + } + if (name == "connection") { + connectionOk = value.toAsciiLower().contains("upgrade") + } + if (name == "sec-websocket-key") { + key = value + } + } + } + if (!upgradeOk || !connectionOk || key == "") { + reject(sock, 400, "Bad Request") + return + } + + // 101 响应 + let accept = FrameStream.computeAccept(key) + let resp = StringBuilder() + resp.append("HTTP/1.1 101 Switching Protocols\r\n") + resp.append("Upgrade: websocket\r\n") + resp.append("Connection: Upgrade\r\n") + resp.append("Sec-WebSocket-Accept: ${accept}\r\n\r\n") + sock.write(resp.toString().toArray()) + sock.flush() + + let conn = WebSocketConnection(Some(this), fs, allocateConnId()) + addConnection(conn) + onConnection(conn) + conn.runReadLoop() + } catch (e: Exception) { + // 握手阶段失败：上报服务端 error 事件 + onError(e) + } + } + + private func reject(sock: TcpSocket, status: Int64, message: String): Unit { + try { + let resp = "HTTP/1.1 ${status} ${message}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + sock.write(resp.toArray()) + sock.flush() + } catch (_) { + () + } + try { + sock.close() + } catch (_) { + () + } + } +} diff --git a/src/tests/websocket_test.cj b/src/tests/websocket_test.cj new file mode 100644 index 0000000..1f50443 --- /dev/null +++ b/src/tests/websocket_test.cj @@ -0,0 +1,237 @@ +/* + * simcu::websocket.tests —— 端到端集成测试（cjpm test）。 + * 覆盖：文本回声 / 广播 / 关闭握手（客户端主动、服务端主动）/ terminate / + * ping 心跳后连接仍可用 / 路径过滤 / maxPayload 超限(1009)。 + * + * 运行：在 websocket-cj 目录执行 cjpm test + */ +package simcu::websocket.tests + +import std.time.* +import std.unittest.* +import std.unittest.testmacro.Expect +import std.unittest.testmacro.TestCase +import std.unittest.testmacro.Test + +import simcu::websocket.client.WebSocketClient +import simcu::websocket.common.MessageType +import simcu::websocket.common.WebSocketException +import simcu::websocket.common.WebSocketMessage +import simcu::websocket.server.WebSocketConnection +import simcu::websocket.server.WebSocketServer + +/// 测试回收集合：把回调中捕获的值存到独立实例字段，避免闭包捕获歧义。 +class WsTestHarness { + var serverConn: ?WebSocketConnection = None + var connMessage: ?String = None + var connErrorCode: Int64 = 0 + var connCloseCode: Int64 = -1 + var connCloseReason: String = "" + var clientMessage: ?String = None + var clientCloseCode: Int64 = -1 + var clientCloseReason: String = "" + var clientErrorCount: Int64 = 0 +} + +/// 轮询等待条件成立（每 50ms 检查一次）。 +private func waitUntil(timeoutMs: Int64, condition: () -> Bool): Bool { + let total = timeoutMs / 50 + for (_ in 0..total) { + if (condition()) { + return true + } + sleep(Duration.millisecond * 50) + } + condition() +} + +@Test +public class WebSocketTest { + /// 1. 文本回声：client.sendText → 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) + client.onMessage = { m: WebSocketMessage => + h.clientMessage = m.text.getOrThrow() + } + client.connect() + client.sendText("hello") + + @Expect(waitUntil(5000) { => h.connMessage == Some("hello") }, true) + @Expect(waitUntil(5000) { => h.clientMessage == Some("echo: hello") }, true) + client.close() + server.close() + } + + /// 2. 广播：server.broadcastText 两个客户端都收到 + @TestCase + public func broadcast(): Unit { + let h1 = WsTestHarness() + let h2 = WsTestHarness() + let server = WebSocketServer(bindAt: 0) + server.listen() + + let c1 = WebSocketClient("127.0.0.1", server.localPort) + c1.onMessage = { m: WebSocketMessage => + h1.clientMessage = m.text.getOrThrow() + } + c1.connect() + let c2 = WebSocketClient("127.0.0.1", server.localPort) + c2.onMessage = { m: WebSocketMessage => + h2.clientMessage = m.text.getOrThrow() + } + c2.connect() + @Expect(waitUntil(5000) { => server.connectionCount == 2 }, true) + + server.broadcastText("hi-all") + @Expect(waitUntil(5000) { => h1.clientMessage == Some("hi-all") }, true) + @Expect(waitUntil(5000) { => h2.clientMessage == Some("hi-all") }, true) + c1.close() + c2.close() + server.close() + } + + /// 3. 关闭握手：client.close(1000, "bye") → 双端 onClose 都收到 (1000, "bye") + @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) + client.onClose = { code: Int64, reason: String => + h.clientCloseCode = code + h.clientCloseReason = reason + } + client.connect() + client.close(code: 1000, reason: "bye") + + @Expect(waitUntil(5000) { => h.clientCloseCode == 1000 && h.clientCloseReason == "bye" }, true) + @Expect(waitUntil(5000) { => h.connCloseCode == 1000 && h.connCloseReason == "bye" }, true) + server.close() + } + + /// 4. 服务端主动关闭：conn.close(4000, "svr") → client onClose 收 (4000, "svr") + @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) + client.onClose = { code: Int64, reason: String => + h.clientCloseCode = code + h.clientCloseReason = reason + } + client.connect() + + @Expect(waitUntil(5000) { => h.clientCloseCode == 4000 && h.clientCloseReason == "svr" }, true) + server.close() + } + + /// 5. terminate：client.terminate() 立即断开 → server conn onClose 收 (1006, "") + @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) + client.connect() + client.terminate() + + @Expect(waitUntil(5000) { => h.connCloseCode == 1006 && h.connCloseReason == "" }, true) + server.close() + } + + /// 6. ping：心跳后连接仍可用，后续消息正常到达 + @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) + client.connect() + + client.ping() + sleep(Duration.millisecond * 100) + client.sendText("after-ping") + @Expect(waitUntil(5000) { => h.connMessage == Some("after-ping") }, true) + client.close() + server.close() + } + + /// 7. 路径过滤：path=Some("/ws") 时 /ws 握手成功，/other 被拒（connect 抛异常） + @TestCase + public func pathFilter(): Unit { + let server = WebSocketServer(bindAt: 0, path: Some("/ws")) + server.listen() + + let ok = WebSocketClient("127.0.0.1", server.localPort, path: "/ws") + ok.connect() + ok.close() + + let bad = WebSocketClient("127.0.0.1", server.localPort, path: "/other") + var rejected = false + try { + bad.connect() + } catch (e: WebSocketException) { + rejected = true + } + @Expect(rejected, true) + server.close() + } + + /// 8. maxPayload：server 限制 10 字节，client 发长文本 → server 连接 onError 收 code=1009 + @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) + client.onError = { e: Exception => + h.clientErrorCount += 1 + } + client.connect() + client.sendText("this message is way too long") + + @Expect(waitUntil(5000) { => h.connErrorCode == 1009 }, true) + client.terminate() + server.close() + } +} diff --git a/src/websocket.cj b/src/websocket.cj new file mode 100644 index 0000000..5508c8b --- /dev/null +++ b/src/websocket.cj @@ -0,0 +1,14 @@ +/* + * simcu::websocket —— SimApi WebSocket 库（RFC 6455）。 + * + * 模块锚点文件，本模块包含以下子包： + * - simcu::websocket.common 共享类型：WebSocketMessage / ReadyState / WebSocketException / FrameStream（帧编解码） + * - simcu::websocket.client WebSocketClient：事件回调式客户端 + * - simcu::websocket.server WebSocketServer + WebSocketConnection：事件回调式服务端 + * + * 典型用法： + * import simcu::websocket.client.WebSocketClient + * import simcu::websocket.server.WebSocketServer + * import simcu::websocket.common.WebSocketMessage + */ +package simcu::websocket