Merge pull request #701 from Stapxs/main

HTTP SSE 消息上报模式

src/onebot/config/config.ts

@@ -59,6 +59,13 @@ export const httpServerDefaultConfigs = createDefaultAdapterConfig({
 });
 export type HttpServerConfig = typeof httpServerDefaultConfigs;
 
+export const httpSseServerDefaultConfigs = createDefaultAdapterConfig({
+    ...httpServerDefaultConfigs,
+    name: 'http-sse-server',
+    reportSelfMessage: false,
+});
+export type HttpSseServerConfig = typeof httpSseServerDefaultConfigs;
+
 export const httpClientDefaultConfigs = createDefaultAdapterConfig({
     name: 'http-client',
     enable: false as boolean,
@@ -99,6 +106,7 @@ export type WebsocketClientConfig = typeof websocketClientDefaultConfigs;
 
 export interface NetworkConfig {
     httpServers: Array<HttpServerConfig>;
+    httpSseServers: Array<HttpSseServerConfig>;
     httpClients: Array<HttpClientConfig>;
     websocketServers: Array<WebsocketServerConfig>;
     websocketClients: Array<WebsocketClientConfig>;
@@ -120,6 +128,7 @@ const createDefaultConfig = <T>(config: T): T => config;
 export const defaultOneBotConfigs = createDefaultConfig<OneBotConfig>({
     network: {
         httpServers: [],
+        httpSseServers: [],
         httpClients: [],
         websocketServers: [],
         websocketClients: [],

src/onebot/index.ts

@@ -53,6 +53,7 @@ import {
 import { OB11Message } from './types';
 import { OB11PluginAdapter } from './network/plugin';
 import { IOB11NetworkAdapter } from "@/onebot/network/adapter";
+import { OB11ActiveHttpSSEAdapter } from './network/active-http-sse';
 
 //OneBot实现类
 export class NapCatOneBot11Adapter {
@@ -87,6 +88,9 @@ export class NapCatOneBot11Adapter {
         for (const key of ob11Config.network.httpServers) {
             log += `HTTP服务: ${key.host}:${key.port}, : ${key.enable ? '已启动' : '未启动'}\n`;
         }
+        for (const key of ob11Config.network.httpSseServers) {
+            log += `HTTP-SSE服务: ${key.host}:${key.port}, : ${key.enable ? '已启动' : '未启动'}\n`;
+        }
         for (const key of ob11Config.network.httpClients) {
             log += `HTTP上报服务: ${key.url}, : ${key.enable ? '已启动' : '未启动'}\n`;
         }
@@ -125,6 +129,13 @@ export class NapCatOneBot11Adapter {
                 );
             }
         }
+        for(const key of ob11Config.network.httpSseServers){
+            if(key.enable) {
+                this.networkManager.registerAdapter(
+                    new OB11ActiveHttpSSEAdapter(key.name, key, this.core, this, this.actions)
+                );
+            }
+        }
         for (const key of ob11Config.network.httpClients) {
             if (key.enable) {
                 this.networkManager.registerAdapter(
@@ -389,7 +400,7 @@ export class NapCatOneBot11Adapter {
                     ) {
                         this.context.logger.logDebug('有加群请求');
                         try {
-                            let requestUin = await this.core.apis.UserApi.getUinByUidV2(notify.user1.uid);
+                            const requestUin = await this.core.apis.UserApi.getUinByUidV2(notify.user1.uid);
                             const groupRequestEvent = new OB11GroupRequestEvent(
                                 this.core,
                                 parseInt(notify.group.groupCode),

src/onebot/network/active-http-sse.ts

@@ -0,0 +1,34 @@
+import { OB11EmitEventContent } from './index';
+import { Request, Response } from 'express';
+import { OB11Response } from '@/onebot/action/OneBotAction';
+import { OB11PassiveHttpAdapter } from './passive-http';
+
+export class OB11ActiveHttpSSEAdapter extends OB11PassiveHttpAdapter {
+    private sseClients: Response[] = [];
+
+    async handleRequest(req: Request, res: Response): Promise<any> {
+        if (req.path === '/_events') {
+            return this.createSseSupport(req, res);
+        } else {
+            super.httpApiRequest(req, res);
+        }
+    }
+
+    private async createSseSupport(req: Request, res: Response) {
+        res.setHeader('Content-Type', 'text/event-stream');
+        res.setHeader('Cache-Control', 'no-cache');
+        res.setHeader('Connection', 'keep-alive');
+        res.flushHeaders();
+
+        this.sseClients.push(res);
+        req.on('close', () => {
+            this.sseClients = this.sseClients.filter((client) => client !== res);
+        });
+    }
+
+    onEvent<T extends OB11EmitEventContent>(event: T) {
+        this.sseClients.forEach((res) => {
+            res.write(`data: ${JSON.stringify(event)}\n\n`);
+        });
+    }
+}

src/onebot/network/passive-http.ts

@@ -1,4 +1,4 @@
-import { OB11NetworkReloadType } from './index';
+import { OB11EmitEventContent, OB11NetworkReloadType } from './index';
 import express, { Express, Request, Response } from 'express';
 import http from 'http';
 import { NapCatCore } from '@/core';
@@ -17,7 +17,7 @@ export class OB11PassiveHttpAdapter extends IOB11NetworkAdapter<HttpServerConfig
         super(name, config, core, obContext, actions);
     }
 
-    onEvent() {
+    onEvent<T extends OB11EmitEventContent>(event: T) {
         // http server is passive, no need to emit event
     }
 
@@ -82,12 +82,7 @@ export class OB11PassiveHttpAdapter extends IOB11NetworkAdapter<HttpServerConfig
         }
     }
 
-    private async handleRequest(req: Request, res: Response) {
-        if (!this.isEnable) {
-            this.core.context.logger.log(`[OneBot] [HTTP Server Adapter] Server is closed`);
-            return res.json(OB11Response.error('Server is closed', 200));
-        }
-
+    async httpApiRequest(req: Request, res: Response) {
         let payload = req.body;
         if (req.method == 'get') {
             payload = req.query;
@@ -113,6 +108,15 @@ export class OB11PassiveHttpAdapter extends IOB11NetworkAdapter<HttpServerConfig
         }
     }
 
+    async handleRequest(req: Request, res: Response) {
+        if (!this.isEnable) {
+            this.core.context.logger.log(`[OneBot] [HTTP Server Adapter] Server is closed`);
+            return res.json(OB11Response.error('Server is closed', 200));
+        }
+
+        return this.httpApiRequest(req, res);
+    }
+
     async reload(newConfig: HttpServerConfig) {
         const wasEnabled = this.isEnable;
         const oldPort = this.config.port;