improve performance of responses (#991)

Co-authored-by: James M Snell <jsnell@cloudflare.com>
Co-authored-by: Victor Berchet <victor@suumit.com>

Author: 3 people

.changeset/twelve-nails-agree.md

@@ -0,0 +1,5 @@
+---
+"@opennextjs/aws": patch
+---
+
+Improves the performance of generating responses

packages/open-next/src/core/routing/util.ts

@@ -1,7 +1,6 @@
 import crypto from "node:crypto";
 import type { OutgoingHttpHeaders } from "node:http";
 import { parse as parseQs, stringify as stringifyQs } from "node:querystring";
-import { Readable } from "node:stream";
 
 import { BuildId, HtmlPages, NextConfig } from "config/index.js";
 import type { IncomingMessage } from "http/index.js";
@@ -18,6 +17,7 @@ import type {
   StreamCreator,
 } from "types/open-next.js";
 
+import { ReadableStream } from "node:stream/web";
 import { debug, error } from "../../adapters/logger.js";
 import { isBinaryContentType } from "../../utils/binary.js";
 import { localizePath } from "./i18n/index.js";
@@ -105,9 +105,16 @@ export function convertRes(res: OpenNextNodeResponse): InternalResult {
   const isBase64Encoded =
     isBinaryContentType(headers["content-type"]) ||
     !!headers["content-encoding"];
-  // We cannot convert the OpenNextNodeResponse to a ReadableStream directly
-  // You can look in the `aws-lambda.ts` file for some context
-  const body = Readable.toWeb(Readable.from(res.getBody()));
+  const body = new ReadableStream({
+    pull(controller) {
+      if (!res._chunks || res._chunks.length === 0) {
+        controller.close();
+        return;
+      }
+
+      controller.enqueue(res._chunks.shift());
+    },
+  });
   return {
     type: "core",
     statusCode,

packages/open-next/src/http/openNextResponse.ts

@@ -282,6 +282,14 @@ export class OpenNextNodeResponse extends Transform implements ServerResponse {
     return Buffer.concat(this._chunks);
   }
 
+  getBodyLength(): number {
+    let size = 0;
+    for (const chunk of this._chunks) {
+      size += chunk.length;
+    }
+    return size;
+  }
+
   private _internalWrite(chunk: any, encoding: BufferEncoding) {
     this._chunks.push(Buffer.from(chunk, encoding));
     this.push(chunk, encoding);
@@ -310,7 +318,7 @@ export class OpenNextNodeResponse extends Transform implements ServerResponse {
     globalThis.__openNextAls
       ?.getStore()
       ?.pendingPromiseRunner.add(this.onEnd(this.headers));
-    const bodyLength = this.getBody().length;
+    const bodyLength = this.getBodyLength();
     this.streamCreator?.onFinish?.(bodyLength);
 
     //This is only here because of aws broken streaming implementation.
@@ -366,8 +374,10 @@ export class OpenNextNodeResponse extends Transform implements ServerResponse {
   }
 
   send() {
-    const body = this.getBody();
-    this.end(body);
+    for (const chunk of this._chunks) {
+      this.write(chunk);
+    }
+    this.end();
   }
 
   body(value: string) {

packages/tests-unit/tests/adapters/cache.test.ts

@@ -515,9 +515,9 @@ describe("CacheHandler", () => {
     it("Should not throw when set cache throws", async () => {
       incrementalCache.set.mockRejectedValueOnce(new Error("Error"));
 
-      expect(
-        async () => await cache.set("key", { kind: "REDIRECT", props: {} }),
-      ).not.toThrow();
+      await expect(
+        cache.set("key", { kind: "REDIRECT", props: {} }),
+      ).resolves.not.toThrow();
     });
   });
 

packages/tests-unit/tests/core/routing/util.test.ts

@@ -51,6 +51,7 @@ function createResponse(res: Partial<Res>) {
     getFixedHeaders: () => res.headers ?? {},
     body: res.body ?? "",
     getBody: () => Buffer.from(res.body ?? ""),
+    _chunks: res.body ? [Buffer.from(res.body)] : [],
   };
 }