improve performance of streams

Author: anonrig

.changeset/shaggy-walls-sniff.md

@@ -0,0 +1,5 @@
+---
+"@opennextjs/aws": patch
+---
+
+Reduces allocations and copies of streams

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

@@ -105,14 +105,15 @@ export function convertRes(res: OpenNextNodeResponse): InternalResult {
   const isBase64Encoded =
     isBinaryContentType(headers["content-type"]) ||
     !!headers["content-encoding"];
+  let index = 0;
   const body = new ReadableStream({
     pull(controller) {
-      if (!res._chunks || res._chunks.length === 0) {
+      if (!res._chunks || index >= res._chunks.length) {
         controller.close();
         return;
       }
 
-      controller.enqueue(res._chunks.shift());
+      controller.enqueue(res._chunks[index++]);
     },
   });
   return {
@@ -217,13 +218,12 @@ export function convertBodyToReadableStream(
 ) {
   if (method === "GET" || method === "HEAD") return undefined;
   if (!body) return undefined;
-  const readable = new ReadableStream({
+  return new ReadableStream({
     start(controller) {
       controller.enqueue(body);
       controller.close();
     },
   });
-  return readable;
 }
 
 enum CommonHeaders {

packages/open-next/src/utils/binary.ts

@@ -60,8 +60,9 @@ const commonBinaryMimeTypes = new Set([
 ]);
 
 export function isBinaryContentType(contentType?: string | null) {
-  if (!contentType) return false;
+  if (contentType == null) return false;
 
-  const value = contentType?.split(";")[0] ?? "";
+  const value = contentType.split(";").at(0);
+  if (value == null) return false;
   return commonBinaryMimeTypes.has(value);
 }

packages/open-next/src/utils/fetch.ts

@@ -18,7 +18,6 @@ export function customFetchClient(client: AwsClient) {
      * This is necessary otherwise we get some error : SocketError: other side closed
      * https://github.com/nodejs/undici/issues/583#issuecomment-855384858
      */
-    const clonedResponse = response.clone();
-    return clonedResponse;
+    return response.clone();
   };
 }

packages/open-next/src/utils/stream.ts

@@ -1,43 +1,72 @@
-import { Readable } from "node:stream";
-import type { ReadableStream } from "node:stream/web";
+import { ReadableStream } from "node:stream/web";
 
-export function fromReadableStream(
+export async function fromReadableStream(
   stream: ReadableStream<Uint8Array>,
   base64?: boolean,
 ): Promise<string> {
   const reader = stream.getReader();
   const chunks: Uint8Array[] = [];
+  let totalLength = 0;
 
-  return new Promise((resolve, reject) => {
-    function pump() {
-      reader
-        .read()
-        .then(({ done, value }) => {
-          if (done) {
-            resolve(Buffer.concat(chunks).toString(base64 ? "base64" : "utf8"));
-            return;
-          }
-          chunks.push(value);
-          pump();
-        })
-        .catch(reject);
+  try {
+    while (true) {
+      const { done, value } = await reader.read();
+      if (done) break;
+      chunks.push(value);
+      totalLength += value.length;
     }
-    pump();
-  });
+
+    if (chunks.length === 0) {
+      return "";
+    }
+
+    if (chunks.length === 1) {
+      return Buffer.from(chunks[0]).toString(base64 ? "base64" : "utf8");
+    }
+
+    // Pre-allocate buffer with exact size to avoid reallocation
+    const buffer = Buffer.allocUnsafe(totalLength);
+    let offset = 0;
+    for (const chunk of chunks) {
+      buffer.set(chunk, offset);
+      offset += chunk.length;
+    }
+
+    return buffer.toString(base64 ? "base64" : "utf8");
+  } finally {
+    reader.releaseLock();
+  }
 }
 
 export function toReadableStream(
   value: string,
   isBase64?: boolean,
 ): ReadableStream {
-  return Readable.toWeb(
-    Readable.from(Buffer.from(value, isBase64 ? "base64" : "utf8")),
-  );
+  const buffer = Buffer.from(value, isBase64 ? "base64" : "utf8");
+
+  return new ReadableStream({
+    start(controller) {
+      controller.enqueue(buffer);
+      controller.close();
+    },
+  });
 }
 
+let maybeSomethingBuffer: Buffer | undefined;
+
 export function emptyReadableStream(): ReadableStream {
   if (process.env.OPEN_NEXT_FORCE_NON_EMPTY_RESPONSE === "true") {
-    return Readable.toWeb(Readable.from([Buffer.from("SOMETHING")]));
+    return new ReadableStream({
+      start(controller) {
+        maybeSomethingBuffer ??= Buffer.from("SOMETHING");
+        controller.enqueue(maybeSomethingBuffer);
+        controller.close();
+      },
+    });
   }
-  return Readable.toWeb(Readable.from([]));
+  return new ReadableStream({
+    start(controller) {
+      controller.close();
+    },
+  });
 }