Repository: denoland/deno
Pull request: feat(ext/process): add collector utilities to `ChildProcess.stdout` & `ChildProcess.stderr`

This commit adds `SubprocessReadableStream` interface that are used
for `ChildProcess.stdout` and `ChildProcess.stderr`. It's an extension to
a regular `ReadableStream` that provides convenience methods to collect
the output and parse it with `.bytes()`, `.arrayBuffer()`, `.text()` and `.json()`
helper methods (similarly to the ones available on `Response`).


Closes https://github.com/denoland/deno/issues/30323

Changed files:
filename: cli/tsc/dts/lib.deno.ns.d.ts
patch: @@ -3719,8 +3719,8 @@ declare namespace Deno {
    */
   export class ChildProcess implements AsyncDisposable {
     get stdin(): WritableStream<Uint8Array<ArrayBufferLike>>;
-    get stdout(): ReadableStream<Uint8Array<ArrayBuffer>>;
-    get stderr(): ReadableStream<Uint8Array<ArrayBuffer>>;
+    get stdout(): SubprocessReadableStream;
+    get stderr(): SubprocessReadableStream;
     readonly pid: number;
     /** Get the status of the child. */
     readonly status: Promise<CommandStatus>;
@@ -3746,6 +3746,36 @@ declare namespace Deno {
     [Symbol.asyncDispose](): Promise<void>;
   }
 
+  /**
+   * The interface for stdout and stderr streams for child process returned from
+   * {@linkcode Deno.Command.spawn}.
+   *
+   * @category Subprocess
+   */
+  export interface SubprocessReadableStream
+    extends ReadableStream<Uint8Array<ArrayBuffer>> {
+    /**
+     * Reads the stream to completion. It returns a promise that resolves with
+     * an `ArrayBuffer`.
+     */
+    arrayBuffer(): Promise<ArrayBuffer>;
+    /**
+     * Reads the stream to completion. It returns a promise that resolves with
+     * a `Uint8Array`.
+     */
+    bytes(): Promise<Uint8Array>;
+    /**
+     * Reads the stream to completion. It returns a promise that resolves with
+     * the result of parsing the body text as JSON.
+     */
+    json(): Promise<any>;
+    /**
+     * Reads the stream to completion. It returns a promise that resolves with
+     * a `USVString` (text).
+     */
+    text(): Promise<string>;
+  }
+
   /**
    * Options which can be set when calling {@linkcode Deno.Command}.
    *

filename: ext/fetch/22_body.js
patch: @@ -543,4 +543,4 @@ webidl.converters["BodyInit_DOMString?"] = webidl.createNullableConverter(
   webidl.converters["BodyInit_DOMString"],
 );
 
-export { extractBody, InnerBody, mixinBody };
+export { extractBody, InnerBody, mixinBody, packageData };

filename: ext/process/40_process.js
patch: @@ -31,8 +31,10 @@ import {
   pathFromURL,
   SymbolAsyncDispose,
 } from "ext:deno_web/00_infra.js";
+import { packageData } from "ext:deno_fetch/22_body.js";
 import * as abortSignal from "ext:deno_web/03_abort_signal.js";
 import {
+  ReadableStream,
   readableStreamCollectIntoUint8Array,
   readableStreamForRidUnrefable,
   readableStreamForRidUnrefableRef,
@@ -292,11 +294,17 @@ class ChildProcess {
     }
 
     if (stdoutRid !== null) {
-      this.#stdout = readableStreamForRidUnrefable(stdoutRid);
+      this.#stdout = readableStreamForRidUnrefable(
+        stdoutRid,
+        ReadableStreamWithCollectors,
+      );
     }
 
     if (stderrRid !== null) {
-      this.#stderr = readableStreamForRidUnrefable(stderrRid);
+      this.#stderr = readableStreamForRidUnrefable(
+        stderrRid,
+        ReadableStreamWithCollectors,
+      );
     }
 
     const onAbort = () => {
@@ -387,6 +395,32 @@ class ChildProcess {
   }
 }
 
+class ReadableStreamWithCollectors extends ReadableStream {
+  constructor(underlyingSource = undefined, strategy = undefined) {
+    super(underlyingSource, strategy);
+  }
+
+  async arrayBuffer() {
+    const buffer = await readableStreamCollectIntoUint8Array(this);
+    return packageData(buffer, "ArrayBuffer", null);
+  }
+
+  async bytes() {
+    const buffer = await readableStreamCollectIntoUint8Array(this);
+    return packageData(buffer, "bytes", null);
+  }
+
+  async json() {
+    const buffer = await readableStreamCollectIntoUint8Array(this);
+    return packageData(buffer, "JSON", null);
+  }
+
+  async text() {
+    const buffer = await readableStreamCollectIntoUint8Array(this);
+    return packageData(buffer, "text", null);
+  }
+}
+
 function spawn(command, options) {
   if (options?.stdin === "piped") {
     throw new TypeError(

filename: ext/web/06_streams.js
patch: @@ -1006,10 +1006,11 @@ const _isUnref = Symbol("isUnref");
  * compatible with fast-path resource-backed streams.
  *
  * @param {number} rid The resource ID to read from.
+ * @param constructor A class that extends ReadableStream.
  * @returns {ReadableStream<Uint8Array>}
  */
-function readableStreamForRidUnrefable(rid) {
-  const stream = new ReadableStream(_brand);
+function readableStreamForRidUnrefable(rid, constructor = ReadableStream) {
+  const stream = new constructor(_brand);
   stream[promiseSymbol] = undefined;
   stream[_isUnref] = false;
   stream[_resourceBackingUnrefable] = { rid, autoClose: true };

filename: tests/unit/command_test.ts
patch: @@ -1115,3 +1115,56 @@ Deno.test(
     Deno.removeSignalListener("SIGCHLD", cb);
   },
 );
+
+Deno.test({ permissions: { run: true } }, async function collectArrayBuffer() {
+  const process = new Deno.Command(Deno.execPath(), {
+    args: ["eval", "console.log('hello')"],
+    stdout: "piped",
+  }).spawn();
+
+  const output = await process.stdout.arrayBuffer();
+  assert(output instanceof ArrayBuffer);
+  assertEquals(
+    new Uint8Array(output),
+    new Uint8Array([104, 101, 108, 108, 111, 10]),
+  );
+
+  await process.status;
+});
+
+Deno.test({ permissions: { run: true } }, async function collectBytes() {
+  const process = new Deno.Command(Deno.execPath(), {
+    args: ["eval", "console.log('hello')"],
+    stdout: "piped",
+  }).spawn();
+
+  const output = await process.stdout.bytes();
+  assert(output instanceof Uint8Array);
+  assertEquals(output, new Uint8Array([104, 101, 108, 108, 111, 10]));
+
+  await process.status;
+});
+
+Deno.test({ permissions: { run: true } }, async function collectJSON() {
+  const process = new Deno.Command(Deno.execPath(), {
+    args: ["eval", "console.log(JSON.stringify({foo: 'bar'}))"],
+    stdout: "piped",
+  }).spawn();
+
+  const output = await process.stdout.json();
+  assertEquals(output, { foo: "bar" });
+
+  await process.status;
+});
+
+Deno.test({ permissions: { run: true } }, async function collectText() {
+  const process = new Deno.Command(Deno.execPath(), {
+    args: ["eval", "console.log('hello')"],
+    stdout: "piped",
+  }).spawn();
+
+  const output = await process.stdout.text();
+  assertEquals(output, "hello\n");
+
+  await process.status;
+});