重构服务端为接口式(WebSocketServer(port)+addHandler 工厂,每连接独立 handler),组管理与发送收敛为链式门面 group(name)/conn(connId),统一 send 三重重载,升级版本 1.1.0

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<Byte>）。
+ * 组管理（链式）：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<String, () -> IWebsocketHandler>()
     private var serverSock: ?TcpServerSocket = None
     private var running = false
-    private var connections = ArrayList<WebSocketConnection>()
+    private var connections = HashMap<Int64, WebSocketConnection>()
+    internal var connGroups = HashMap<String, ArrayList<Int64>>()
     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<Byte>）。
+    public func broadcast(text: String): Unit {
         broadcast(WebSocketMessage.fromText(text))
     }
 
     /// 广播二进制消息。
-    public func broadcastBinary(data: Array<Byte>): Unit {
+    public func broadcast(data: Array<Byte>): Unit {
         broadcast(WebSocketMessage.fromBinary(data))
     }
 
-    // ===== 内部回调转发（供 WebSocketConnection 调用）=====
-
-    internal func fireMessage(conn: WebSocketConnection, msg: WebSocketMessage): Unit {
-        onMessage(conn, msg)
+    /// 所有组名（快照）。
+    public func groups(): ArrayList<String> {
+        synchronized(connLock) {
+            let copy = ArrayList<String>()
+            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<String>()
+        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<WebSocketConnection> {
         synchronized(connLock) {
             let copy = ArrayList<WebSocketConnection>()
-            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}")
         }
     }