refactor(onebot): 精简 WebSocket 适配器实现 (#1644)

* refactor(onebot): 移除 `async-mutex` 依赖

* fix(onebot): 避免重复发送 WebSocket pong

packages/napcat-onebot/network/http-server.ts

@@ -13,14 +13,12 @@ import { URL } from 'url';
 import { ActionName } from '@/napcat-onebot/action/router';
 import { OB11HeartbeatEvent } from '@/napcat-onebot/event/meta/OB11HeartbeatEvent';
 import { OB11LifeCycleEvent, LifeCycleSubType } from '@/napcat-onebot/event/meta/OB11LifeCycleEvent';
-import { Mutex } from 'async-mutex';
 
 export class OB11HttpServerAdapter extends IOB11NetworkAdapter<HttpServerConfig> {
   private app: Express | undefined;
   private server: http.Server | undefined;
   private wsServer?: WebSocketServer;
   private wsClients: WebSocket[] = [];
-  private wsClientsMutex = new Mutex();
   private heartbeatIntervalId: NodeJS.Timeout | null = null;
   private wsClientWithEvent: WebSocket[] = [];
 
@@ -30,7 +28,6 @@ export class OB11HttpServerAdapter extends IOB11NetworkAdapter<HttpServerConfig>
 
   override async onEvent<T extends OB11EmitEventContent> (event: T) {
     // http server is passive, no need to emit event
-    this.wsClientsMutex.runExclusive(async () => {
     const promises = this.wsClientWithEvent.map((wsClient) => {
       return new Promise<void>((resolve, reject) => {
         if (wsClient.readyState === WebSocket.OPEN) {
@@ -42,7 +39,6 @@ export class OB11HttpServerAdapter extends IOB11NetworkAdapter<HttpServerConfig>
       });
     });
     await Promise.allSettled(promises);
-    });
   }
 
   open () {
@@ -65,13 +61,9 @@ export class OB11HttpServerAdapter extends IOB11NetworkAdapter<HttpServerConfig>
     this.server?.close();
     this.app = undefined;
     this.stopHeartbeat();
-    await this.wsClientsMutex.runExclusive(async () => {
+    this.wsClients.forEach((wsClient) => wsClient.close());
-      this.wsClients.forEach((wsClient) => {
-        wsClient.close();
-      });
     this.wsClients = [];
     this.wsClientWithEvent = [];
-    });
     this.wsServer?.close();
   }
 
@@ -153,14 +145,10 @@ export class OB11HttpServerAdapter extends IOB11NetworkAdapter<HttpServerConfig>
       wsClient.on('message', (message) => {
         this.handleWSMessage(wsClient, message).then().catch(e => this.logger.logError(e));
       });
-      wsClient.on('ping', () => {
-        wsClient.pong();
-      });
       wsClient.on('pong', () => {
         // this.logger.logDebug('[OneBot] [HTTP WebSocket] Pong received');
       });
       wsClient.once('close', () => {
-        this.wsClientsMutex.runExclusive(async () => {
         const NormolIndex = this.wsClients.indexOf(wsClient);
         if (NormolIndex !== -1) {
           this.wsClients.splice(NormolIndex, 1);
@@ -173,8 +161,6 @@ export class OB11HttpServerAdapter extends IOB11NetworkAdapter<HttpServerConfig>
           this.stopHeartbeat();
         }
       });
-      });
-      await this.wsClientsMutex.runExclusive(async () => {
       if (!isApiConnect) {
         this.wsClientWithEvent.push(wsClient);
       }
@@ -182,7 +168,6 @@ export class OB11HttpServerAdapter extends IOB11NetworkAdapter<HttpServerConfig>
       if (this.wsClientWithEvent.length > 0) {
         this.startHeartbeat();
       }
-      });
     }).on('error', (err) => this.logger.log('[OneBot] [HTTP WebSocket] Server Error:', err.message));
   }
 
@@ -197,13 +182,11 @@ export class OB11HttpServerAdapter extends IOB11NetworkAdapter<HttpServerConfig>
   private startHeartbeat () {
     if (this.heartbeatIntervalId) return;
     this.heartbeatIntervalId = setInterval(() => {
-      this.wsClientsMutex.runExclusive(async () => {
       this.wsClientWithEvent.forEach((wsClient) => {
         if (wsClient.readyState === WebSocket.OPEN) {
           wsClient.send(JSON.stringify(new OB11HeartbeatEvent(this.core, 30000, this.core.selfInfo.online ?? true, true)));
         }
       });
-      });
     }, 30000);
   }
 

packages/napcat-onebot/network/websocket-client.ts

@@ -85,9 +85,6 @@ export class OB11WebSocketClientAdapter extends IOB11NetworkAdapter<WebsocketCli
         },
 
       });
-      this.connection.on('ping', () => {
-        this.connection?.pong();
-      });
       this.connection.on('pong', () => {
         // this.logger.logDebug('[OneBot] [WebSocket Client] 收到pong');
       });

packages/napcat-onebot/network/websocket-server.ts

@@ -1,7 +1,6 @@
 import { OB11EmitEventContent, OB11NetworkReloadType } from './index';
 import { URL } from 'url';
 import { RawData, WebSocket, WebSocketServer } from 'ws';
-import { Mutex } from 'async-mutex';
 import { OB11Response } from '@/napcat-onebot/action/OneBotAction';
 import { ActionName } from '@/napcat-onebot/action/router';
 import { NapCatCore } from 'napcat-core';
@@ -17,7 +16,6 @@ import json5 from 'json5';
 export class OB11WebSocketServerAdapter extends IOB11NetworkAdapter<WebsocketServerConfig> {
   wsServer?: WebSocketServer;
   wsClients: WebSocket[] = [];
-  wsClientsMutex = new Mutex();
   private heartbeatIntervalId: NodeJS.Timeout | null = null;
   wsClientWithEvent: WebSocket[] = [];
 
@@ -58,14 +56,10 @@ export class OB11WebSocketServerAdapter extends IOB11NetworkAdapter<WebsocketSer
       wsClient.on('message', (message) => {
         this.handleMessage(wsClient, message).then().catch(e => this.logger.logError(e));
       });
-      wsClient.on('ping', () => {
-        wsClient.pong();
-      });
       wsClient.on('pong', () => {
         // this.logger.logDebug('[OneBot] [WebSocket Server] Pong received');
       });
       wsClient.once('close', () => {
-        this.wsClientsMutex.runExclusive(async () => {
         const NormolIndex = this.wsClients.indexOf(wsClient);
         if (NormolIndex !== -1) {
           this.wsClients.splice(NormolIndex, 1);
@@ -78,8 +72,6 @@ export class OB11WebSocketServerAdapter extends IOB11NetworkAdapter<WebsocketSer
           this.stopHeartbeat();
         }
       });
-      });
-      await this.wsClientsMutex.runExclusive(async () => {
       if (!isApiConnect) {
         this.wsClientWithEvent.push(wsClient);
       }
@@ -87,7 +79,6 @@ export class OB11WebSocketServerAdapter extends IOB11NetworkAdapter<WebsocketSer
       if (this.wsClientWithEvent.length > 0) {
         this.startHeartbeat();
       }
-      });
     }).on('error', (err) => this.logger.log('[OneBot] [WebSocket Server] Server Error:', err.message));
   }
 
@@ -100,7 +91,6 @@ export class OB11WebSocketServerAdapter extends IOB11NetworkAdapter<WebsocketSer
   }
 
   async onEvent<T extends OB11EmitEventContent> (event: T) {
-    this.wsClientsMutex.runExclusive(async () => {
     const promises = this.wsClientWithEvent.map((wsClient) => {
       return new Promise<void>((resolve, reject) => {
         if (wsClient.readyState === WebSocket.OPEN) {
@@ -112,7 +102,6 @@ export class OB11WebSocketServerAdapter extends IOB11NetworkAdapter<WebsocketSer
       });
     });
     await Promise.allSettled(promises);
-    });
   }
 
   open () {
@@ -136,25 +125,19 @@ export class OB11WebSocketServerAdapter extends IOB11NetworkAdapter<WebsocketSer
       }
     });
     this.stopHeartbeat();
-    await this.wsClientsMutex.runExclusive(async () => {
+    this.wsClients.forEach((wsClient) => wsClient.close());
-      this.wsClients.forEach((wsClient) => {
-        wsClient.close();
-      });
     this.wsClients = [];
     this.wsClientWithEvent = [];
-    });
   }
 
   private startHeartbeat () {
     if (this.heartbeatIntervalId || this.config.heartInterval <= 0) return;
     this.heartbeatIntervalId = setInterval(() => {
-      this.wsClientsMutex.runExclusive(async () => {
       this.wsClientWithEvent.forEach((wsClient) => {
         if (wsClient.readyState === WebSocket.OPEN) {
           wsClient.send(JSON.stringify(new OB11HeartbeatEvent(this.core, this.config.heartInterval, this.core.selfInfo.online ?? true, true)));
         }
       });
-      });
     }, this.config.heartInterval);
   }
 

packages/napcat-onebot/package.json

@@ -26,7 +26,6 @@
         "express": "^5.0.0",
         "ws": "^8.18.3",
         "file-type": "^21.0.0",
-        "async-mutex": "^0.5.0",
         "napcat-protobuf": "workspace:*",
         "json5": "^2.2.3",
         "napcat-core": "workspace:*",

packages/napcat-types/external-shims.d.ts

@@ -61,20 +61,6 @@ declare module 'yaml' {
   export const stringify: (...args: any[]) => any;
 }
 
-declare module 'async-mutex' {
-  export class Mutex {
-    acquire (): Promise<() => void>;
-    runExclusive<T> (callback: () => T | Promise<T>): Promise<T>;
-  }
-  export class Semaphore {
-    acquire (): Promise<[() => void, number]>;
-    runExclusive<T> (callback: () => T | Promise<T>): Promise<T>;
-    release (): void;
-  }
-  const _async_mutex_default: { Mutex: typeof Mutex; Semaphore: typeof Semaphore; };
-  export default _async_mutex_default;
-}
-
 declare module 'napcat-protobuf' {
   export class NapProtoMsg<T = any> {
     constructor (schema: any);

packages/napcat-types/scripts/post-build.mjs

@@ -35,9 +35,6 @@ const EXTERNAL_TYPE_REPLACEMENTS = {
   'ValidateFunction<T>': 'any',
   // inversify
   'Container': 'any',
-  // async-mutex
-  'Mutex': 'any',
-  'Semaphore': 'any',
   // napcat-protobuf
   'NapProtoDecodeStructType': 'any',
   'NapProtoEncodeStructType': 'any',
@@ -90,15 +87,15 @@ function replaceExternalTypes (content) {
   // 使用类型上下文的模式匹配
   const typeContextPatterns = [
     // : Type
-    /:\s*(WebSocket|WebSocketServer|RawData|Ajv|AnySchema|ValidateFunction|Container|Mutex|Semaphore|NapProtoDecodeStructType|NapProtoEncodeStructType|Express|Request|Response|NextFunction)(?=\s*[;,)\]\}|&]|$)/g,
+    /:\s*(WebSocket|WebSocketServer|RawData|Ajv|AnySchema|ValidateFunction|Container|NapProtoDecodeStructType|NapProtoEncodeStructType|Express|Request|Response|NextFunction)(?=\s*[;,)\]\}|&]|$)/g,
     // <Type>
-    /<(WebSocket|WebSocketServer|RawData|Ajv|AnySchema|ValidateFunction|Container|Mutex|Semaphore|NapProtoDecodeStructType|NapProtoEncodeStructType|Express|Request|Response|NextFunction)>/g,
+    /<(WebSocket|WebSocketServer|RawData|Ajv|AnySchema|ValidateFunction|Container|NapProtoDecodeStructType|NapProtoEncodeStructType|Express|Request|Response|NextFunction)>/g,
     // Type[]
-    /(WebSocket|WebSocketServer|RawData|Ajv|AnySchema|ValidateFunction|Container|Mutex|Semaphore|NapProtoDecodeStructType|NapProtoEncodeStructType|Express|Request|Response|NextFunction)\[\]/g,
+    /(WebSocket|WebSocketServer|RawData|Ajv|AnySchema|ValidateFunction|Container|NapProtoDecodeStructType|NapProtoEncodeStructType|Express|Request|Response|NextFunction)\[\]/g,
     // extends Type
-    /extends\s+(WebSocket|WebSocketServer|RawData|Ajv|AnySchema|ValidateFunction|Container|Mutex|Semaphore|NapProtoDecodeStructType|NapProtoEncodeStructType|Express|Request|Response|NextFunction)(?=\s*[{,])/g,
+    /extends\s+(WebSocket|WebSocketServer|RawData|Ajv|AnySchema|ValidateFunction|Container|NapProtoDecodeStructType|NapProtoEncodeStructType|Express|Request|Response|NextFunction)(?=\s*[{,])/g,
     // implements Type
-    /implements\s+(WebSocket|WebSocketServer|RawData|Ajv|AnySchema|ValidateFunction|Container|Mutex|Semaphore|NapProtoDecodeStructType|NapProtoEncodeStructType|Express|Request|Response|NextFunction)(?=\s*[{,])/g,
+    /implements\s+(WebSocket|WebSocketServer|RawData|Ajv|AnySchema|ValidateFunction|Container|NapProtoDecodeStructType|NapProtoEncodeStructType|Express|Request|Response|NextFunction)(?=\s*[{,])/g,
   ];
 
   for (const pattern of typeContextPatterns) {

pnpm-lock.yaml

@@ -176,9 +176,6 @@ importers:
       ajv:
         specifier: ^8.13.0
         version: 8.17.1
-      async-mutex:
-        specifier: ^0.5.0
-        version: 0.5.0
       cors:
         specifier: ^2.8.5
         version: 2.8.5
@@ -3563,9 +3560,6 @@ packages:
     resolution: {integrity: sha512-hsU18Ae8CDTR6Kgu9DYf0EbCr/a5iGL0rytQDobUcdpYOKokk8LEjVphnXkDkgpi0wYVsqrXuP0bZxJaTqdgoA==}
     engines: {node: '>= 0.4'}
 
-  async-mutex@0.5.0:
-    resolution: {integrity: sha512-1A94B18jkJ3DYq284ohPxoXbfTA5HsQ7/Mf4DEhcyLx3Bz27Rh59iScbB6EPiP+B+joue6YCxcMXSbFC1tZKwA==}
-
   async@3.2.6:
     resolution: {integrity: sha512-htCUDlxyyCLMgaM3xXg0C0LW2xqfuQ6p05pCEIsXuyQ+a1koYKTuBMzRNwmybfLgvJDMd0r1LTn4+E0Ti6C2AA==}
 
@@ -10443,10 +10437,6 @@ snapshots:
 
   async-function@1.0.0: {}
 
-  async-mutex@0.5.0:
-    dependencies:
-      tslib: 2.8.1
-
   async@3.2.6: {}
 
   asynckit@0.4.0: {}