From 3c42cc17c81953db866ab7c4ff345a3b59b3ab7d Mon Sep 17 00:00:00 2001 From: "Wesley F. Young" Date: Sun, 11 Aug 2024 22:58:57 +0800 Subject: [PATCH 1/3] chore: clean-up --- src/onebot/network/active-http.ts | 8 +-- src/onebot/network/passive-http.ts | 27 ++++----- src/onebot/network/passive-websocket.ts | 74 ++++++++++++------------- 3 files changed, 51 insertions(+), 58 deletions(-) diff --git a/src/onebot/network/active-http.ts b/src/onebot/network/active-http.ts index ca41dc92f..ee166bdf9 100644 --- a/src/onebot/network/active-http.ts +++ b/src/onebot/network/active-http.ts @@ -12,7 +12,7 @@ export class OB11ActiveHttpAdapter implements IOB11NetworkAdapter { heartbeatInterval: number; secret: string | undefined; coreContext: NapCatCore; - onebotContext: NapCatOneBot11Adapter; + obContext: NapCatOneBot11Adapter; logger: LogWrapper; constructor(url: string, heartbeatInterval: number, secret: string | undefined, coreContext: NapCatCore, onebotContext: NapCatOneBot11Adapter) { @@ -20,15 +20,13 @@ export class OB11ActiveHttpAdapter implements IOB11NetworkAdapter { 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>) { } + registerAction, P, R>(action: T) { // Passive http adapter does not need to register actions } diff --git a/src/onebot/network/passive-http.ts b/src/onebot/network/passive-http.ts index 859aa7670..88e5be351 100644 --- a/src/onebot/network/passive-http.ts +++ b/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> = 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, P, R>(action: T) { this.actionMap.set(action.actionName, action); } - registerActionMap(actionMap: Map>) { + registerActionMap(actionMap: Map>) { this.actionMap = actionMap; } - registerHeartBeat() { - //空心跳 - } onEvent(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; - } } diff --git a/src/onebot/network/passive-websocket.ts b/src/onebot/network/passive-websocket.ts index 8a5856cd0..80482715a 100644 --- a/src/onebot/network/passive-websocket.ts +++ b/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> = new Map(); - onebotContext: NapCatOneBot11Adapter; + obContext: NapCatOneBot11Adapter; coreContext: NapCatCore; logger: LogWrapper; + private actionMap: Map> = 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(event: T) { this.wsClientsMutex.runExclusive(async () => { this.wsClients.forEach((wsClient) => { @@ -117,17 +93,39 @@ export class OB11PassiveWebSocketAdapter implements IOB11NetworkAdapter { } } - async WsReplyAll(data: T) { - this.wsClientsMutex.runExclusive(async () => { + private 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); + } + + 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(data: T) { + await this.wsClientsMutex.runExclusive(async () => { this.wsClients.forEach((wsClient) => { - if (wsClient.readyState === WebSocket.OPEN) { - wsClient.send(JSON.stringify(data)); - } + this.checkStateAndReply(data, wsClient); }); }); } - async WsReply(data: T, wsClient: WebSocket) { + private checkStateAndReply(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(OB11Response.error('json解析失败,请检查数据格式', 1400, echo), wsClient); + this.checkStateAndReply(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(packet, wsClient); + this.checkStateAndReply(packet, wsClient); } catch (e) { - this.WsReply(OB11Response.error('不支持的api ' + receiveData.action, 1404, echo), wsClient); + this.checkStateAndReply(OB11Response.error('不支持的api ' + receiveData.action, 1404, echo), wsClient); } } private wrapEvent(event: T) { return { type: 'event', - data: event + data: event, }; } } From a955937e02407a2aa29e5c76a9d37d30957eebdc Mon Sep 17 00:00:00 2001 From: "Wesley F. Young" Date: Sun, 11 Aug 2024 23:09:00 +0800 Subject: [PATCH 2/3] feat: complete active websocket --- src/onebot/network/active-websocket.ts | 64 +++++++++++++++++--------- 1 file changed, 43 insertions(+), 21 deletions(-) diff --git a/src/onebot/network/active-websocket.ts b/src/onebot/network/active-websocket.ts index 2ed2ebfc9..3a2f4af3b 100644 --- a/src/onebot/network/active-websocket.ts +++ b/src/onebot/network/active-websocket.ts @@ -1,41 +1,38 @@ import { IOB11NetworkAdapter, OB11EmitEventContent } from '@/onebot/network/index'; -import { WebSocket as NodeWebSocket } from 'ws'; +import { WebSocket, WebSocket as NodeWebSocket } 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 { ActionName } from '@/onebot/action/types'; +import { OB11Response } from '@/onebot/action/OB11Response'; +import { LogWrapper } from '@/common/utils/log'; export class OB11ActiveWebSocketAdapter implements IOB11NetworkAdapter { url: string; reconnectIntervalInMillis: number; isClosed: boolean = false; + heartbeatInterval: number; + obContext: NapCatOneBot11Adapter; + coreContext: NapCatCore; + logger: LogWrapper; private connection: NodeWebSocket | null = null; private actionMap: Map> = new Map(); - heartbeatInterval: number; private heartbeatTimer: NodeJS.Timeout | null = null; - onebotContext: NapCatOneBot11Adapter; - coreContext: NapCatCore; constructor(url: string, reconnectIntervalInMillis: number, heartbeatInterval: number, coreContext: NapCatCore, onebotContext: NapCatOneBot11Adapter) { this.url = url; this.heartbeatInterval = heartbeatInterval; this.reconnectIntervalInMillis = reconnectIntervalInMillis; this.coreContext = coreContext; - this.onebotContext = onebotContext; + this.obContext = onebotContext; + this.logger = coreContext.context.logger; } + registerActionMap(actionMap: Map>) { 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, P, R>(action: T) { this.actionMap.set(action.actionName, action); @@ -69,6 +66,22 @@ 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(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 { @@ -91,15 +104,24 @@ export class OB11ActiveWebSocketAdapter implements IOB11NetworkAdapter { } } - private handleMessage(data: any) { + private async handleMessage(message: any) { + let receiveData: { action: ActionName, params?: any, echo?: any } = { action: ActionName.Unknown, params: {} }; + let echo = undefined; try { - const message = JSON.parse(data); - const action = this.actionMap.get(message.actionName); - if (action) { - action.handle(message.payload); + try { + receiveData = JSON.parse(message.toString()); + echo = receiveData.echo; + this.logger.logDebug('收到正向Websocket消息', receiveData); + } catch (e) { + this.checkStateAndReply(OB11Response.error('json解析失败,请检查数据格式', 1400, echo)); } + receiveData.params = (receiveData?.params) ? receiveData.params : {};//兼容类型验证 + const retdata = await this.actionMap.get(receiveData.action) + ?.websocketHandle(receiveData.params, echo || ''); + const packet = Object.assign({}, retdata); + this.checkStateAndReply(packet); } catch (e) { - console.error('Failed to handle message:', e); + this.checkStateAndReply(OB11Response.error('不支持的api ' + receiveData.action, 1404, echo)); } } -} \ No newline at end of file +} From 57821b839ec494aec8ec7f719a6b95f2e61a3c6c Mon Sep 17 00:00:00 2001 From: "Wesley F. Young" Date: Sun, 11 Aug 2024 23:09:59 +0800 Subject: [PATCH 3/3] feat: implement ActiveHttp.open / close --- src/onebot/network/active-http.ts | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/src/onebot/network/active-http.ts b/src/onebot/network/active-http.ts index ee166bdf9..c170aa2b0 100644 --- a/src/onebot/network/active-http.ts +++ b/src/onebot/network/active-http.ts @@ -14,6 +14,7 @@ export class OB11ActiveHttpAdapter implements IOB11NetworkAdapter { coreContext: NapCatCore; obContext: NapCatOneBot11Adapter; logger: LogWrapper; + isOpen: boolean = false; constructor(url: string, heartbeatInterval: number, secret: string | undefined, coreContext: NapCatCore, onebotContext: NapCatOneBot11Adapter) { this.heartbeatInterval = heartbeatInterval; @@ -32,6 +33,9 @@ export class OB11ActiveHttpAdapter implements IOB11NetworkAdapter { } onEvent(event: T) { + if (!this.isOpen) { + return; + } const headers: Record = { 'Content-Type': 'application/json', 'x-self-id': this.coreContext.selfInfo.uin, @@ -64,13 +68,11 @@ export class OB11ActiveHttpAdapter implements IOB11NetworkAdapter { }); } - async open() { - // HTTP adapter does not need to establish a persistent connection - //console.log('HTTP adapter is ready to send events.'); + open() { + this.isOpen = true; } close() { - // HTTP adapter does not need to close a persistent connection - // console.log('HTTP adapter does not maintain a persistent connection.'); + this.isOpen = false; } }