feat: automatically enable duplex to stream request body (#275)

unjs · Aug 23, 2023 · d58b961 · d58b961
1 parent cb05d8f
commit d58b961
Show file tree

Hide file tree

Showing 4 changed files with 88 additions and 26 deletions.
diff --git a/package.json b/package.json
@@ -86,6 +86,7 @@
     "jiti": "^1.19.3",
     "listhen": "^1.3.0",
     "prettier": "^3.0.2",
+    "std-env": "^3.4.3",
     "typescript": "^5.1.6",
     "unbuild": "2.0.0",
     "vitest": "^0.34.2"

diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
diff --git a/src/fetch.ts b/src/fetch.ts
@@ -1,3 +1,4 @@
+import type { Readable } from "node:stream";
 import destr from "destr";
 import { withBase, withQuery } from "ufo";
 import type { Fetch, RequestInfo, RequestInit, Response } from "./types";
@@ -45,6 +46,13 @@ export interface FetchOptions<R extends ResponseType = ResponseType>
   parseResponse?: (responseText: string) => any;
   responseType?: R;
 
+  /**
+   * @experimental Set to "half" to enable duplex streaming.
+   * Will be automatically set to "half" when using a ReadableStream as body.
+   * https://fetch.spec.whatwg.org/#enumdef-requestduplex
+   */
+  duplex?: "half" | undefined;
+
   /** timeout in milliseconds */
   timeout?: number;
 
@@ -182,26 +190,37 @@ export function createFetch(globalOptions: CreateFetchOptions = {}): $Fetch {
           ...context.options.query,
         });
       }
-      if (
-        context.options.body &&
-        isPayloadMethod(context.options.method) &&
-        isJSONSerializable(context.options.body)
-      ) {
-        // Automatically JSON stringify request bodies, when not already a string.
-        context.options.body =
-          typeof context.options.body === "string"
-            ? context.options.body
-            : JSON.stringify(context.options.body);
-
-        // Set Content-Type and Accept headers to application/json by default
-        // for JSON serializable request bodies.
-        // Pass empty object as older browsers don't support undefined.
-        context.options.headers = new Headers(context.options.headers || {});
-        if (!context.options.headers.has("content-type")) {
-          context.options.headers.set("content-type", "application/json");
-        }
-        if (!context.options.headers.has("accept")) {
-          context.options.headers.set("accept", "application/json");
+      if (context.options.body && isPayloadMethod(context.options.method)) {
+        if (isJSONSerializable(context.options.body)) {
+          // JSON Body
+          // Automatically JSON stringify request bodies, when not already a string.
+          context.options.body =
+            typeof context.options.body === "string"
+              ? context.options.body
+              : JSON.stringify(context.options.body);
+
+          // Set Content-Type and Accept headers to application/json by default
+          // for JSON serializable request bodies.
+          // Pass empty object as older browsers don't support undefined.
+          context.options.headers = new Headers(context.options.headers || {});
+          if (!context.options.headers.has("content-type")) {
+            context.options.headers.set("content-type", "application/json");
+          }
+          if (!context.options.headers.has("accept")) {
+            context.options.headers.set("accept", "application/json");
+          }
+        } else if (
+          // ReadableStream Body
+          ("pipeTo" in (context.options.body as ReadableStream) &&
+            typeof (context.options.body as ReadableStream).pipeTo ===
+              "function") ||
+          // Node.js Stream Body
+          typeof (context.options.body as Readable).pipe === "function"
+        ) {
+          // eslint-disable-next-line unicorn/no-lonely-if
+          if (!("duplex" in context.options)) {
+            context.options.duplex = "half";
+          }
         }
       }
     }

diff --git a/test/index.test.ts b/test/index.test.ts
@@ -1,3 +1,4 @@
+import { Readable } from "node:stream";
 import { listen } from "listhen";
 import { getQuery, joinURL } from "ufo";
 import {
@@ -10,6 +11,7 @@ import {
 } from "h3";
 import { describe, beforeAll, afterAll, it, expect } from "vitest";
 import { Headers, FormData, Blob } from "node-fetch-native";
+import { nodeMajorVersion } from "std-env";
 import { $fetch } from "../src/node";
 
 describe("ofetch", () => {
@@ -170,7 +172,7 @@ describe("ofetch", () => {
     expect(body).to.deep.eq(message);
   });
 
-  it("Handle buffer body", async () => {
+  it("Handle Buffer body", async () => {
     const message = "Hallo von Pascal";
     const { body } = await $fetch(getURL("echo"), {
       method: "POST",
@@ -180,6 +182,43 @@ describe("ofetch", () => {
     expect(body).to.deep.eq(message);
   });
 
+  it.skipIf(Number(nodeMajorVersion) < 18)(
+    "Handle ReadableStream body",
+    async () => {
+      const message = "Hallo von Pascal";
+      const { body } = await $fetch(getURL("echo"), {
+        method: "POST",
+        headers: {
+          "content-length": "16",
+        },
+        body: new ReadableStream({
+          start(controller) {
+            controller.enqueue(new TextEncoder().encode(message));
+            controller.close();
+          },
+        }),
+      });
+      expect(body).to.deep.eq(message);
+    }
+  );
+
+  it.skipIf(Number(nodeMajorVersion) < 18)("Handle Readable body", async () => {
+    const message = "Hallo von Pascal";
+    const { body } = await $fetch(getURL("echo"), {
+      method: "POST",
+      headers: {
+        "content-length": "16",
+      },
+      body: new Readable({
+        read() {
+          this.push(message);
+          this.push(null); // eslint-disable-line unicorn/no-null
+        },
+      }),
+    });
+    expect(body).to.deep.eq(message);
+  });
+
   it("Bypass FormData body", async () => {
     const data = new FormData();
     data.append("foo", "bar");