chore: update version to 0.0.47 and enhance WebSocket handling in server implementation

package.json

@@ -1,7 +1,7 @@
 {
   "$schema": "https://json.schemastore.org/package",
   "name": "@kevisual/router",
-  "version": "0.0.43",
+  "version": "0.0.47",
   "description": "",
   "type": "module",
   "main": "./dist/router.js",

src/app.ts

@@ -134,7 +134,7 @@ export class App<U = {}> {
     }
     this.server.on({
       id: 'app-request-listener',
-      fun: fn
+      fun: fn as any,
     });
   }
 }

src/index.ts

@@ -20,4 +20,15 @@ export { App } from './app.ts';
 export * from './router-define.ts';
 
 
-export { RouterReq, RouterRes } from './server/server-type.ts';
+export {
+  RouterReq,
+  RouterRes,
+  OnWebSocketFn,
+  WS,
+  WebSocketReq,
+  WebSocketRes,
+  Listener,
+  WebSocketListenerFun,
+  HttpListenerFun,
+  OnListener,
+} from './server/server-type.ts';

src/server/server-base.ts

@@ -1,9 +1,9 @@
 import type { IncomingMessage, ServerResponse } from 'node:http';
 import { handleServer } from './handle-server.ts';
 import * as cookie from './cookie.ts';
-import { ServerType, Listener, OnListener, ServerOpts } from './server-type.ts';
+import { ServerType, Listener, OnListener, ServerOpts, OnWebSocketOptions, OnWebSocketFn, WebScoketListenerFun, ListenerFun, HttpListenerFun, WS } from './server-type.ts';
 import { parseIfJson } from '../utils/parse.ts';
-
+import { EventEmitter } from 'events';
 type CookieFn = (name: string, value: string, options?: cookie.SerializeOptions, end?: boolean) => void;
 
 export type HandleCtx = {
@@ -63,6 +63,7 @@ export class ServerBase implements ServerType {
   _callback: any;
   cors: Cors;
   listeners: Listener[] = [];
+  emitter = new EventEmitter();
   constructor(opts?: ServerOpts) {
     this.path = opts?.path || '/api/router';
     this.handle = opts?.handle;
@@ -118,9 +119,9 @@ export class ServerBase implements ServerType {
       }
       const listeners = that.listeners || [];
       for (const item of listeners) {
-        const fun = item.fun;
-        if (typeof fun === 'function' && !item.io) {
-          await fun(req, res);
+        const func = item.func as any;
+        if (typeof func === 'function' && !item.io) {
+          await func(req, res);
         }
       }
       if (res.headersSent) {
@@ -178,13 +179,13 @@ export class ServerBase implements ServerType {
   on(listener: OnListener) {
     this.listeners = [];
     if (typeof listener === 'function') {
-      this.listeners.push({ fun: listener });
+      this.listeners.push({ func: listener });
       return;
     }
     if (Array.isArray(listener)) {
       for (const item of listener) {
         if (typeof item === 'function') {
-          this.listeners.push({ fun: item });
+          this.listeners.push({ func: item });
         } else {
           this.listeners.push(item);
         }
@@ -193,7 +194,7 @@ export class ServerBase implements ServerType {
       this.listeners.push(listener);
     }
   }
-  async onWebSocket({ ws, message, pathname, token, id }) {
+  async onWebSocket({ ws, message, pathname, token, id }: OnWebSocketOptions) {
     const listener = this.listeners.find((item) => item.path === pathname && item.io);
     const data: any = parseIfJson(message);
 
@@ -201,7 +202,8 @@ export class ServerBase implements ServerType {
       const end = (data: any) => {
         ws.send(JSON.stringify(data));
       }
-      listener.fun({
+      (listener.func as WebScoketListenerFun)({
+        emitter: this.emitter,
         data,
         token,
         id,
@@ -262,4 +264,15 @@ export class ServerBase implements ServerType {
       end({ code: 500, message: `${type} server is error` });
     }
   }
+  async onWsClose(ws: WS) {
+    const id = ws?.data?.id || '';
+    if (id) {
+      this.emitter.emit('close--' + id, { type: 'close', ws, id });
+      setTimeout(() => {
+        // 关闭后 5 秒清理监听器, 避免内存泄漏， 理论上原本的自己就应该被清理掉了，这里是保险起见
+        this.emitter.removeAllListeners('close--' + id);
+        this.emitter.removeAllListeners(id);
+      }, 5000);
+    }
+  }
 }

src/server/server-bun.ts

@@ -223,7 +223,7 @@ export class BunServer extends ServerBase implements ServerType {
       },
       websocket: {
         open: (ws: any) => {
-          ws.send('connected');
+          ws.send(JSON.stringify({ type: 'connected' }));
         },
         message: async (ws: any, message: string | Buffer) => {
           const pathname = ws.data.pathname || '';
@@ -233,6 +233,7 @@ export class BunServer extends ServerBase implements ServerType {
         },
         close: (ws: any) => {
           // WebSocket 连接关闭
+          this.onWsClose(ws);
         },
       },
     });

src/server/server-type.ts

@@ -1,13 +1,7 @@
-import * as http from 'http';
+import EventEmitter from 'node:events';
+import * as http from 'node:http';
+
 
-export type Listener = {
-  id?: string;
-  io?: boolean;
-  path?: string;
-  fun: (...args: any[]) => Promise<void> | void;
-}
-export type ListenerFun = (...args: any[]) => Promise<void> | void;
-export type OnListener = Listener | ListenerFun | (Listener | ListenerFun)[];
 export type Cors = {
   /**
    * @default '*''
@@ -44,14 +38,49 @@ export interface ServerType {
    * @param listener
    */
   on(listener: OnListener): void;
-  onWebSocket({ ws, message, pathname, token, id }: { ws: WS; message: string | Buffer; pathname: string, token?: string, id?: string }): void;
+  onWebSocket({ ws, message, pathname, token, id }: OnWebSocketOptions): void;
+  onWsClose(ws: WS): void;
 }
 
-type WS = {
+export type OnWebSocketOptions = { ws: WS; message: string | Buffer; pathname: string, token?: string, id?: string }
+export type OnWebSocketFn = (options: OnWebSocketOptions) => Promise<void> | void;
+export type WS = {
   send: (data: any) => void;
-  close: () => void;
+  close: (code?: number, reason?: string) => void;
+  data?: {
+    url: URL;
+    pathname: string;
+    token?: string;
+    id?: string;
+    /**
+     * 鉴权后的获取的信息
+     */
+    userApp?: string;
+  }
+}
+export type Listener = {
+  id?: string;
+  io?: boolean;
+  path?: string;
+  func: WebSocketListenerFun | HttpListenerFun;
 }
 
+export type WebSocketListenerFun = (req: WebSocketReq, res: WebSocketRes) => Promise<void> | void;
+export type HttpListenerFun = (req: RouterReq, res: RouterRes) => Promise<void> | void;
+
+export type WebSocketReq = {
+  emitter?: EventEmitter;
+  ws: WS;
+  data: any;
+  pathname?: string;
+  token?: string;
+  id?: string;
+}
+export type WebSocketRes = {
+  end: (data: any) => void;
+}
+export type ListenerFun = WebSocketListenerFun | HttpListenerFun;;
+export type OnListener = Listener | ListenerFun | (Listener | ListenerFun)[];
 export type RouterReq<T = {}> = {
   url: string;
   method: string;

src/server/ws-server.ts

@@ -24,7 +24,6 @@ export type Listener<T = 'router' | 'chat' | 'ai'> = {
 
 export class WsServerBase {
   wss: WebSocketServer | null;
-  listeners: Listener[] = [];
   listening: boolean = false;
   server: ServerType;
 
@@ -51,11 +50,22 @@ export class WsServerBase {
       const pathname = url.pathname;
       const token = url.searchParams.get('token') || '';
       const id = url.searchParams.get('id') || '';
+      // @ts-ignore
+      ws.data = {
+        url: url,
+        pathname,
+        token,
+        id,
+      }
       ws.on('message', async (message: string | Buffer) => {
         await this.server.onWebSocket({ ws, message, pathname, token, id });
       });
-      ws.send('connected');
+      ws.send(JSON.stringify({ type: 'connected' }));
+      this.wss.on('close', () => {
+        this.server.onWsClose(ws);
+      });
     });
+
   }
 }
 // TODO: ws handle and path and routerContext