http sse 消息上报模式

This commit is contained in:
stapxs
2025-01-09 13:22:46 +08:00
parent 15156bac1e
commit 88fd1f9eb1
4 changed files with 84 additions and 4 deletions

View File

@@ -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: [],

View File

@@ -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),

View File

@@ -0,0 +1,60 @@
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 (!this.isEnable) {
this.core.context.logger.log(`[OneBot] [HTTP Server Adapter] Server is closed`);
return res.json(OB11Response.error('Server is closed', 200));
}
let payload = req.body;
if (req.method == 'get') {
payload = req.query;
} else if (req.query) {
payload = { ...req.query, ...req.body };
}
if (req.path === '' || req.path === '/') {
const hello = OB11Response.ok({});
hello.message = 'NapCat4 Ss Running';
return res.json(hello);
}
if (req.path === '/_events') {
return this.createSseSupport(req, res);
}
const actionName = req.path.split('/')[1];
const action = this.actions.get(actionName as any);
if (action) {
try {
const result = await action.handle(payload, this.name);
return res.json(result);
} catch (error: any) {
return res.json(OB11Response.error(error?.stack?.toString() || error?.message || 'Error Handle', 200));
}
} else {
return res.json(OB11Response.error('不支持的Api ' + actionName, 200));
}
}
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`);
});
}
}

View File

@@ -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,7 +82,7 @@ export class OB11PassiveHttpAdapter extends IOB11NetworkAdapter<HttpServerConfig
}
}
private async handleRequest(req: Request, res: Response) {
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));