test bun stream sse and http-stream

demo/bun/src/index.ts

@@ -0,0 +1,38 @@
+const server = Bun.serve({
+  port: 5002,
+  fetch(request: Bun.BunRequest, server) {
+    const url = new URL(request.url);
+
+    if (url.pathname === '/stream') {
+      // 直接使用 Bun 的原生 ReadableStream
+      const readable = new ReadableStream({
+        async start(controller) {
+          for (let i = 1; i <= 10; i++) {
+            // 检查客户端是否断开
+            if (request.signal.aborted) {
+              console.log('客户端已断开');
+              controller.close();
+              return;
+            }
+
+            controller.enqueue(`${new Date().toISOString()} 第 ${i} 批数据\n`);
+            await new Promise(r => setTimeout(r, 100)); // 模拟延迟
+          }
+          controller.close();
+        }
+      });
+      request.signal.addEventListener('abort', () => {
+        console.log('Request aborted by client');
+      });
+      return new Response(readable, {
+        status: 200,
+        headers: {
+          'Content-Type': 'text/plain',
+          'Transfer-Encoding': 'chunked'
+        }
+      });
+    }
+
+    return new Response('Not Found', { status: 404 });
+  }
+});

package.json

@@ -1,7 +1,7 @@
 {
   "$schema": "https://json.schemastore.org/package",
   "name": "@kevisual/router",
-  "version": "0.0.65",
+  "version": "0.0.66",
   "description": "",
   "type": "module",
   "main": "./dist/router.js",

src/server/server-bun.ts

@@ -50,7 +50,7 @@ export class BunServer extends ServerBase implements ServerType {
       port,
       hostname,
       idleTimeout: 0, // 4 minutes idle timeout (max 255 seconds)
-      fetch: async (request: Bun.BunRequest, server: any) => {
+      fetch: async (request: Bun.BunRequest, server: Bun.Server<{}>) => {
         const host = request.headers.get('host') || 'localhost';
         const clientInfo = server.requestIP(request); // 返回 { address: string, port: number } 或 null
         const url = new URL(request.url, `http://${host}`);
@@ -72,6 +72,7 @@ export class BunServer extends ServerBase implements ServerType {
 
         // 将 Bun 的 Request 转换为 Node.js 风格的 req/res
         return new Promise(async (resolve) => {
+          const reqListener: { event: string; listener: Function }[] = [];
           const req: RouterReq = {
             url: url.pathname + url.search,
             method: request.method,
@@ -81,12 +82,29 @@ export class BunServer extends ServerBase implements ServerType {
               remoteAddress: request?.remoteAddress || request?.ip || clientInfo?.address || '',
               remotePort: clientInfo?.port || 0,
             },
-            // @ts-ignore
+            on: (event: string, listener: Function) => {
+              reqListener.push({ event, listener });
+            },
             bun: {
               request, // 原始请求对象
               server,  // 原始服务器对象
+              resolve
             }
           };
+          const onClose = () => {
+            reqListener.forEach(item => {
+              if (item.event === 'close') {
+                item.listener();
+              }
+            });
+            reqListener.length = 0;
+          }
+          // 监听请求的取消事件
+          if (request.signal) {
+            request.signal.addEventListener('abort', () => {
+              onClose();
+            });
+          }
 
           const res: RouterRes = {
             statusCode: 200,
@@ -143,7 +161,7 @@ export class BunServer extends ServerBase implements ServerType {
               if (callback) callback();
               return true;
             },
-            pipe(stream: any) {
+            pipe(stream: ReadableStream | NodeJS.ReadableStream) {
               this.writableEnded = true;
 
               // 如果是 ReadableStream，直接使用
@@ -164,6 +182,7 @@ export class BunServer extends ServerBase implements ServerType {
                     controller.enqueue(chunk);
                   });
                   stream.on('end', () => {
+                    onClose();
                     controller.close();
                   });
                   stream.on('error', (err: any) => {
@@ -171,9 +190,9 @@ export class BunServer extends ServerBase implements ServerType {
                   });
                 },
                 cancel() {
-                  if (stream.destroy) {
+                  // 只有NODE流才有destroy方法
-                    stream.destroy();
+                  // @ts-ignore
-                  }
+                  stream?.destroy?.();
                 }
               });
 

src/server/server-type.ts

@@ -104,6 +104,12 @@ export type RouterReq<T = {}> = {
   };
   body?: string;
   cookies?: Record<string, string>;
+  bun?: {
+    request: Bun.BunRequest;
+    server: Bun.Server<{}>;
+    resolve: (response: Response) => void;
+  }
+  on: (event: 'close', listener: Function) => void;
 } & T;
 
 export type RouterRes<T = {}> = {
@@ -116,6 +122,6 @@ export type RouterRes<T = {}> = {
   setHeader: (name: string, value: string | string[]) => void;
   cookie: (name: string, value: string, options?: any) => void;
   write: (chunk: any) => void;
-  pipe: (stream: any) => void;
+  pipe: (stream: ReadableStream) => void;
   end: (data?: any) => void;
 } & T;