kevisual/router
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Refactor server implementation to support Bun and Node environments

Browse Source 
- Introduced `ServerNode` and `BunServer` classes to handle server logic for Node and Bun respectively.
- Updated `App` class to initialize the appropriate server based on the runtime environment.
- Enhanced `parseBody` function to handle request body parsing for both environments.
- Modified WebSocket handling to support Bun's WebSocket upgrade mechanism.
- Improved error handling and response structure across the server implementation.
- Added support for custom middleware in the server's request handling.
- Refactored server base functionality into `ServerBase` for better code organization.
- Updated type definitions to reflect changes in server options and listener handling.
- Added a new demo for testing the server functionality with various request types.
This commit is contained in:
熊潇
2025-12-20 05:11:51 +08:00
parent e1a53c01ea
commit a6a7e74559
21 changed files with 1173 additions and 454 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
src/app.ts
View File

@@ -1,20 +1,21 @@
import { QueryRouter, Route, RouteContext, RouteOpts } from './route.ts';
import { Server, ServerOpts, HandleCtx } from './server/server.ts';
import { WsServer } from './server/ws-server.ts';
import { ServerNode, ServerNodeOpts } from './server/server.ts';
import { HandleCtx } from './server/server-base.ts';
import { ServerType } from './server/server-type.ts';
import { CustomError } from './result/error.ts';
import { handleServer } from './server/handle-server.ts';
import { IncomingMessage, ServerResponse } from 'http';
import { isBun } from './utils/is-engine.ts';
import { BunServer } from './server/server-bun.ts';
type RouterHandle = (msg: { path: string;[key: string]: any }) => { code: string; data?: any; message?: string;[key: string]: any };
type AppOptions<T = {}> = {
router?: QueryRouter;
server?: Server;
server?: ServerType;
/** handle msg 关联 */
routerHandle?: RouterHandle;
routerContext?: RouteContext<T>;
serverOptions?: ServerOpts;
io?: boolean;
ioOpts?: { routerHandle?: RouterHandle; routerContext?: RouteContext<T>; path?: string };
serverOptions?: ServerNodeOpts;
};
export type AppRouteContext<T = {}> = HandleCtx & RouteContext<T> & { app: App<T> };
@@ -25,18 +26,22 @@ export type AppRouteContext<T = {}> = HandleCtx & RouteContext<T> & { app: App<T
*/
export class App<U = {}> {
router: QueryRouter;
server: Server;
io: WsServer;
server: ServerType;
constructor(opts?: AppOptions<U>) {
const router = opts?.router || new QueryRouter();
const server = opts?.server || new Server(opts?.serverOptions || {});
let server = opts?.server;
if (!server) {
const serverOptions = opts?.serverOptions || {};
if (!isBun) {
server = new ServerNode(serverOptions)
} else {
server = new BunServer(serverOptions)
}
}
server.setHandle(router.getHandle(router, opts?.routerHandle, opts?.routerContext));
router.setContext({ needSerialize: true, ...opts?.routerContext });
this.router = router;
this.server = server;
if (opts?.io) {
this.io = new WsServer(server, opts?.ioOpts);
}
}
listen(port: number, hostname?: string, backlog?: number, listeningListener?: () => void): void;
listen(port: number, hostname?: string, listeningListener?: () => void): void;
@@ -49,9 +54,6 @@ export class App<U = {}> {
listen(...args: any[]) {
// @ts-ignore
this.server.listen(...args);
if (this.io) {
this.io.listen();
}
}
use(path: string, fn: (ctx: any) => any, opts?: RouteOpts) {
const route = new Route(path, '', opts);
@@ -130,7 +132,10 @@ export class App<U = {}> {
if (!this.server) {
throw new Error('Server is not initialized');
}
this.server.on(fn);
this.server.on({
id: 'app-request-listener',
fun: fn
});
}
}

4
src/index.ts
View File

@@ -5,7 +5,7 @@ export type { RouteContext, RouteOpts, RouteMiddleware } from './route.ts';
export type { Run } from './route.ts';
export { Server, handleServer } from './server/index.ts';
export { ServerNode, handleServer } from './server/index.ts';
/**
* 自定义错误
*/
@@ -13,7 +13,7 @@ export { CustomError } from './result/error.ts';
export { createSchema } from './validator/index.ts';
export type { Rule, Schema, } from './validator/index.ts';
export type { Rule, Schema, } from './validator/index.ts';
export { App } from './app.ts';

1
src/router-simple.ts
View File

@@ -62,6 +62,7 @@ export class SimpleRouter {
}
isSse(req: Req) {
const { headers } = req;
if (!headers) return false;
if (headers['accept'] && headers['accept'].includes('text/event-stream')) {
return true;
}

2
src/server/handle-server.ts
View File

@@ -1,7 +1,7 @@
import type { IncomingMessage, ServerResponse } from 'node:http';
import { parseBody } from './parse-body.ts';
import url from 'node:url';
import { createHandleCtx } from './server.ts';
import { createHandleCtx } from './server-base.ts';
/**
* get params and body

3
src/server/index.ts
View File

@@ -1,2 +1,3 @@
export { Server } from './server.ts';
export { ServerNode } from './server.ts';
export { BunServer } from './server-bun.ts';
export { handleServer } from './handle-server.ts';

77
src/server/parse-body.ts
View File

@@ -1,7 +1,49 @@
import type { IncomingMessage } from 'node:http';
import url from 'node:url';
import { isBun } from '../utils/is-engine.ts';
export const parseBody = async <T = Record<string, any>>(req: IncomingMessage) => {
const resolveBody = (body: string) => {
// 获取 Content-Type 头信息
const contentType = req.headers['content-type'] || '';
const resolve = (data: T) => {
return data;
}
// 处理 application/json
if (contentType.includes('application/json')) {
return resolve(JSON.parse(body) as T);
}
// 处理 application/x-www-form-urlencoded
if (contentType.includes('application/x-www-form-urlencoded')) {
const formData = new URLSearchParams(body);
const result: Record<string, any> = {};
formData.forEach((value, key) => {
// 尝试将值解析为 JSON，如果失败则保留原始字符串
try {
result[key] = JSON.parse(value);
} catch {
result[key] = value;
}
});
return resolve(result as T);
}
// 默认尝试 JSON 解析
try {
return resolve(JSON.parse(body) as T);
} catch {
return resolve({} as T);
}
}
if (isBun) {
// @ts-ignore
const body = req.body;
if (body) {
return resolveBody(body)
}
return {} as T;
}
return new Promise<T>((resolve, reject) => {
const arr: any[] = [];
req.on('data', (chunk) => {
@@ -10,39 +52,8 @@ export const parseBody = async <T = Record<string, any>>(req: IncomingMessage) =
req.on('end', () => {
try {
const body = Buffer.concat(arr).toString();
resolve(resolveBody(body));
// 获取 Content-Type 头信息
const contentType = req.headers['content-type'] || '';
// 处理 application/json
if (contentType.includes('application/json')) {
resolve(JSON.parse(body) as T);
return;
}
// 处理 application/x-www-form-urlencoded
if (contentType.includes('application/x-www-form-urlencoded')) {
const formData = new URLSearchParams(body);
const result: Record<string, any> = {};
formData.forEach((value, key) => {
// 尝试将值解析为 JSON，如果失败则保留原始字符串
try {
result[key] = JSON.parse(value);
} catch {
result[key] = value;
}
});
resolve(result as T);
return;
}
// 默认尝试 JSON 解析
try {
resolve(JSON.parse(body) as T);
} catch {
resolve({} as T);
}
} catch (e) {
resolve({} as T);
}

265
src/server/server-base.ts Normal file
View File

@@ -0,0 +1,265 @@
import type { IncomingMessage, ServerResponse } from 'node:http';
import { handleServer } from './handle-server.ts';
import * as cookie from './cookie.ts';
import { ServerType, Listener, OnListener, ServerOpts } from './server-type.ts';
import { parseIfJson } from '../utils/parse.ts';
type CookieFn = (name: string, value: string, options?: cookie.SerializeOptions, end?: boolean) => void;
export type HandleCtx = {
req: IncomingMessage & { cookies: Record<string, string> };
res: ServerResponse & {
/**
* cookie 函数， end 参数用于设置是否立即设置到响应头，设置了后面的cookie再设置会覆盖前面的
*/
cookie: CookieFn; //
};
};
// 实现函数
export function createHandleCtx(req: IncomingMessage, res: ServerResponse): HandleCtx {
// 用于存储所有的 Set-Cookie 字符串
const cookies: string[] = [];
let handReq = req as HandleCtx['req'];
let handRes = res as HandleCtx['res'];
// 扩展 res.cookie 方法
const cookieFn: CookieFn = (name, value, options = {}, end = true) => {
// 序列化新的 Cookie
const serializedCookie = cookie.serialize(name, value, options);
cookies.push(serializedCookie); // 将新的 Cookie 添加到数组
if (end) {
// 如果设置了 end 参数，则立即设置到响应头
res.setHeader('Set-Cookie', cookies);
}
};
// 解析请求中的现有 Cookie
const parsedCookies = cookie.parse(req.headers.cookie || '');
handReq.cookies = parsedCookies;
handRes.cookie = cookieFn;
// 返回扩展的上下文
return {
req: handReq,
res: handRes,
};
}
export type Cors = {
/**
* @default '*''
*/
origin?: string | undefined;
};
export const resultError = (error: string, code = 500) => {
const r = {
code: code,
message: error,
};
return JSON.stringify(r);
};
export class ServerBase implements ServerType {
path = '/api/router';
_server: any;
handle: ServerOpts['handle'];
_callback: any;
cors: Cors;
listeners: Listener[] = [];
constructor(opts?: ServerOpts) {
this.path = opts?.path || '/api/router';
this.handle = opts?.handle;
this.cors = opts?.cors;
}
listen(port: number, hostname?: string, backlog?: number, listeningListener?: () => void): void;
listen(port: number, hostname?: string, listeningListener?: () => void): void;
listen(port: number, backlog?: number, listeningListener?: () => void): void;
listen(port: number, listeningListener?: () => void): void;
listen(path: string, backlog?: number, listeningListener?: () => void): void;
listen(path: string, listeningListener?: () => void): void;
listen(handle: any, backlog?: number, listeningListener?: () => void): void;
listen(handle: any, listeningListener?: () => void): void;
listen(...args: any[]) {
this.customListen(...args);
}
/**
* child class can custom listen method
* @param args
*/
customListen(...args: any[]) {
console.error('Please use createServer to create server instance');
}
get handleServer() {
return this._callback;
}
set handleServer(fn: any) {
this._callback = fn;
}
get callback() {
return this._callback || this.createCallback();
}
get server() {
return this._server;
}
setHandle(handle?: any) {
this.handle = handle;
}
/**
* get callback
* @returns
*/
createCallback() {
const path = this.path;
const handle = this.handle;
const cors = this.cors;
const that = this;
const _callback = async (req: IncomingMessage, res: ServerResponse) => {
// only handle /api/router
if (req.url === '/favicon.ico') {
return;
}
const listeners = that.listeners || [];
for (const item of listeners) {
const fun = item.fun;
if (typeof fun === 'function' && !item.io) {
await fun(req, res);
}
}
if (res.headersSent) {
// 程序已经在其他地方响应了
return;
}
if (!req.url.startsWith(path)) {
// 判断不是当前路径的请求，交给其他监听处理
return;
}
if (cors) {
res.setHeader('Access-Control-Allow-Origin', cors?.origin || '*'); // 允许所有域名的请求访问，可以根据需要设置具体的域名
res.setHeader('Access-Control-Allow-Headers', 'Origin, X-Requested-With, Content-Type, Accept, Authorization');
res.setHeader('Access-Control-Allow-Methods', 'GET, POST');
if (req.method === 'OPTIONS') {
res.end();
return;
}
}
const url = req.url;
if (!url.startsWith(path)) {
res.end(resultError(`not path:[${path}]`));
return;
}
const messages = await handleServer(req, res);
if (!handle) {
res.end(resultError('no handle'));
return;
}
try {
const end = await handle(messages as any, { req, res });
if (res.writableEnded) {
// 如果响应已经结束，则不进行任何操作
return;
}
res.setHeader('Content-Type', 'application/json; charset=utf-8');
if (typeof end === 'string') {
res.end(end);
} else {
res.end(JSON.stringify(end));
}
} catch (e) {
console.error(e);
res.setHeader('Content-Type', 'application/json; charset=utf-8');
if (e.code && typeof e.code === 'number') {
res.end(resultError(e.message || `Router Server error`, e.code));
} else {
res.end(resultError('Router Server error'));
}
}
};
this._callback = _callback;
return _callback;
}
on(listener: OnListener) {
this.listeners = [];
if (typeof listener === 'function') {
this.listeners.push({ fun: listener });
return;
}
if (Array.isArray(listener)) {
for (const item of listener) {
if (typeof item === 'function') {
this.listeners.push({ fun: item });
} else {
this.listeners.push(item);
}
}
} else {
this.listeners.push(listener);
}
}
async onWebSocket({ ws, message, pathname, token, id }) {
const listener = this.listeners.find((item) => item.path === pathname && item.io);
const data: any = parseIfJson(message);
if (listener) {
const end = (data: any) => {
ws.send(JSON.stringify(data));
}
listener.fun({
data,
token,
id,
ws,
}, { end });
return;
}
if (typeof data === 'string') {
const cleanMessage = data.trim().replace(/^["']|["']$/g, '');
if (cleanMessage === 'close') {
ws.close();
return;
}
if (cleanMessage === 'ping') {
ws.send('pong');
return;
}
}
const { type, data: typeData, ...rest } = data;
if (!type) {
ws.send(JSON.stringify({ code: 500, message: 'type is required' }));
return;
}
const res = {
type,
data: {} as any,
...rest,
};
const end = (data: any, all?: Record<string, any>) => {
const result = {
...res,
data,
...all,
};
ws.send(JSON.stringify(result));
};
// 调用 handle 处理消息
if (type === 'router' && this.handle) {
try {
const result = await this.handle(typeData as any);
end(result);
} catch (e: any) {
if (e.code && typeof e.code === 'number') {
end({
code: e.code,
message: e.message,
});
} else {
end({ code: 500, message: 'Router Server error' });
}
}
} else {
end({ code: 500, message: `${type} server is error` });
}
}
}

174
src/server/server-bun.ts Normal file
View File

@@ -0,0 +1,174 @@
/**
* @title Bun Server Implementation
* @description Bun 服务器实现,提供基于 Bun.serve 的 HTTP 和 WebSocket 功能
* @tags bun, server, websocket, http
* @createdAt 2025-12-20
*/
import type { IncomingMessage, ServerResponse } from 'node:http';
import { ServerType, type ServerOpts, type Cors, Listener } from './server-type.ts';
import { handleServer } from './handle-server.ts';
import { ServerBase } from './server-base.ts';
import { parseIfJson } from '../utils/parse.ts';
const resultError = (error: string, code = 500) => {
const r = {
code: code,
message: error,
};
return JSON.stringify(r);
};
export class BunServer extends ServerBase implements ServerType {
declare _server: any;
declare _callback: any;
declare cors: Cors;
constructor(opts?: ServerOpts) {
super(opts);
}
customListen(...args: any[]): void {
this.listenWithBun(...args);
}
/**
* Bun 运行时的 listen 实现
*/
private listenWithBun(...args: any[]) {
// @ts-ignore - Bun 全局 API
if (typeof Bun === 'undefined' || !Bun.serve) {
throw new Error('Bun runtime not detected');
}
let port: number = 3000;
let hostname: string = 'localhost';
let callback: (() => void) | undefined;
// 解析参数
if (typeof args[0] === 'number') {
port = args[0];
if (typeof args[1] === 'string') {
hostname = args[1];
callback = args[2] || args[3];
} else if (typeof args[1] === 'function') {
callback = args[1];
} else {
callback = args[2];
}
}
const requestCallback = this.createCallback();
const wsPath = this.path;
// @ts-ignore
this._server = Bun.serve({
port,
hostname,
idleTimeout: 0, // 4 minutes idle timeout (max 255 seconds)
fetch: async (request: Request, server: any) => {
const host = request.headers.get('host') || 'localhost';
const url = new URL(request.url, `http://${host}`);
// 处理 WebSocket 升级请求
if (request.headers.get('upgrade') === 'websocket') {
const listenPath = this.listeners.map((item) => item.path).filter((item) => item);
if (listenPath.includes(url.pathname) || url.pathname === wsPath) {
const token = url.searchParams.get('token') || '';
const id = url.searchParams.get('id') || '';
const upgraded = server.upgrade(request, {
data: { url: url, pathname: url.pathname, token, id },
});
if (upgraded) {
return undefined; // WebSocket 连接成功
}
}
return new Response('WebSocket upgrade failed', { status: 400 });
}
// 将 Bun 的 Request 转换为 Node.js 风格的 req/res
return new Promise((resolve) => {
const req: any = {
url: url.pathname + url.search,
method: request.method,
headers: Object.fromEntries(request.headers.entries()),
};
const res: any = {
statusCode: 200,
headersSent: false,
writableEnded: false,
_headers: {} as Record<string, string | string[]>,
writeHead(statusCode: number, headers: Record<string, string | string[]>) {
this.statusCode = statusCode;
for (const key in headers) {
this._headers[key] = headers[key];
}
this.headersSent = true;
},
setHeader(name: string, value: string | string[]) {
this._headers[name] = value;
},
cookie(name: string, value: string, options?: any) {
let cookieString = `${name}=${value}`;
if (options) {
if (options.maxAge) {
cookieString += `; Max-Age=${options.maxAge}`;
}
if (options.domain) {
cookieString += `; Domain=${options.domain}`;
}
if (options.path) {
cookieString += `; Path=${options.path}`;
}
if (options.expires) {
cookieString += `; Expires=${options.expires.toUTCString()}`;
}
if (options.httpOnly) {
cookieString += `; HttpOnly`;
}
if (options.secure) {
cookieString += `; Secure`;
}
if (options.sameSite) {
cookieString += `; SameSite=${options.sameSite}`;
}
}
this.setHeader('Set-Cookie', cookieString);
},
end(data?: string) {
this.writableEnded = true;
resolve(
new Response(data, {
status: this.statusCode,
headers: this._headers as any,
})
);
},
};
// 处理请求体
if (request.method !== 'GET' && request.method !== 'HEAD') {
request.text().then((body) => {
(req as any).body = body;
requestCallback(req, res);
});
} else {
requestCallback(req, res);
}
});
},
websocket: {
open: (ws: any) => {
ws.send('connected');
},
message: async (ws: any, message: string | Buffer) => {
const pathname = ws.data.pathname || '';
const token = ws.data.token || '';
const id = ws.data.id || '';
await this.onWebSocket({ ws, message, pathname, token, id });
},
close: (ws: any) => {
// WebSocket 连接关闭
},
},
});
if (callback) {
callback();
}
}
}

70
src/server/server-type.ts Normal file
View File

@@ -0,0 +1,70 @@
import * as http from 'http';
export type Listener = {
id?: string;
io?: boolean;
path?: string;
fun: (...args: any[]) => Promise<void> | void;
}
export type ListenerFun = (...args: any[]) => Promise<void> | void;
export type OnListener = Listener | Listener[] | ListenerFun | ListenerFun[];
export type Cors = {
/**
* @default '*''
*/
origin?: string | undefined;
};
export type ServerOpts<T = {}> = {
/**path default `/api/router` */
path?: string;
/**handle Fn */
handle?: (msg?: { path: string; key?: string;[key: string]: any }, ctx?: { req: http.IncomingMessage; res: http.ServerResponse }) => any;
cors?: Cors;
io?: boolean;
} & T;
export interface ServerType {
path?: string;
server?: any;
handle: ServerOpts['handle'];
setHandle(handle?: any): void;
listeners: Listener[];
listen(port: number, hostname?: string, backlog?: number, listeningListener?: () => void): void;
listen(port: number, hostname?: string, listeningListener?: () => void): void;
listen(port: number, backlog?: number, listeningListener?: () => void): void;
listen(port: number, listeningListener?: () => void): void;
listen(path: string, backlog?: number, listeningListener?: () => void): void;
listen(path: string, listeningListener?: () => void): void;
listen(handle: any, backlog?: number, listeningListener?: () => void): void;
listen(handle: any, listeningListener?: () => void): void;
/**
* 兜底监听，当除开 `/api/router` 之外的请求，框架只监听一个api，所以有其他的请求都执行其他的监听
* @description 主要是为了兼容其他的监听
* @param listener
*/
on(listener: OnListener): void;
onWebSocket({ ws, message, pathname, token, id }: { ws: WS; message: string | Buffer; pathname: string, token?: string, id?: string }): void;
}
type WS = {
send: (data: any) => void;
close: () => void;
}
export type CommonReq = {
url: string;
method: string;
headers: Record<string, string>;
[key: string]: any;
}
export type CommonRes = {
statusCode: number;
writableEnded: boolean;
writeHead: (statusCode: number, headers?: Record<string, string>) => void;
setHeader: (name: string, value: string | string[]) => void;
cookie: (name: string, value: string, options?: any) => void;
end: (data?: any) => void;
[key: string]: any;
}

202
src/server/server.ts
View File

@@ -1,64 +1,22 @@
import type { IncomingMessage, ServerResponse } from 'node:http';
import http from 'node:http';
import https from 'node:https';
import http2 from 'node:http2';
import { handleServer } from './handle-server.ts';
import * as cookie from './cookie.ts';
export type Listener = (...args: any[]) => void;
import { isBun } from '../utils/is-engine.ts';
import { ServerType, Listener, ServerOpts } from './server-type.ts';
import { ServerBase } from './server-base.ts';
import { WsServer } from './ws-server.ts';
type CookieFn = (name: string, value: string, options?: cookie.SerializeOptions, end?: boolean) => void;
export type HandleCtx = {
req: IncomingMessage & { cookies: Record<string, string> };
res: ServerResponse & {
/**
* cookie 函数， end 参数用于设置是否立即设置到响应头，设置了后面的cookie再设置会覆盖前面的
*/
cookie: CookieFn; //
};
};
// 实现函数
export function createHandleCtx(req: IncomingMessage, res: ServerResponse): HandleCtx {
// 用于存储所有的 Set-Cookie 字符串
const cookies: string[] = [];
let handReq = req as HandleCtx['req'];
let handRes = res as HandleCtx['res'];
// 扩展 res.cookie 方法
const cookieFn: CookieFn = (name, value, options = {}, end = true) => {
// 序列化新的 Cookie
const serializedCookie = cookie.serialize(name, value, options);
cookies.push(serializedCookie); // 将新的 Cookie 添加到数组
if (end) {
// 如果设置了 end 参数，则立即设置到响应头
res.setHeader('Set-Cookie', cookies);
}
};
// 解析请求中的现有 Cookie
const parsedCookies = cookie.parse(req.headers.cookie || '');
handReq.cookies = parsedCookies;
handRes.cookie = cookieFn;
// 返回扩展的上下文
return {
req: handReq,
res: handRes,
};
}
export type Cors = {
/**
* @default '*''
*/
origin?: string | undefined;
};
export type ServerOpts = {
/**path default `/api/router` */
path?: string;
/**handle Fn */
handle?: (msg?: { path: string; key?: string;[key: string]: any }, ctx?: { req: http.IncomingMessage; res: http.ServerResponse }) => any;
cors?: Cors;
export type ServerNodeOpts = ServerOpts<{
httpType?: 'http' | 'https' | 'http2';
httpsKey?: string;
httpsCert?: string;
};
}>;
export const resultError = (error: string, code = 500) => {
const r = {
code: code,
@@ -67,41 +25,39 @@ export const resultError = (error: string, code = 500) => {
return JSON.stringify(r);
};
export class Server {
path = '/api/router';
private _server: http.Server | https.Server | http2.Http2SecureServer;
public handle: ServerOpts['handle'];
private _callback: any;
private cors: Cors;
private hasOn = false;
export class ServerNode extends ServerBase implements ServerType {
declare _server: http.Server | https.Server | http2.Http2SecureServer;
declare _callback: any;
declare cors: Cors;
private httpType = 'http';
declare listeners: Listener[];
private options = {
key: '',
cert: '',
};
constructor(opts?: ServerOpts) {
this.path = opts?.path || '/api/router';
this.handle = opts?.handle;
this.cors = opts?.cors;
io: WsServer | undefined;
constructor(opts?: ServerNodeOpts) {
super(opts);
this.httpType = opts?.httpType || 'http';
this.options = {
key: opts?.httpsKey || '',
cert: opts?.httpsCert || '',
};
const io = opts?.io ?? false;
if (io) {
this.io = new WsServer(this);
}
}
listen(port: number, hostname?: string, backlog?: number, listeningListener?: () => void): void;
listen(port: number, hostname?: string, listeningListener?: () => void): void;
listen(port: number, backlog?: number, listeningListener?: () => void): void;
listen(port: number, listeningListener?: () => void): void;
listen(path: string, backlog?: number, listeningListener?: () => void): void;
listen(path: string, listeningListener?: () => void): void;
listen(handle: any, backlog?: number, listeningListener?: () => void): void;
listen(handle: any, listeningListener?: () => void): void;
listen(...args: any[]) {
customListen(...args: any[]): void {
if (isBun) {
throw new Error('Use BunServer from server-bun module for Bun runtime');
}
this._server = this.createServer();
const callback = this.createCallback();
this._server.on('request', callback);
this._server.listen(...args);
this.io?.listen();
}
createServer() {
let server: http.Server | https.Server | http2.Http2SecureServer;
@@ -132,112 +88,4 @@ export class Server {
server = http.createServer();
return server;
}
setHandle(handle?: any) {
this.handle = handle;
}
/**
* get callback
* @returns
*/
createCallback() {
const path = this.path;
const handle = this.handle;
const cors = this.cors;
const _callback = async (req: IncomingMessage, res: ServerResponse) => {
// only handle /api/router
if (req.url === '/favicon.ico') {
return;
}
if (res.headersSent) {
// 程序已经在其他地方响应了
return;
}
if (this.hasOn && !req.url.startsWith(path)) {
// 其他监听存在，不判断不是当前路径的请求，
// 也就是不处理!url.startsWith(path)这个请求了
// 交给其他监听处理
return;
}
if (cors) {
res.setHeader('Access-Control-Allow-Origin', cors?.origin || '*'); // 允许所有域名的请求访问，可以根据需要设置具体的域名
res.setHeader('Access-Control-Allow-Headers', 'Origin, X-Requested-With, Content-Type, Accept, Authorization');
res.setHeader('Access-Control-Allow-Methods', 'GET, POST');
if (req.method === 'OPTIONS') {
res.end();
return;
}
}
const url = req.url;
if (!url.startsWith(path)) {
res.end(resultError(`not path:[${path}]`));
return;
}
const messages = await handleServer(req, res);
if (!handle) {
res.end(resultError('no handle'));
return;
}
try {
const end = await handle(messages as any, { req, res });
if (res.writableEnded) {
// 如果响应已经结束，则不进行任何操作
return;
}
res.setHeader('Content-Type', 'application/json; charset=utf-8');
if (typeof end === 'string') {
res.end(end);
} else {
res.end(JSON.stringify(end));
}
} catch (e) {
console.error(e);
res.setHeader('Content-Type', 'application/json; charset=utf-8');
if (e.code && typeof e.code === 'number') {
res.end(resultError(e.message || `Router Server error`, e.code));
} else {
res.end(resultError('Router Server error'));
}
}
};
this._callback = _callback;
return _callback;
}
get handleServer() {
return this._callback;
}
set handleServer(fn: any) {
this._callback = fn;
}
/**
* 兜底监听，当除开 `/api/router` 之外的请求，框架只监听一个api，所以有其他的请求都执行其他的监听
* @description 主要是为了兼容其他的监听
* @param listener
*/
on(listener: Listener | Listener[]) {
this._server = this._server || this.createServer();
this._server.removeAllListeners('request');
this.hasOn = true;
if (Array.isArray(listener)) {
listener.forEach((l) => this._server.on('request', l));
} else {
this._server.on('request', listener);
}
const callbackListener = this._callback || this.createCallback();
this._server.on('request', callbackListener);
return () => {
if (Array.isArray(listener)) {
listener.forEach((l) => this._server.removeListener('request', l as Listener));
} else {
this._server.removeListener('request', listener as Listener);
}
this.hasOn = false;
this._server.removeListener('request', callbackListener);
}
}
get callback() {
return this._callback || this.createCallback();
}
get server() {
return this._server;
}
}
}

138
src/server/ws-server.ts
View File

@@ -1,39 +1,38 @@
// @ts-type=ws
import { WebSocketServer } from 'ws';
import type { WebSocket } from 'ws';
import { Server } from './server.ts';
import { ServerType } from './server-type.ts'
import { parseIfJson } from '../utils/parse.ts';
import { isBun } from '../utils/is-engine.ts';
export const createWsServer = (server: Server) => {
export const createWsServer = (server: ServerType) => {
// 将 WebSocket 服务器附加到 HTTP 服务器
const wss = new WebSocketServer({ server: server.server as any });
return wss;
};
type WsServerBaseOpts = {
wss?: WebSocketServer;
wss?: WebSocketServer | null;
path?: string;
};
export type ListenerFn = (message: { data: Record<string, any>; ws: WebSocket; end: (data: any) => any }) => Promise<any>;
export type Listener<T = 'router' | 'chat' | 'ai'> = {
type: T;
path?: string;
listener: ListenerFn;
};
export class WsServerBase {
wss: WebSocketServer;
path: string;
listeners: { type: string; listener: ListenerFn }[] = [];
wss: WebSocketServer | null;
listeners: Listener[] = [];
listening: boolean = false;
server: ServerType;
constructor(opts: WsServerBaseOpts) {
this.wss = opts.wss;
if (!this.wss) {
if (!this.wss && !isBun) {
throw new Error('wss is required');
}
this.path = opts.path || '';
}
setPath(path: string) {
this.path = path;
}
listen() {
if (this.listening) {
@@ -42,116 +41,49 @@ export class WsServerBase {
}
this.listening = true;
this.wss.on('connection', (ws) => {
ws.on('message', async (message: string | Buffer) => {
const data = parseIfJson(message);
if (typeof data === 'string') {
const cleanMessage = data.trim().replace(/^["']|["']$/g, '');
ws.emit('string', cleanMessage);
return;
}
const { type, data: typeData, ...rest } = data;
if (!type) {
ws.send(JSON.stringify({ code: 500, message: 'type is required' }));
}
const listeners = this.listeners.find((item) => item.type === type);
const res = {
type,
data: {} as any,
...rest,
};
const end = (data: any, all?: Record<string, any>) => {
const result = {
...res,
data,
...all,
};
ws.send(JSON.stringify(result));
};
if (!this.wss) {
// Bun 环境下，wss 可能为 null
return;
}
if (!listeners) {
const data = { code: 500, message: `${type} server is error` };
end(data);
return;
}
listeners.listener({
data: typeData,
ws,
end: end,
});
});
ws.on('string', (message: string) => {
if (message === 'close') {
ws.close();
}
if (message == 'ping') {
ws.send('pong');
}
this.wss.on('connection', (ws, req) => {
const url = new URL(req.url, 'http://localhost');
const pathname = url.pathname;
const token = url.searchParams.get('token') || '';
const id = url.searchParams.get('id') || '';
ws.on('message', async (message: string | Buffer) => {
await this.server.onWebSocket({ ws, message, pathname, token, id });
});
ws.send('connected');
});
}
addListener(type: string, listener: ListenerFn) {
if (!type || !listener) {
throw new Error('type and listener is required');
}
const find = this.listeners.find((item) => item.type === type);
if (find) {
this.listeners = this.listeners.filter((item) => item.type !== type);
}
this.listeners.push({ type, listener });
}
removeListener(type: string) {
this.listeners = this.listeners.filter((item) => item.type !== type);
}
}
// TODO: ws handle and path and routerContext
export class WsServer extends WsServerBase {
server: Server;
constructor(server: Server, opts?: any) {
const wss = new WebSocketServer({ noServer: true });
const path = server.path;
constructor(server: ServerType) {
const wss = isBun ? null : new WebSocketServer({ noServer: true });
super({ wss });
this.server = server;
this.setPath(opts?.path || path);
this.initListener();
}
initListener() {
const server = this.server;
const listener: Listener = {
type: 'router',
listener: async ({ data, ws, end }) => {
if (!server) {
end({ code: 500, message: 'server handle is error' });
return;
}
const handle = this.server.handle;
try {
const result = await handle(data as any);
end(result);
} catch (e) {
if (e.code && typeof e.code === 'number') {
end({
code: e.code,
message: e.message,
});
} else {
end({ code: 500, message: 'Router Server error' });
}
}
},
};
this.addListener(listener.type, listener.listener);
}
listen() {
if (isBun) {
// Bun 的 WebSocket 在 Bun.serve 中处理，这里不需要额外操作
// WebSocket 升级会在 listenWithBun 中处理
this.listening = true;
return;
}
super.listen();
const server = this.server;
const wss = this.wss;
// HTTP 服务器的 upgrade 事件
// @ts-ignore
server.server.on('upgrade', (req, socket, head) => {
if (req.url === this.path) {
const url = new URL(req.url, 'http://localhost');
const listenPath = this.server.listeners.map((item) => item.path).filter((item) => item);
if (listenPath.includes(url.pathname) || url.pathname === this.server.path) {
wss.handleUpgrade(req, socket, head, (ws) => {
// 这里手动触发 connection 事件
// 这里手动触发 connection事件
// @ts-ignore
wss.emit('connection', ws, req);
});

4
src/test/listen-ip.ts
View File

@@ -1,7 +1,7 @@
// import { Server } from 'node:http';
import { Server } from '../server/server.ts'
import { ServerNode } from '../server/server.ts'
const server = new Server({
const server = new ServerNode({
path: '/',
handle: async (data, ctx) => {
console.log('ctx', ctx.req.url)

6
src/utils/is-engine.ts
View File

@@ -2,9 +2,13 @@ export const isNode = typeof process !== 'undefined' && process.versions != null
export const isBrowser = typeof window !== 'undefined' && typeof document !== 'undefined' && typeof document.createElement === 'function';
// @ts-ignore
export const isDeno = typeof Deno !== 'undefined' && typeof Deno.version === 'object' && typeof Deno.version.deno === 'string';
// @ts-ignore
export const isBun = typeof Bun !== 'undefined' && typeof Bun.version === 'string';
export const getEngine = () => {
if (isNode) {
if (isBun) {
return 'bun';
} else if (isNode) {
return 'node';
} else if (isBrowser) {
return 'browser';