feat: 增强流管道功能，添加 Bun 环境支持，重构代理请求和响应处理逻辑

assistant/src/module/assistant/proxy/http-proxy.ts

@@ -2,7 +2,7 @@ import http from 'node:http';
 import https from 'node:https';
 import { rewriteCookieDomain } from '../https/cookie-rewrite.ts';
 import { ProxyInfo } from './proxy.ts';
-import { pipeStream } from './pipe.ts';
+import { pipeProxyReq, pipeIncoming, pipeStream } from './pipe.ts';
 export const defaultApiProxy = [
   {
     path: '/api/router',
@@ -100,7 +100,8 @@ export const httpProxy = (req: http.IncomingMessage, res: http.ServerResponse, p
     // 将代理服务器的响应头和状态码返回给客户端
     res.writeHead(proxyRes.statusCode, responseHeaders);
     // 将代理响应流写入客户端响应
-    proxyRes.pipe(res, { end: true });
+    // proxyRes.pipe(res, { end: true });
+    pipeIncoming(proxyRes, res);
   });
   // 处理代理请求的错误事件
   proxyReq.on('error', (err) => {
@@ -109,7 +110,6 @@ export const httpProxy = (req: http.IncomingMessage, res: http.ServerResponse, p
     res.write(`Proxy request error: ${err.message}`);
   });
   // 处理 POST 请求的请求体（传递数据到目标服务器）,end:true 表示当请求体结束时，关闭请求
-  req.pipe(proxyReq, { end: true });
+  // req.pipe(proxyReq, { end: true });
-  // @ts-ignore
+  pipeProxyReq(req, proxyReq);
-  // pipeStream(proxyReq, res);
 }

assistant/src/module/assistant/proxy/pipe.ts

@@ -2,18 +2,118 @@ import * as http from 'http';
 import * as fs from 'fs';
 import { isBun } from './utils.ts';
 
+/**
+ * 文件流管道传输函数
+ * 将指定文件的内容通过流的方式传输给客户端响应
+ * @param filePath 要传输的文件路径
+ * @param res HTTP服务器响应对象
+ */
 export const pipeFileStream = (filePath: string, res: http.ServerResponse) => {
   const readStream = fs.createReadStream(filePath);
   if (isBun) {
+    // Bun环境下的流处理方式
     res.pipe(readStream as any);
   } else {
+    // Node.js标准环境下的流处理方式，end:true表示在流结束时自动关闭响应
     readStream.pipe(res, { end: true });
   }
 }
+
+/**
+ * 通用流管道传输函数
+ * 将可读流的数据传输给客户端响应
+ * @param readStream 可读流对象
+ * @param res HTTP服务器响应对象
+ */
 export const pipeStream = (readStream: fs.ReadStream, res: http.ServerResponse) => {
   if (isBun) {
+    // Bun环境下的流处理方式
     res.pipe(readStream as any);
   } else {
+    // Node.js标准环境下的流处理方式
     readStream.pipe(res, { end: true });
   }
+}
+
+/**
+ * 代理响应流传输函数
+ * 将代理服务器返回的响应数据传输给客户端
+ * 处理从目标服务器收到的响应流并转发给原始客户端
+ * @param proxyRes 代理服务器的响应对象
+ * @param res HTTP服务器响应对象
+ */
+export const pipeIncoming = (proxyRes: http.IncomingMessage, res: http.ServerResponse) => {
+  if (isBun) {
+    // Bun环境下需要手动收集数据并end，因为Bun的pipe机制与Node.js不同
+    const chunks: Buffer[] = [];
+
+    // 监听数据到达事件，收集所有数据块
+    proxyRes.on('data', (chunk: Buffer) => {
+      chunks.push(chunk);
+    });
+
+    // 监听数据结束事件，将收集的数据合并并发送给客户端
+    proxyRes.on('end', () => {
+      const result = Buffer.concat(chunks).toString();
+      res.end(result);
+    });
+
+    // 监听错误事件，处理代理响应过程中的错误
+    proxyRes.on('error', (error) => {
+      res.writeHead(500);
+      res.end(JSON.stringify({ error: error.message }));
+    });
+
+  } else {
+    // Node.js标准环境下直接使用pipe进行流传输
+    proxyRes.pipe(res, { end: true });
+  }
+}
+
+/**
+ * 代理请求流传输函数
+ * 将客户端的请求数据传输给代理服务器
+ * 处理来自客户端的请求流并转发给目标服务器
+ * @param req 客户端的请求对象
+ * @param proxyReq 代理服务器的请求对象
+ */
+export const pipeProxyReq = (req: http.IncomingMessage, proxyReq: http.ClientRequest) => {
+  if (isBun) {
+    // 检查 req 对象是否具有必要的事件监听方法，如果没有，说明可能已经被处理过
+    if (typeof req.on !== 'function') {
+      // 如果 req.on 不存在，直接结束代理请求
+      proxyReq.end();
+      return;
+    }
+
+    // Bun环境下需要手动收集请求数据并发送
+    const chunks: Buffer[] = [];
+
+    // 监听请求数据到达事件，收集所有数据块
+    req.on('data', (chunk: Buffer) => {
+      chunks.push(chunk);
+    });
+
+    // 监听请求数据结束事件，将收集的数据合并后发送给目标服务器
+    req.on('end', () => {
+      const result = Buffer.concat(chunks);
+      proxyReq.end(result);
+    });
+
+    // 监听请求错误事件，处理请求传输过程中的错误
+    req.on('error', (error) => {
+      // ClientRequest没有writeHead方法，直接销毁请求以触发错误处理
+      proxyReq.destroy(error);
+    });
+
+  } else {
+    // 非 Bun 环境下使用标准的pipe方法进行流传输
+    // 同样检查 pipe 方法是否存在以确保兼容性
+    if (typeof req.pipe === 'function') {
+      req.pipe(proxyReq, { end: true });
+    } else {
+      // 如果 pipe 方法不存在，直接结束代理请求
+      proxyReq.end();
+    }
+  }
 }