Merge branch 'main' of https://github.com/NapNeko/NapCatQQ

src/onebot/network/active-http.ts

@@ -12,28 +12,30 @@ export class OB11ActiveHttpAdapter implements IOB11NetworkAdapter {
     heartbeatInterval: number;
     secret: string | undefined;
     coreContext: NapCatCore;
-    onebotContext: NapCatOneBot11Adapter;
+    obContext: NapCatOneBot11Adapter;
     logger: LogWrapper;
+    isOpen: boolean = false;
 
     constructor(url: string, heartbeatInterval: number, secret: string | undefined, coreContext: NapCatCore, onebotContext: NapCatOneBot11Adapter) {
         this.heartbeatInterval = heartbeatInterval;
         this.url = url;
         this.secret = secret;
         this.coreContext = coreContext;
-        this.onebotContext = onebotContext;
+        this.obContext = onebotContext;
         this.logger = coreContext.context.logger;
     }
 
-    registerHeartBeat() {
-        // HttpPost 心跳
-    }
     registerActionMap(actionMap: Map<string, BaseAction<any, any>>) {
     }
+
     registerAction<T extends BaseAction<P, R>, P, R>(action: T) {
         // Passive http adapter does not need to register actions
     }
 
     onEvent<T extends OB11EmitEventContent>(event: T) {
+        if (!this.isOpen) {
+            return;
+        }
         const headers: Record<string, string> = {
             'Content-Type': 'application/json',
             'x-self-id': this.coreContext.selfInfo.uin,
@@ -66,13 +68,11 @@ export class OB11ActiveHttpAdapter implements IOB11NetworkAdapter {
         });
     }
 
-    async open() {
+    open() {
-        // HTTP adapter does not need to establish a persistent connection
+        this.isOpen = true;
-        //console.log('HTTP adapter is ready to send events.');
     }
 
     close() {
-        // HTTP adapter does not need to close a persistent connection
+        this.isOpen = false;
-        // console.log('HTTP adapter does not maintain a persistent connection.');
     }
 }

src/onebot/network/active-websocket.ts

@@ -1,48 +1,40 @@
 import { IOB11NetworkAdapter, OB11EmitEventContent } from '@/onebot/network/index';
-import { WebSocket as NodeWebSocket } from 'ws';
+import { WebSocket } from 'ws';
 import BaseAction from '@/onebot/action/BaseAction';
 import { sleep } from '@/common/utils/helper';
 import { OB11HeartbeatEvent } from '../event/meta/OB11HeartbeatEvent';
 import { NapCatCore } from '@/core';
 import { NapCatOneBot11Adapter } from '../main';
-import { OB11Response } from '../action/OB11Response';
+import { ActionName } from '@/onebot/action/types';
+import { OB11Response } from '@/onebot/action/OB11Response';
 import { LogWrapper } from '@/common/utils/log';
-import { ActionName } from '../action/types';
 
 export class OB11ActiveWebSocketAdapter implements IOB11NetworkAdapter {
     url: string;
     reconnectIntervalInMillis: number;
     isClosed: boolean = false;
-    private connection: NodeWebSocket | null = null;
-    private actionMap: Map<string, BaseAction<any, any>> = new Map();
     heartbeatInterval: number;
-    private heartbeatTimer: NodeJS.Timeout | null = null;
+    obContext: NapCatOneBot11Adapter;
-    onebotContext: NapCatOneBot11Adapter;
     coreContext: NapCatCore;
-    token: string;
     logger: LogWrapper;
-
+    private connection: WebSocket | null = null;
-    constructor(url: string, reconnectIntervalInMillis: number, heartbeatInterval: number, token: string, coreContext: NapCatCore, onebotContext: NapCatOneBot11Adapter) {
+    private actionMap: Map<string, BaseAction<any, any>> = new Map();
+    private heartbeatTimer: NodeJS.Timeout | null = null;
+    private readonly token: string;
+    
+    constructor(url: string, reconnectIntervalInMillis: number, heartbeatInterval: number, token:string, coreContext: NapCatCore, onebotContext: NapCatOneBot11Adapter) {
         this.url = url;
-        this.logger = coreContext.context.logger;
         this.token = token;
         this.heartbeatInterval = heartbeatInterval;
         this.reconnectIntervalInMillis = reconnectIntervalInMillis;
         this.coreContext = coreContext;
-        this.onebotContext = onebotContext;
+        this.obContext = onebotContext;
+        this.logger = coreContext.context.logger;
     }
+
     registerActionMap(actionMap: Map<string, BaseAction<any, any>>) {
         this.actionMap = actionMap;
     }
-    registerHeartBeat() {
-        if (this.connection) {
-            this.heartbeatTimer = setInterval(() => {
-                if (this.connection && this.connection.readyState === NodeWebSocket.OPEN) {
-                    this.connection.send(JSON.stringify(new OB11HeartbeatEvent(this.coreContext, this.heartbeatInterval, this.coreContext.selfInfo.online, true)));
-                }
-            }, this.heartbeatInterval);
-        }
-    }
 
     registerAction<T extends BaseAction<P, R>, P, R>(action: T) {
         this.actionMap.set(action.actionName, action);
@@ -63,7 +55,7 @@ export class OB11ActiveWebSocketAdapter implements IOB11NetworkAdapter {
 
     close() {
         if (this.isClosed) {
-            this.logger.logError('Cannot close a closed WebSocket connection');
+            throw new Error('Cannot close a closed WebSocket connection');
         }
         this.isClosed = true;
         if (this.connection) {
@@ -76,10 +68,26 @@ export class OB11ActiveWebSocketAdapter implements IOB11NetworkAdapter {
         }
     }
 
+    private registerHeartBeat() {
+        if (this.connection) {
+            this.heartbeatTimer = setInterval(() => {
+                if (this.connection && this.connection.readyState === NodeWebSocket.OPEN) {
+                    this.connection.send(JSON.stringify(new OB11HeartbeatEvent(this.coreContext, this.heartbeatInterval, this.coreContext.selfInfo.online, true)));
+                }
+            }, this.heartbeatInterval);
+        }
+    }
+
+    private checkStateAndReply<T>(data: T) {
+        if (this.connection && this.connection.readyState === WebSocket.OPEN) {
+            this.connection.send(JSON.stringify(data));
+        }
+    }
+
     private async tryConnect() {
         while (!this.connection && !this.isClosed) {
             try {
-                this.connection = new NodeWebSocket(this.url, {
+                this.connection = new WebSocket(this.url, {
                     headers: {
                         'X-Self-ID': this.coreContext.selfInfo.uin,
                         'Authorization': `Bearer ${this.token}`,
@@ -104,11 +112,7 @@ export class OB11ActiveWebSocketAdapter implements IOB11NetworkAdapter {
             }
         }
     }
-    WsReply(data: any) {
+
-        if (this.connection?.readyState === NodeWebSocket.OPEN && !this.isClosed) {
-            this.connection?.send(JSON.stringify(data));
-        }
-    }
     private async handleMessage(message: any) {
         let receiveData: { action: ActionName, params?: any, echo?: any } = { action: ActionName.Unknown, params: {} };
         let echo = undefined;
@@ -118,14 +122,15 @@ export class OB11ActiveWebSocketAdapter implements IOB11NetworkAdapter {
                 echo = receiveData.echo;
                 this.logger.logDebug('收到正向Websocket消息', receiveData);
             } catch (e) {
-                this.WsReply(OB11Response.error('json解析失败,请检查数据格式', 1400, echo));
+                this.checkStateAndReply<any>(OB11Response.error('json解析失败,请检查数据格式', 1400, echo));
             }
             receiveData.params = (receiveData?.params) ? receiveData.params : {};//兼容类型验证
-            let retdata = await this.actionMap.get(receiveData.action)?.websocketHandle(receiveData.params, echo || "");
+            const retdata = await this.actionMap.get(receiveData.action)
+                ?.websocketHandle(receiveData.params, echo || '');
             const packet = Object.assign({}, retdata);
-            this.WsReply(packet);
+            this.checkStateAndReply<any>(packet);
         } catch (e) {
-            this.WsReply(OB11Response.error('不支持的api ' + receiveData.action, 1404, echo));
+            this.checkStateAndReply<any>(OB11Response.error('不支持的api ' + receiveData.action, 1404, echo));
         }
     }
-}
+}

src/onebot/network/passive-http.ts

@@ -6,33 +6,30 @@ import { NapCatCore } from '@/core';
 import { NapCatOneBot11Adapter } from '../main';
 
 export class OB11PassiveHttpAdapter implements IOB11NetworkAdapter {
+    token: string;
+    coreContext: NapCatCore;
+    obContext: NapCatOneBot11Adapter;
     private app: Express | undefined;
     private server: http.Server | undefined;
     private isOpen: boolean = false;
     private hasBeenClosed: boolean = false;
     private actionMap: Map<string, BaseAction<any, any>> = new Map();
     private port: number;
-    token: string;
-    coreContext: NapCatCore;
-    onebotContext: NapCatOneBot11Adapter;
 
     constructor(port: number, token: string, coreContext: NapCatCore, onebotContext: NapCatOneBot11Adapter) {
         this.port = port;
         this.token = token;
         this.coreContext = coreContext;
-        this.onebotContext = onebotContext;
+        this.obContext = onebotContext;
     }
 
     registerAction<T extends BaseAction<P, R>, P, R>(action: T) {
         this.actionMap.set(action.actionName, action);
     }
-    registerActionMap(actionMap: Map<string, BaseAction<any, any>>) {
 
+    registerActionMap(actionMap: Map<string, BaseAction<any, any>>) {
         this.actionMap = actionMap;
     }
-    registerHeartBeat() {
-        //空心跳
-    }
 
     onEvent<T extends OB11EmitEventContent>(event: T) {
         // 事件处理逻辑可以在这里实现
@@ -48,6 +45,13 @@ export class OB11PassiveHttpAdapter implements IOB11NetworkAdapter {
         }
     }
 
+    async close() {
+        this.isOpen = false;
+        this.hasBeenClosed = true;
+        this.server?.close();
+        this.app = undefined;
+    }
+
     private initializeServer() {
         this.app = express();
         this.server = http.createServer(this.app);
@@ -80,11 +84,4 @@ export class OB11PassiveHttpAdapter implements IOB11NetworkAdapter {
             res.status(404).send('Action not found');
         }
     }
-
-    async close() {
-        this.isOpen = false;
-        this.hasBeenClosed = true;
-        this.server?.close();
-        this.app = undefined;
-    }
 }

src/onebot/network/passive-websocket.ts

@@ -18,27 +18,15 @@ export class OB11PassiveWebSocketAdapter implements IOB11NetworkAdapter {
     isOpen: boolean = false;
     hasBeenClosed: boolean = false;
     heartbeatInterval: number = 0;
-    private actionMap: Map<string, BaseAction<any, any>> = new Map();
+    obContext: NapCatOneBot11Adapter;
-    onebotContext: NapCatOneBot11Adapter;
     coreContext: NapCatCore;
     logger: LogWrapper;
+    private actionMap: Map<string, BaseAction<any, any>> = new Map();
     private heartbeatIntervalId: NodeJS.Timeout | null = null;
 
-    authorize(token: string | undefined, wsClient: WebSocket, wsReq: IncomingMessage) {
-        if (!token || token.length == 0) return;//客户端未设置密钥
-        let QueryClientToken = urlParse.parse(wsReq?.url || "", true).query.access_token;
-        let HeaderClientToken = wsReq.headers.authorization?.split('Bearer ').pop() || '';
-        let ClientToken = typeof (QueryClientToken) === 'string' && QueryClientToken !== "" ? QueryClientToken : HeaderClientToken;
-        if (ClientToken === token) {
-            return;
-        }
-        wsClient.send(JSON.stringify(OB11Response.res(null, 'failed', 1403, 'token验证失败')));
-        wsClient.close();
-    }
-
     constructor(ip: string, port: number, heartbeatInterval: number, token: string, coreContext: NapCatCore, onebotContext: NapCatOneBot11Adapter) {
         this.coreContext = coreContext;
-        this.onebotContext = onebotContext;
+        this.obContext = onebotContext;
         this.logger = coreContext.context.logger;
 
         this.heartbeatInterval = heartbeatInterval;
@@ -76,18 +64,6 @@ export class OB11PassiveWebSocketAdapter implements IOB11NetworkAdapter {
         this.actionMap.set(action.actionName, action);
     }
 
-    registerHeartBeat() {
-        this.heartbeatIntervalId = setInterval(() => {
-            this.wsClientsMutex.runExclusive(async () => {
-                this.wsClients.forEach((wsClient) => {
-                    if (wsClient.readyState === WebSocket.OPEN) {
-                        wsClient.send(JSON.stringify(new OB11HeartbeatEvent(this.coreContext, this.heartbeatInterval, this.coreContext.selfInfo.online, true)));
-                    }
-                });
-            });
-        }, this.heartbeatInterval);
-    }
-
     onEvent<T extends OB11EmitEventContent>(event: T) {
         this.wsClientsMutex.runExclusive(async () => {
             this.wsClients.forEach((wsClient) => {
@@ -117,17 +93,39 @@ export class OB11PassiveWebSocketAdapter implements IOB11NetworkAdapter {
         }
     }
 
-    async WsReplyAll<T>(data: T) {
+    private registerHeartBeat() {
-        this.wsClientsMutex.runExclusive(async () => {
+        this.heartbeatIntervalId = setInterval(() => {
+            this.wsClientsMutex.runExclusive(async () => {
+                this.wsClients.forEach((wsClient) => {
+                    if (wsClient.readyState === WebSocket.OPEN) {
+                        wsClient.send(JSON.stringify(new OB11HeartbeatEvent(this.coreContext, this.heartbeatInterval, this.coreContext.selfInfo.online, true)));
+                    }
+                });
+            });
+        }, this.heartbeatInterval);
+    }
+
+    private authorize(token: string | undefined, wsClient: WebSocket, wsReq: IncomingMessage) {
+        if (!token || token.length == 0) return;//客户端未设置密钥
+        const QueryClientToken = urlParse.parse(wsReq?.url || '', true).query.access_token;
+        const HeaderClientToken = wsReq.headers.authorization?.split('Bearer ').pop() || '';
+        const ClientToken = typeof (QueryClientToken) === 'string' && QueryClientToken !== '' ? QueryClientToken : HeaderClientToken;
+        if (ClientToken === token) {
+            return;
+        }
+        wsClient.send(JSON.stringify(OB11Response.res(null, 'failed', 1403, 'token验证失败')));
+        wsClient.close();
+    }
+
+    private async checkStateAndAnnounce<T>(data: T) {
+        await this.wsClientsMutex.runExclusive(async () => {
             this.wsClients.forEach((wsClient) => {
-                if (wsClient.readyState === WebSocket.OPEN) {
+                this.checkStateAndReply(data, wsClient);
-                    wsClient.send(JSON.stringify(data));
-                }
             });
         });
     }
 
-    async WsReply<T>(data: T, wsClient: WebSocket) {
+    private checkStateAndReply<T>(data: T, wsClient: WebSocket) {
         if (wsClient.readyState === WebSocket.OPEN) {
             wsClient.send(JSON.stringify(data));
         }
@@ -142,21 +140,21 @@ export class OB11PassiveWebSocketAdapter implements IOB11NetworkAdapter {
                 echo = receiveData.echo;
                 this.logger.logDebug('收到正向Websocket消息', receiveData);
             } catch (e) {
-                this.WsReply<any>(OB11Response.error('json解析失败,请检查数据格式', 1400, echo), wsClient);
+                this.checkStateAndReply<any>(OB11Response.error('json解析失败,请检查数据格式', 1400, echo), wsClient);
             }
             receiveData.params = (receiveData?.params) ? receiveData.params : {};//兼容类型验证
-            let retdata = await this.actionMap.get(receiveData.action)?.websocketHandle(receiveData.params, echo || "");
+            const retdata = await this.actionMap.get(receiveData.action)?.websocketHandle(receiveData.params, echo || '');
             const packet = Object.assign({}, retdata);
-            this.WsReply<any>(packet, wsClient);
+            this.checkStateAndReply<any>(packet, wsClient);
         } catch (e) {
-            this.WsReply<any>(OB11Response.error('不支持的api ' + receiveData.action, 1404, echo), wsClient);
+            this.checkStateAndReply<any>(OB11Response.error('不支持的api ' + receiveData.action, 1404, echo), wsClient);
         }
     }
 
     private wrapEvent<T extends OB11EmitEventContent>(event: T) {
         return {
             type: 'event',
-            data: event
+            data: event,
         };
     }
 }