fix(fromNodeHandler): pipe responses once (#1273)

Author: pi0

src/adapters.ts

@@ -89,8 +89,22 @@ function callNodeHandler(
   return new Promise((resolve, reject) => {
     res.once("close", () => resolve(kHandled));
     res.once("finish", () => resolve(kHandled));
-    res.once("pipe", (stream) => resolve(stream));
     res.once("error", (error) => reject(error));
+    res.once("pipe", (stream) => {
+      resolve(
+        new Promise((resolve, reject) => {
+          stream.once("close", () => resolve(kHandled));
+          stream.once("error", (error: any) => {
+            console.error("[h3] Stream error in Node.js handler", {
+              cause: error,
+            });
+            // We cannot alter the outgoing response at this point
+            // TODO: We might at least call h3 error hook here by exposing app to node request
+            reject(kHandled);
+          });
+        }),
+      );
+    });
     try {
       if (isMiddleware) {
         Promise.resolve(

test/app.test.ts

@@ -290,6 +290,44 @@ describeMatrix("app", (t, { it, expect }) => {
     },
   );
 
+  it.skipIf(t.target !== "node")("fromNodeHandler + piping", async () => {
+    t.app.all(
+      "/*",
+      fromNodeHandler((req, res) => {
+        const iterator = (async function* () {
+          yield "item1,";
+          yield "item2,";
+          yield "item3";
+        })();
+        NodeStreamReadable.from(iterator).pipe(res);
+      }),
+    );
+    const res = await t.fetch("/");
+    expect(res.status).toBe(200);
+    expect(await res.text()).toBe("item1,item2,item3");
+  });
+
+  it.skipIf(t.target !== "node")(
+    "fromNodeHandler + piping (with Error and custom status)",
+    async () => {
+      t.app.all(
+        "/*",
+        fromNodeHandler((req, res) => {
+          res.statusCode = 201;
+          const iterator = (async function* () {
+            yield "item1,";
+            yield "item2";
+            throw new Error("Test Error");
+          })();
+          NodeStreamReadable.from(iterator).pipe(res);
+        }),
+      );
+      const res = await t.fetch("/");
+      expect(res.status).toBe(201);
+      expect(await res.text()).toBe("item1,item2");
+    },
+  );
+
   it("set headers via event.res + Response (mutable)", async () => {
     t.app.use((event) => {
       event.res.headers.set("x-from-event", "1");