Repository: unjs/h3
Pull request #432 (2023-07-10 10:35:23)

feat: support web streams
<!---
☝️ PR title should follow conventional commits (https://conventionalcommits.org) 
-->

### 🔗 Linked issue

[<!-- Please ensure there is an open issue and mention its number as #123 -->](https://github.com/unjs/nitro/issues/1327)

### ❓ Type of change

<!-- What types of changes does your code introduce? Put an `x` in all the boxes that apply. -->

- [ ] 📖 Documentation (updates to the documentation, readme, or JSdoc annotations)
- [ ] 🐞 Bug fix (a non-breaking change that fixes an issue)
- [x] 👌 Enhancement (improving an existing functionality like performance)
- [ ] ✨ New feature (a non-breaking change that adds functionality)
- [ ] 🧹 Chore (updates to the build process or auxiliary tools and libraries)
- [ ] ⚠️ Breaking change (fix or feature that would cause existing functionality to change)

### 📚 Description

This PR adds support for sending Web Streams for Node.js and for Worker environments (only via unenv+nitro). Also enables tests.

Users can directly return a web stream in an event handler.

Notes:
- Support for workers is experimental
- Node.js 18 is required for global `ReadableStream` (Node.js 16 does not support Web Streams)

### 📝 Checklist

<!-- Put an `x` in all the boxes that apply. -->
<!-- If your change requires a documentation PR, please link it appropriately -->
<!-- If you're unsure about any of these, don't hesitate to ask. We're here to help! -->

- [ ] I have linked an issue or discussion.
- [ ] I have updated the documentation accordingly.


codecov[bot]: ## [Codecov](https://app.codecov.io/gh/unjs/h3/pull/432?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=unjs) Report
> Merging [#432](https://app.codecov.io/gh/unjs/h3/pull/432?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=unjs) (1b05b7e) into [main](https://app.codecov.io/gh/unjs/h3/commit/e359f5f378a380fc8fd37b87d6dbbf1fdd0d78f3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=unjs) (e359f5f) will **decrease** coverage by `1.15%`.
> The diff coverage is `22.03%`.

```diff
@@            Coverage Diff             @@
##             main     #432      +/-   ##
==========================================
- Coverage   76.33%   75.19%   -1.15%     
==========================================
  Files          26       26              
  Lines        2502     2548      +46     
  Branches      354      356       +2     
==========================================
+ Hits         1910     1916       +6     
- Misses        592      632      +40     
```


| [Impacted Files](https://app.codecov.io/gh/unjs/h3/pull/432?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=unjs) | Coverage Δ | |
|---|---|---|
| [src/utils/response.ts](https://app.codecov.io/gh/unjs/h3/pull/432?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=unjs#diff-c3JjL3V0aWxzL3Jlc3BvbnNlLnRz) | `54.10% <22.03%> (-8.51%)` | :arrow_down: |



Patch:
diff --git a/src/utils/response.ts b/src/utils/response.ts
index d3ab95f..bf0117f 100644
--- a/src/utils/response.ts
+++ b/src/utils/response.ts
@@ -1,6 +1,6 @@
 import type { OutgoingMessage } from "node:http";
+import type { Readable } from "node:stream";
 import type { Socket } from "node:net";
-import { createError } from "../error";
 import type { H3Event } from "../event";
 import { MIMES } from "./consts";
 import { sanitizeStatusCode, sanitizeStatusMessage } from "./sanitize";
@@ -147,21 +147,71 @@ export function appendResponseHeader(
 
 export const appendHeader = appendResponseHeader;
 
-export function isStream(data: any) {
-  return (
-    data &&
-    typeof data === "object" &&
-    typeof data.pipe === "function" &&
-    typeof data.on === "function"
-  );
+export function isStream(data: any): data is Readable | ReadableStream {
+  if (!data || typeof data !== "object") {
+    return false;
+  }
+  // Node.js Readable Streams
+  if (typeof data.pipe === "function" && typeof data.on === "function") {
+    return true;
+  }
+  // Web Streams
+  if (typeof data.pipeTo === "function") {
+    return true;
+  }
+  return false;
 }
 
-export function sendStream(event: H3Event, data: any): Promise<void> {
-  return new Promise((resolve, reject) => {
-    data.pipe(event.node.res);
-    data.on("end", () => resolve());
-    data.on("error", (error: Error) => reject(createError(error)));
-  });
+export function sendStream(
+  event: H3Event,
+  stream: Readable | ReadableStream
+): Promise<void> {
+  // Validate input
+  if (!stream || typeof stream !== "object") {
+    throw new Error("[h3] Invalid stream provided.");
+  }
+
+  // Directly expose stream for worker environments (unjs/unenv)
+  (event.node.res as unknown as { _data: BodyInit })._data = stream as BodyInit;
+
+  // Early return if response Socket is not available for worker environments (unjs/nitro)
+  if (!event.node.res.socket) {
+    event._handled = true;
+    // TODO: Hook and handle stream errors
+    return Promise.resolve();
+  }
+
+  // Native Web Streams
+  if ("pipeTo" in stream) {
+    return stream
+      .pipeTo(
+        new WritableStream({
+          write(chunk) {
+            event.node.res.write(chunk);
+          },
+        })
+      )
+      .then(() => {
+        event.node.res.end();
+      });
+  }
+
+  // Node.js Readable streams
+  // https://nodejs.org/api/stream.html#readable-streams
+  if ("pipe" in stream) {
+    return new Promise<void>((resolve, reject) => {
+      stream.pipe(event.node.res);
+      stream.on("end", () => {
+        event.node.res.end();
+        resolve();
+      });
+      stream.on("error", (error: Error) => {
+        reject(error);
+      });
+    });
+  }
+
+  throw new Error("[h3] Invalid or incompatible stream provided.");
 }
 
 const noop = () => {};


Test patch:
diff --git a/test/app.test.ts b/test/app.test.ts
index 3eea7be..df56deb 100644
--- a/test/app.test.ts
+++ b/test/app.test.ts
@@ -7,8 +7,12 @@ import {
   App,
   eventHandler,
   fromNodeMiddleware,
+  createError,
 } from "../src";
 
+const readableStreamSupported =
+  typeof ReadableStream !== "undefined"; /* Node.js 16 */
+
 describe("app", () => {
   let app: App;
   let request: SuperTest<Test>;
@@ -58,12 +62,15 @@ describe("app", () => {
     expect(res.text).toBe("<h1>Hello world!</h1>");
   });
 
-  it.todo("can return Readable stream directly", async () => {
+  it("Node.js Readable Stream", async () => {
     app.use(
       eventHandler(() => {
-        const readable = new Readable();
-        readable.push(Buffer.from("<h1>Hello world!</h1>", "utf8"));
-        return readable;
+        return new Readable({
+          read() {
+            this.push(Buffer.from("<h1>Hello world!</h1>", "utf8"));
+            this.push(null);
+          },
+        });
       })
     );
     const res = await request.get("/");
@@ -72,23 +79,67 @@ describe("app", () => {
     expect(res.header["transfer-encoding"]).toBe("chunked");
   });
 
-  it.todo("can return Readable stream that may throw", async () => {
+  it("Node.js Readable Stream with Error", async () => {
+    app.use(
+      eventHandler(() => {
+        return new Readable({
+          read() {
+            this.push(Buffer.from("123", "utf8"));
+            this.push(null);
+          },
+        }).pipe(
+          new Transform({
+            transform(_chunk, _encoding, callback) {
+              const err = createError({
+                statusCode: 500,
+                statusText: "test",
+              });
+              setTimeout(() => callback(err), 0);
+            },
+          })
+        );
+      })
+    );
+    const res = await request.get("/");
+    expect(res.statusCode).toBe(500);
+    expect(JSON.parse(res.text).statusMessage).toBe("test");
+  });
+
+  it.runIf(readableStreamSupported)("Web Stream", async () => {
     app.use(
       eventHandler(() => {
-        const readable = new Readable();
-        const willThrow = new Transform({
-          transform(_chunk, _encoding, callback) {
-            setTimeout(() => callback(new Error("test")), 0);
+        return new ReadableStream({
+          start(controller) {
+            const encoder = new TextEncoder();
+            controller.enqueue(encoder.encode("<h1>Hello world!</h1>"));
+            controller.close();
           },
         });
-        readable.push(Buffer.from("<h1>Hello world!</h1>", "utf8"));
+      })
+    );
+    const res = await request.get("/");
 
-        return readable.pipe(willThrow);
+    expect(res.text).toBe("<h1>Hello world!</h1>");
+    expect(res.header["transfer-encoding"]).toBe("chunked");
+  });
+
+  it.runIf(readableStreamSupported)("Web Stream with Error", async () => {
+    app.use(
+      eventHandler(() => {
+        return new ReadableStream({
+          start() {
+            throw createError({
+              statusCode: 500,
+              statusText: "test",
+            });
+          },
+        });
       })
     );
     const res = await request.get("/");
 
-    expect(res.status).toBe(500);
+    expect(res.statusCode).toBe(500);
+    expect(JSON.parse(res.text).statusMessage).toBe("test");
   });
 
   it("can return HTML directly", async () => {