Merge pull request #5871 from NomicFoundation/output-streams

Await output streams

v-next/hardhat-node-test-runner/src/task-action.ts

@@ -2,12 +2,13 @@ import type { HardhatConfig } from "@ignored/hardhat-vnext/types/config";
 import type { NewTaskActionFunction } from "@ignored/hardhat-vnext/types/tasks";
 import type { LastParameter } from "@ignored/hardhat-vnext/types/utils";
 
-import { finished } from "node:stream/promises";
+import { pipeline } from "node:stream/promises";
 import { run } from "node:test";
 import { fileURLToPath } from "node:url";
 
 import { hardhatTestReporter } from "@ignored/hardhat-vnext-node-test-reporter";
 import { getAllFilesMatching } from "@ignored/hardhat-vnext-utils/fs";
+import { createNonClosingWriter } from "@ignored/hardhat-vnext-utils/stream";
 
 interface TestActionArguments {
   testFiles: string[];
@@ -92,9 +93,7 @@ const testWithHardhat: NewTaskActionFunction<TestActionArguments> = async (
       })
       .compose(customReporter);
 
-    reporterStream.pipe(process.stdout);
-
-    await finished(reporterStream);
+    await pipeline(reporterStream, createNonClosingWriter(process.stdout));
 
     return failures;
   }

v-next/hardhat-utils/package.json

@@ -32,6 +32,7 @@
     "./request": "./dist/src/request.js",
     "./resolve": "./dist/src/resolve.js",
     "./string": "./dist/src/string.js",
+    "./stream": "./dist/src/stream.js",
     "./subprocess": "./dist/src/subprocess.js"
   },
   "keywords": [

v-next/hardhat-utils/src/stream.ts

@@ -0,0 +1,16 @@
+import { Writable } from "node:stream";
+
+/**
+ * Creates a Tansform that writes everything to actualWritable, without closing it
+ * when finished.
+ *
+ * This is useful to pipe things to stdout, without closing it, while being
+ * able to await for the result of the pipe to finish.
+ */
+export function createNonClosingWriter(actualWritable: Writable): Writable {
+  return new Writable({
+    write(chunk, encoding, callback) {
+      actualWritable.write(chunk, encoding, callback);
+    },
+  });
+}

v-next/hardhat-utils/test/stream.ts

@@ -0,0 +1,65 @@
+import assert from "node:assert/strict";
+import { Readable, Writable } from "node:stream";
+import { pipeline } from "node:stream/promises";
+import { describe, it } from "node:test";
+
+import { createNonClosingWriter } from "../src/stream.js";
+
+function createFixedReadable(data: Buffer[]) {
+  const streamData = [...data];
+
+  const stream = new Readable({
+    encoding: "utf8",
+    read(_size) {
+      const chunk = streamData.shift();
+
+      if (chunk === undefined) {
+        this.push(null);
+        return;
+      }
+
+      this.push(chunk);
+    },
+  });
+
+  return stream;
+}
+
+function createWritable() {
+  const data: string[] = [];
+  const writable = new Writable({
+    write(chunk, _encoding, callback) {
+      data.push(chunk);
+
+      callback();
+    },
+  });
+
+  return { writable, data };
+}
+
+describe("stream", () => {
+  describe("createNonClosingWriter", () => {
+    const FIXTURE_DATA = [Buffer.from("a"), Buffer.from("b"), Buffer.from("c")];
+
+    it("Should not close the actual writable when finished", async () => {
+      const readable = createFixedReadable(FIXTURE_DATA);
+      const { writable } = createWritable();
+      const writer = createNonClosingWriter(writable);
+      await pipeline(readable, writer);
+
+      assert.equal(readable.closed, true);
+      assert.equal(writer.closed, true);
+      assert.equal(writable.closed, false);
+    });
+
+    it("Should write all the data to the actual writable", async () => {
+      const readable = createFixedReadable(FIXTURE_DATA);
+      const { writable, data } = createWritable();
+      const writer = createNonClosingWriter(writable);
+      await pipeline(readable, writer);
+
+      assert.deepEqual(data, FIXTURE_DATA);
+    });
+  });
+});

v-next/hardhat/src/internal/builtin-plugins/solidity-test/task-action.ts

@@ -4,6 +4,8 @@ import type { NewTaskActionFunction } from "../../../types/tasks.js";
 
 import { finished } from "node:stream/promises";
 
+import { createNonClosingWriter } from "@ignored/hardhat-vnext-utils/stream";
+
 import { getArtifacts, isTestArtifact } from "./helpers.js";
 import { testReporter } from "./reporter.js";
 import { run } from "./runner.js";
@@ -53,19 +55,18 @@ const runSolidityTests: NewTaskActionFunction<TestActionArguments> = async (
     })
     .compose(testReporter);
 
-  testReporterStream.pipe(process.stdout);
+  const outputStream = testReporterStream.pipe(
+    createNonClosingWriter(process.stdout),
+  );
 
   try {
     // NOTE: We're awaiting the original run stream to finish to catch any
     // errors produced by the runner.
     await finished(runStream);
 
-    // We also await the test reporter stream to finish to catch any error, and
+    // We also await the output stream to finish, as we want to wait for it
     // to avoid returning before the whole output was generated.
-    //
-    // NOTE: We don't await the restult of piping it to stdout, as that is
-    // ignored.
-    await finished(testReporterStream);
+    await finished(outputStream);
   } catch (error) {
     console.error(error);
     includesErrors = true;