Feat readable body (#486)

* make body a readableStream instead of a string

* force non empty when necessry

* fix error lambda streaming

* fix some issue

* changeset

Author: conico974

.changeset/lucky-dots-obey.md

@@ -0,0 +1,5 @@
+---
+"open-next": minor
+---
+
+Replace InternalResult body from string to ReadableStream

packages/open-next/src/adapters/edge-adapter.ts

@@ -1,4 +1,7 @@
-import { InternalEvent, InternalResult } from "types/open-next";
+import type { ReadableStream } from "node:stream/web";
+
+import type { InternalEvent, InternalResult } from "types/open-next";
+import { emptyReadableStream } from "utils/stream";
 
 // We import it like that so that the edge plugin can replace it
 import { NextConfig } from "../adapters/config";
@@ -8,11 +11,14 @@ import {
   convertToQueryString,
 } from "../core/routing/util";
 
+declare global {
+  var isEdgeRuntime: true;
+}
+
 const defaultHandler = async (
   internalEvent: InternalEvent,
 ): Promise<InternalResult> => {
-  // TODO: We need to handle splitted function here
-  // We should probably create an host resolver to redirect correctly
+  globalThis.isEdgeRuntime = true;
 
   const host = internalEvent.headers.host
     ? `https://${internalEvent.headers.host}`
@@ -35,10 +41,6 @@ const defaultHandler = async (
     url,
     body: convertBodyToReadableStream(internalEvent.method, internalEvent.body),
   });
-
-  const arrayBuffer = await response.arrayBuffer();
-  const buffer = Buffer.from(arrayBuffer);
-
   const responseHeaders: Record<string, string | string[]> = {};
   response.headers.forEach((value, key) => {
     if (key.toLowerCase() === "set-cookie") {
@@ -49,9 +51,9 @@ const defaultHandler = async (
       responseHeaders[key] = value;
     }
   });
-  // console.log("responseHeaders", responseHeaders);
-  const body = buffer.toString();
-  // console.log("body", body);
+
+  const body =
+    (response.body as ReadableStream<Uint8Array>) ?? emptyReadableStream();
 
   return {
     type: "core",

packages/open-next/src/adapters/image-optimization-adapter.ts

@@ -19,6 +19,7 @@ import {
 // @ts-ignore
 import type { NextUrlWithParsedQuery } from "next/dist/server/request-meta";
 import { InternalEvent, InternalResult } from "types/open-next.js";
+import { emptyReadableStream, toReadableStream } from "utils/stream.js";
 
 import { createGenericHandler } from "../core/createGenericHandler.js";
 import { resolveImageLoader } from "../core/resolve.js";
@@ -82,7 +83,7 @@ export async function defaultHandler(
       return {
         statusCode: 304,
         headers: {},
-        body: "",
+        body: emptyReadableStream(),
         isBase64Encoded: false,
         type: "core",
       };
@@ -169,7 +170,7 @@ function buildSuccessResponse(
   return {
     type: "core",
     statusCode: 200,
-    body: result.buffer.toString("base64"),
+    body: toReadableStream(result.buffer, true),
     isBase64Encoded: true,
     headers,
   };
@@ -191,7 +192,7 @@ function buildFailureResponse(
       "Cache-Control": `public,max-age=60,immutable`,
       "Content-Type": "application/json",
     });
-    response.end(e?.message || e?.toString() || e);
+    response.end(e?.message || e?.toString() || "An error occurred");
   }
   return {
     type: "core",
@@ -203,7 +204,7 @@ function buildFailureResponse(
       "Cache-Control": `public,max-age=60,immutable`,
       "Content-Type": "application/json",
     },
-    body: e?.message || e?.toString() || e,
+    body: toReadableStream(e?.message || e?.toString() || "An error occurred"),
   };
 }
 

packages/open-next/src/converters/aws-apigw-v1.ts

@@ -1,5 +1,6 @@
 import { APIGatewayProxyEvent, APIGatewayProxyResult } from "aws-lambda";
 import type { Converter, InternalEvent, InternalResult } from "types/open-next";
+import { fromReadableStream } from "utils/stream";
 
 import { debug } from "../adapters/logger";
 import { removeUndefinedFromQuery } from "./utils";
@@ -74,9 +75,9 @@ async function convertFromAPIGatewayProxyEvent(
   };
 }
 
-function convertToApiGatewayProxyResult(
+async function convertToApiGatewayProxyResult(
   result: InternalResult,
-): APIGatewayProxyResult {
+): Promise<APIGatewayProxyResult> {
   const headers: Record<string, string> = {};
   const multiValueHeaders: Record<string, string[]> = {};
   Object.entries(result.headers).forEach(([key, value]) => {
@@ -91,10 +92,12 @@ function convertToApiGatewayProxyResult(
     }
   });
 
+  const body = await fromReadableStream(result.body, result.isBase64Encoded);
+
   const response: APIGatewayProxyResult = {
     statusCode: result.statusCode,
     headers,
-    body: result.body,
+    body,
     isBase64Encoded: result.isBase64Encoded,
     multiValueHeaders,
   };

packages/open-next/src/converters/aws-apigw-v2.ts

@@ -1,6 +1,7 @@
 import { APIGatewayProxyEventV2, APIGatewayProxyResultV2 } from "aws-lambda";
 import { parseCookies } from "http/util";
 import type { Converter, InternalEvent, InternalResult } from "types/open-next";
+import { fromReadableStream } from "utils/stream";
 
 import { debug } from "../adapters/logger";
 import { convertToQuery } from "../core/routing/util";
@@ -85,9 +86,9 @@ async function convertFromAPIGatewayProxyEventV2(
   };
 }
 
-function convertToApiGatewayProxyResultV2(
+async function convertToApiGatewayProxyResultV2(
   result: InternalResult,
-): APIGatewayProxyResultV2 {
+): Promise<APIGatewayProxyResultV2> {
   const headers: Record<string, string> = {};
   Object.entries(result.headers)
     .filter(
@@ -104,11 +105,13 @@ function convertToApiGatewayProxyResultV2(
       headers[key] = Array.isArray(value) ? value.join(", ") : `${value}`;
     });
 
+  const body = await fromReadableStream(result.body, result.isBase64Encoded);
+
   const response: APIGatewayProxyResultV2 = {
     statusCode: result.statusCode,
     headers,
     cookies: parseCookies(result.headers["set-cookie"]),
-    body: result.body,
+    body,
     isBase64Encoded: result.isBase64Encoded,
   };
   debug(response);

packages/open-next/src/converters/aws-cloudfront.ts

@@ -8,6 +8,7 @@ import {
 import { OutgoingHttpHeader } from "http";
 import { parseCookies } from "http/util";
 import type { Converter, InternalEvent, InternalResult } from "types/open-next";
+import { fromReadableStream } from "utils/stream";
 
 import { debug } from "../adapters/logger";
 import {
@@ -159,14 +160,18 @@ async function convertToCloudFrontRequestResult(
       const serverResponse = createServerResponse(result.internalEvent, {});
       await proxyRequest(result.internalEvent, serverResponse);
       const externalResult = convertRes(serverResponse);
+      const body = await fromReadableStream(
+        externalResult.body,
+        externalResult.isBase64Encoded,
+      );
       const cloudfrontResult = {
         status: externalResult.statusCode.toString(),
         statusDescription: "OK",
         headers: convertToCloudfrontHeaders(externalResult.headers, true),
         bodyEncoding: externalResult.isBase64Encoded
           ? ("base64" as const)
           : ("text" as const),
-        body: externalResult.body,
+        body,
       };
       debug("externalResult", cloudfrontResult);
       return cloudfrontResult;
@@ -208,12 +213,14 @@ async function convertToCloudFrontRequestResult(
     return response;
   }
 
+  const body = await fromReadableStream(result.body, result.isBase64Encoded);
+
   const response: CloudFrontRequestResult = {
     status: result.statusCode.toString(),
     statusDescription: "OK",
     headers: convertToCloudfrontHeaders(responseHeaders, true),
     bodyEncoding: result.isBase64Encoded ? "base64" : "text",
-    body: result.body,
+    body,
   };
   debug(response);
   return response;

packages/open-next/src/converters/edge.ts

@@ -90,7 +90,7 @@ const converter: Converter<
       for (const [key, value] of Object.entries(result.headers)) {
         headers.set(key, Array.isArray(value) ? value.join(",") : value);
       }
-      return new Response(result.body, {
+      return new Response(result.body as ReadableStream, {
         status: result.statusCode,
         headers: headers,
       });

packages/open-next/src/converters/node.ts

@@ -44,7 +44,7 @@ const converter: Converter = {
     };
   },
   // Nothing to do here, it's streaming
-  convertTo: (internalResult: InternalResult) => ({
+  convertTo: async (internalResult: InternalResult) => ({
     body: internalResult.body,
     headers: internalResult.headers,
     statusCode: internalResult.statusCode,

packages/open-next/src/core/requestHandler.ts

@@ -7,6 +7,7 @@ import {
 } from "http/index.js";
 import { InternalEvent, InternalResult } from "types/open-next";
 import { DetachedPromiseRunner } from "utils/promise";
+import { fromReadableStream } from "utils/stream";
 
 import { debug, error, warn } from "../adapters/logger";
 import { convertRes, createServerResponse, proxyRequest } from "./routing/util";
@@ -63,12 +64,22 @@ export async function openNextHandler(
   }, {});
 
   if ("type" in preprocessResult) {
-    // res is used only in the streaming case
-    const res = createServerResponse(internalEvent, headers, responseStreaming);
-    res.statusCode = preprocessResult.statusCode;
-    res.flushHeaders();
-    res.write(preprocessResult.body);
-    res.end();
+    // // res is used only in the streaming case
+    if (responseStreaming) {
+      const res = createServerResponse(
+        internalEvent,
+        headers,
+        responseStreaming,
+      );
+      res.statusCode = preprocessResult.statusCode;
+      res.flushHeaders();
+      const body = await fromReadableStream(
+        preprocessResult.body,
+        preprocessResult.isBase64Encoded,
+      );
+      res.write(body);
+      res.end();
+    }
     return preprocessResult;
   } else {
     const preprocessedEvent = preprocessResult.internalEvent;

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

@@ -14,6 +14,7 @@ import type {
   RouteHas,
 } from "types/next-types";
 import { InternalEvent, InternalResult } from "types/open-next";
+import { emptyReadableStream, toReadableStream } from "utils/stream";
 
 import { debug } from "../../adapters/logger";
 import { localizePath } from "./i18n";
@@ -243,8 +244,11 @@ export function handleRewrites<T extends RewriteDefinition>(
   };
 }
 
-function handleTrailingSlashRedirect(event: InternalEvent) {
+function handleTrailingSlashRedirect(
+  event: InternalEvent,
+): false | InternalResult {
   const url = new URL(event.url, "http://localhost");
+  const emptyBody = emptyReadableStream();
 
   if (
     // Someone is trying to redirect to a different origin, let's not do that
@@ -270,7 +274,7 @@ function handleTrailingSlashRedirect(event: InternalEvent) {
           headersLocation[1] ? `?${headersLocation[1]}` : ""
         }`,
       },
-      body: "",
+      body: emptyBody,
       isBase64Encoded: false,
     };
     // eslint-disable-next-line sonarjs/elseif-without-else
@@ -288,7 +292,7 @@ function handleTrailingSlashRedirect(event: InternalEvent) {
           headersLocation[1] ? `?${headersLocation[1]}` : ""
         }`,
       },
-      body: "",
+      body: emptyBody,
       isBase64Encoded: false,
     };
   } else return false;
@@ -311,7 +315,7 @@ export function handleRedirects(
       headers: {
         Location: internalEvent.url,
       },
-      body: "",
+      body: emptyReadableStream(),
       isBase64Encoded: false,
     };
   }
@@ -328,7 +332,7 @@ export function fixDataPage(
     return {
       type: internalEvent.type,
       statusCode: 404,
-      body: "{}",
+      body: toReadableStream("{}"),
       headers: {
         "Content-Type": "application/json",
       },