Add .readableStream(), .writableStream() and .transformStream() methods

docs/api.md

@@ -498,6 +498,33 @@ Converts the subprocess to a duplex stream.
 
 [More info.](streams.md#converting-a-subprocess-to-a-stream)
 
+### subprocess.readableStream(readableOptions?)
+
+`readableOptions`: [`ReadableOptions`](#readableoptions)\
+_Returns_: [`ReadableStream`](https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream) web stream
+
+Converts the subprocess to a readable web stream.
+
+[More info.](streams.md#converting-a-subprocess-to-a-web-stream)
+
+### subprocess.writableStream(writableOptions?)
+
+`writableOptions`: [`WritableOptions`](#writableoptions)\
+_Returns_: [`WritableStream`](https://developer.mozilla.org/en-US/docs/Web/API/WritableStream) web stream
+
+Converts the subprocess to a writable web stream.
+
+[More info.](streams.md#converting-a-subprocess-to-a-web-stream)
+
+### subprocess.transformStream(duplexOptions?)
+
+`duplexOptions`: [`ReadableOptions | WritableOptions`](#readableoptions)\
+_Returns_: [`{readable: ReadableStream, writable: WritableStream}`](https://developer.mozilla.org/en-US/docs/Web/API/TransformStream)
+
+Converts the subprocess to a [`{readable, writable}`](https://developer.mozilla.org/en-US/docs/Web/API/TransformStream) pair of web streams.
+
+[More info.](streams.md#converting-a-subprocess-to-a-web-stream)
+
 ## Result
 
 _TypeScript:_ [`Result`](typescript.md) or [`SyncResult`](typescript.md)\

docs/streams.md

@@ -118,6 +118,20 @@ When using [`subprocess.readable()`](api.md#subprocessreadablereadableoptions),
 
 This means you do not need to `await` the subprocess' [promise](execution.md#result). On the other hand, you (or the library using the stream) do need to both consume the stream, and handle its `error` event. This can be done by using [`await finished(stream)`](https://nodejs.org/api/stream.html#streamfinishedstream-options), [`await pipeline(..., stream, ...)`](https://nodejs.org/api/stream.html#streampipelinesource-transforms-destination-options) or [`await text(stream)`](https://nodejs.org/api/webstreams.html#streamconsumerstextstream) which throw an exception when the stream errors.
 
+## Converting a subprocess to a web stream
+
+The [`subprocess.readableStream()`](api.md#subprocessreadablestreamreadableoptions), [`subprocess.writableStream()`](api.md#subprocesswritablestreamwritableoptions) and [`subprocess.transformStream()`](api.md#subprocesstransformstreamduplexoptions) methods are the [web streams](#web-streams) counterparts of [`subprocess.readable()`](api.md#subprocessreadablereadableoptions), [`subprocess.writable()`](api.md#subprocesswritablewritableoptions) and [`subprocess.duplex()`](api.md#subprocessduplexduplexoptions). They behave the same way, including [error handling](#error-handling) and the [`from`](api.md#readableoptionsfrom)/[`to`](api.md#writableoptionsto) options, but return a [`ReadableStream`](https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream), a [`WritableStream`](https://developer.mozilla.org/en-US/docs/Web/API/WritableStream) and a [`{readable, writable}`](https://developer.mozilla.org/en-US/docs/Web/API/TransformStream) pair instead of Node.js streams.
+
+This is useful when using a library or API that expects web streams as arguments. In every other situation, the [Node.js stream methods](#converting-a-subprocess-to-a-stream) can be used instead.
+
+```js
+const readableStream = execa`npm run scaffold`.readableStream();
+
+const writableStream = execa`npm run scaffold`.writableStream();
+
+const {readable, writable} = execa`npm run scaffold`.transformStream();
+```
+
 <hr>
 
 [**Next**: 📞 Inter-process communication](ipc.md)\

lib/convert/add.js

@@ -2,6 +2,7 @@ import {initializeConcurrentStreams} from './concurrent.js';
 import {createReadable} from './readable.js';
 import {createWritable} from './writable.js';
 import {createDuplex} from './duplex.js';
+import {createReadableStream, createWritableStream, createTransformStream} from './web.js';
 import {createIterable} from './iterable.js';
 
 // Add methods to convert the subprocess to a stream or iterable
@@ -10,6 +11,9 @@ export const addConvertedStreams = (subprocess, {encoding}) => {
 	subprocess.readable = createReadable.bind(undefined, {subprocess, concurrentStreams, encoding});
 	subprocess.writable = createWritable.bind(undefined, {subprocess, concurrentStreams});
 	subprocess.duplex = createDuplex.bind(undefined, {subprocess, concurrentStreams, encoding});
+	subprocess.readableStream = createReadableStream.bind(undefined, subprocess);
+	subprocess.writableStream = createWritableStream.bind(undefined, subprocess);
+	subprocess.transformStream = createTransformStream.bind(undefined, subprocess);
 	subprocess.iterable = createIterable.bind(undefined, subprocess, encoding);
 	subprocess[Symbol.asyncIterator] = createIterable.bind(undefined, subprocess, encoding, {});
 };

lib/convert/web.js

@@ -0,0 +1,7 @@
+import {Readable, Writable, Duplex} from 'node:stream';
+
+// Web stream versions of `subprocess.readable()`, `subprocess.writable()` and `subprocess.duplex()`.
+// They wrap those methods with `.toWeb()`, so all error and abort handling is reused as is.
+export const createReadableStream = (subprocess, readableOptions) => Readable.toWeb(subprocess.readable(readableOptions));
+export const createWritableStream = (subprocess, writableOptions) => Writable.toWeb(subprocess.writable(writableOptions));
+export const createTransformStream = (subprocess, duplexOptions) => Duplex.toWeb(subprocess.duplex(duplexOptions));

lib/pipe/setup.js

@@ -3,6 +3,7 @@ import {SUBPROCESS_OPTIONS} from '../arguments/fd-options.js';
 import {initializeConcurrentStreams} from '../convert/concurrent.js';
 import {createIterable} from '../convert/iterable.js';
 import {createReadable} from '../convert/readable.js';
+import {createReadableStream} from '../convert/web.js';
 import {internalGetOneMessageOptions} from '../ipc/get-one.js';
 import {internalGetEachMessageOptions} from '../ipc/get-each.js';
 import {normalizePipeArguments} from './pipe-arguments.js';
@@ -60,6 +61,7 @@ const forwardReadableMethods = (promise, destination) => {
 		concurrentStreams,
 		encoding,
 	});
+	promise.readableStream = createReadableStream.bind(undefined, promise);
 	forwardAll(promise, destination);
 };
 
@@ -175,7 +177,7 @@ const abortOnSignal = (signal, controller) => {
 	}, {once: true, signal: controller.signal});
 };
 
-// `writable()` and `duplex()` are intentionally not forwarded: they write to the destination's `stdin`, which is already being piped from the source.
+// `writable()`, `duplex()`, `writableStream()` and `transformStream()` are intentionally not forwarded: they write to the destination's `stdin`, which is already being piped from the source.
 
 // Asynchronous logic when piping subprocesses
 const handlePipePromise = async ({

lib/return/early-error.js

@@ -16,7 +16,14 @@ export const handleEarlyError = ({error, command, escapedCommand, fileDescriptor
 
 	const subprocess = new ChildProcess();
 	createDummyStreams(subprocess, fileDescriptors);
-	Object.assign(subprocess, {readable, writable, duplex});
+	Object.assign(subprocess, {
+		readable,
+		writable,
+		duplex,
+		readableStream,
+		writableStream,
+		transformStream,
+	});
 
 	const earlyError = makeEarlyError({
 		error,
@@ -56,5 +63,8 @@ const createDummyStream = () => {
 const readable = () => new Readable({read() {}});
 const writable = () => new Writable({write() {}});
 const duplex = () => new Duplex({read() {}, write() {}});
+const readableStream = () => Readable.toWeb(readable());
+const writableStream = () => Writable.toWeb(writable());
+const transformStream = () => Duplex.toWeb(duplex());
 
 const handleDummyPromise = async (error, verboseInfo, options) => handleResult(error, verboseInfo, options);

test-d/convert/readable-stream.test-d.ts

@@ -0,0 +1,26 @@
+import type {ReadableStream} from 'node:stream/web';
+import {expectType, expectError} from 'tsd';
+import {execa} from '../../index.js';
+
+const subprocess = execa('unicorns');
+
+expectType<ReadableStream>(subprocess.readableStream());
+
+subprocess.readableStream({from: 'stdout'});
+subprocess.readableStream({from: 'stderr'});
+subprocess.readableStream({from: 'all'});
+subprocess.readableStream({from: 'fd3'});
+expectError(subprocess.readableStream({from: 'fd3' as string}));
+expectError(subprocess.readableStream({from: 'stdin'}));
+expectError(subprocess.readableStream({from: 'fd'}));
+expectError(subprocess.readableStream({from: 'fdNotANumber'}));
+expectError(subprocess.readableStream({to: 'stdin'}));
+
+subprocess.readableStream({binary: false});
+expectError(subprocess.readableStream({binary: 'false'}));
+
+subprocess.readableStream({preserveNewlines: false});
+expectError(subprocess.readableStream({preserveNewlines: 'false'}));
+
+expectError(subprocess.readableStream('stdout'));
+expectError(subprocess.readableStream({other: 'stdout'}));

test-d/convert/transform-stream.test-d.ts

@@ -0,0 +1,32 @@
+import type {ReadableWritablePair} from 'node:stream/web';
+import {expectType, expectError} from 'tsd';
+import {execa} from '../../index.js';
+
+const subprocess = execa('unicorns');
+
+expectType<ReadableWritablePair>(subprocess.transformStream());
+
+subprocess.transformStream({from: 'stdout'});
+subprocess.transformStream({from: 'stderr'});
+subprocess.transformStream({from: 'all'});
+subprocess.transformStream({from: 'fd3'});
+subprocess.transformStream({to: 'fd3'});
+subprocess.transformStream({from: 'stdout', to: 'stdin'});
+subprocess.transformStream({from: 'stdout', to: 'fd3'});
+expectError(subprocess.transformStream({from: 'stdout' as string}));
+expectError(subprocess.transformStream({to: 'fd3' as string}));
+expectError(subprocess.transformStream({from: 'stdin'}));
+expectError(subprocess.transformStream({from: 'stderr', to: 'stdout'}));
+expectError(subprocess.transformStream({from: 'fd'}));
+expectError(subprocess.transformStream({from: 'fdNotANumber'}));
+expectError(subprocess.transformStream({to: 'fd'}));
+expectError(subprocess.transformStream({to: 'fdNotANumber'}));
+
+subprocess.transformStream({binary: false});
+expectError(subprocess.transformStream({binary: 'false'}));
+
+subprocess.transformStream({preserveNewlines: false});
+expectError(subprocess.transformStream({preserveNewlines: 'false'}));
+
+expectError(subprocess.transformStream('stdout'));
+expectError(subprocess.transformStream({other: 'stdout'}));

test-d/convert/writable-stream.test-d.ts

@@ -0,0 +1,25 @@
+import type {WritableStream} from 'node:stream/web';
+import {expectType, expectError} from 'tsd';
+import {execa} from '../../index.js';
+
+const subprocess = execa('unicorns');
+
+expectType<WritableStream>(subprocess.writableStream());
+
+subprocess.writableStream({to: 'stdin'});
+subprocess.writableStream({to: 'fd3'});
+expectError(subprocess.writableStream({to: 'fd3' as string}));
+expectError(subprocess.writableStream({to: 'stdout'}));
+expectError(subprocess.writableStream({to: 'fd'}));
+expectError(subprocess.writableStream({to: 'fdNotANumber'}));
+expectError(subprocess.writableStream({from: 'stdout'}));
+
+expectError(subprocess.writableStream({binary: false}));
+
+expectError(subprocess.writableStream({preserveNewlines: false}));
+
+expectError(subprocess.writableStream('stdin'));
+expectError(subprocess.writableStream({other: 'stdin'}));
+
+const inputPipeSubprocess = execa('unicorns', {stdio: ['pipe', 'pipe', 'pipe', {value: 'pipe', input: true}]});
+inputPipeSubprocess.writableStream({to: 'fd3'});

test-d/pipe.test-d.ts

@@ -1,5 +1,6 @@
 import {createWriteStream} from 'node:fs';
 import type {Readable} from 'node:stream';
+import type {ReadableStream} from 'node:stream/web';
 import {expectType, expectNotType, expectError} from 'tsd';
 import {
 	execa,
@@ -269,6 +270,7 @@ for await (const pipeLine of subprocess.pipe(bufferSubprocess)) {
 }
 
 expectType<Readable>(subprocess.pipe`stdin`.readable());
+expectType<ReadableStream>(subprocess.pipe`stdin`.readableStream());
 
 expectType<Readable>(subprocess.pipe({all: true})`stdin`.all);
 expectType<undefined>(subprocess.pipe`stdin`.all);
@@ -279,6 +281,8 @@ expectType<Promise<void>>(ipcPipeResult.sendMessage('message'));
 expectType<Promise<Message<'advanced'>>>(ipcPipeResult.getOneMessage());
 expectType<AsyncIterableIterator<Message<'advanced'>>>(ipcPipeResult.getEachMessage());
 
-// `writable()` and `duplex()` write to the destination's `stdin`, which is already piped from the source, so they are not forwarded.
+// `writable()`, `duplex()`, `writableStream()` and `transformStream()` write to the destination's `stdin`, which is already piped from the source, so they are not forwarded.
 expectError(subprocess.pipe`stdin`.writable());
 expectError(subprocess.pipe`stdin`.duplex());
+expectError(subprocess.pipe`stdin`.writableStream());
+expectError(subprocess.pipe`stdin`.transformStream());

test/convert/web.js

@@ -0,0 +1,104 @@
+import {ReadableStream, WritableStream} from 'node:stream/web';
+import test from 'ava';
+import {execa} from '../../index.js';
+import {setFixtureDirectory} from '../helpers/fixtures-directory.js';
+import {
+	assertStreamOutput,
+	assertStreamReadError,
+	assertSubprocessOutput,
+	assertSubprocessError,
+	assertPromiseError,
+	getReadableSubprocess,
+	getWritableSubprocess,
+	getReadWriteSubprocess,
+} from '../helpers/convert.js';
+import {foobarString} from '../helpers/input.js';
+
+setFixtureDirectory();
+
+const writeToStream = async (stream, input = foobarString) => {
+	const writer = stream.getWriter();
+	await writer.write(new TextEncoder().encode(input));
+	await writer.close();
+};
+
+test('.readableStream() success', async t => {
+	const subprocess = getReadableSubprocess();
+	const stream = subprocess.readableStream();
+
+	t.true(stream instanceof ReadableStream);
+
+	await assertStreamOutput(t, stream);
+	await assertSubprocessOutput(t, subprocess);
+});
+
+test('.readableStream() can use from', async t => {
+	const subprocess = execa('noop-fd.js', ['2', foobarString]);
+	const stream = subprocess.readableStream({from: 'stderr'});
+
+	await assertStreamOutput(t, stream);
+});
+
+test('.writableStream() success', async t => {
+	const subprocess = getWritableSubprocess();
+	const stream = subprocess.writableStream();
+
+	t.true(stream instanceof WritableStream);
+
+	await writeToStream(stream);
+	await assertSubprocessOutput(t, subprocess, foobarString, 2);
+});
+
+test('.writableStream() can use to', async t => {
+	const subprocess = execa('stdin-fd.js', ['0']);
+	const stream = subprocess.writableStream({to: 'stdin'});
+
+	await writeToStream(stream);
+	await assertSubprocessOutput(t, subprocess);
+});
+
+test('.transformStream() success', async t => {
+	const subprocess = getReadWriteSubprocess();
+	const {readable, writable} = subprocess.transformStream();
+
+	t.true(readable instanceof ReadableStream);
+	t.true(writable instanceof WritableStream);
+
+	await writeToStream(writable);
+	await assertStreamOutput(t, readable);
+	await assertSubprocessOutput(t, subprocess);
+});
+
+test('subprocess fail -> .readableStream() error', async t => {
+	const subprocess = getReadWriteSubprocess();
+	const stream = subprocess.readableStream();
+
+	const cause = new Error(foobarString);
+	subprocess.kill(cause);
+
+	await assertStreamReadError(t, stream, {cause});
+	await assertSubprocessError(t, subprocess, {cause});
+});
+
+test('subprocess fail -> .writableStream() error', async t => {
+	const subprocess = getReadWriteSubprocess();
+	const stream = subprocess.writableStream();
+	const writer = stream.getWriter();
+
+	const cause = new Error(foobarString);
+	subprocess.kill(cause);
+
+	await assertPromiseError(t, writer.closed, {cause});
+	await assertSubprocessError(t, subprocess, {cause});
+});
+
+test('subprocess fail -> .transformStream() error', async t => {
+	const subprocess = getReadWriteSubprocess();
+	const {readable} = subprocess.transformStream();
+
+	const cause = new Error(foobarString);
+	subprocess.kill(cause);
+
+	await assertStreamReadError(t, readable, {cause});
+	await assertSubprocessError(t, subprocess, {cause});
+});

test/pipe/methods.js

@@ -66,6 +66,10 @@ test('The .pipe() return value has a .readable() method', async t => {
 	t.is(await text(pipeSimple().readable()), simpleFull);
 });
 
+test('The .pipe() return value has a .readableStream() method', async t => {
+	t.is(await text(pipeSimple().readableStream()), simpleFull);
+});
+
 test('The .pipe() return value .readable() can be called multiple times', async t => {
 	const piped = pipeSimple();
 	const [output, secondOutput] = await Promise.all([
@@ -91,6 +95,8 @@ test('The .pipe() return value does not have a .writable() method', async t => {
 	const piped = pipeSimple();
 	t.is(piped.writable, undefined);
 	t.is(piped.duplex, undefined);
+	t.is(piped.writableStream, undefined);
+	t.is(piped.transformStream, undefined);
 	await piped;
 });
 
@@ -375,6 +381,12 @@ test('The .pipe() return value .readable() waits for source failure', async t =>
 	await t.throwsAsync(text(piped.readable()), {message: /Command failed with exit code 2/});
 });
 
+test('The .pipe() return value .readableStream() waits for source failure', async t => {
+	const piped = execa('fail.js').pipe('stdin.js');
+
+	await t.throwsAsync(text(piped.readableStream()), {message: /Command failed with exit code 2/});
+});
+
 test('The .pipe() return value .all waits for source failure', async t => {
 	const piped = execa('fail.js').pipe('stdin.js', {all: true});
 

test/return/early-error.js

@@ -103,6 +103,16 @@ test('child_process.spawn() early errors can use .readable()', testEarlyErrorCon
 test('child_process.spawn() early errors can use .writable()', testEarlyErrorConvertor, 'writable');
 test('child_process.spawn() early errors can use .duplex()', testEarlyErrorConvertor, 'duplex');
 
+const testEarlyErrorWebConvertor = async (t, streamMethod) => {
+	const subprocess = getEarlyErrorSubprocess();
+	subprocess[streamMethod]();
+	await t.throwsAsync(subprocess);
+};
+
+test('child_process.spawn() early errors can use .readableStream()', testEarlyErrorWebConvertor, 'readableStream');
+test('child_process.spawn() early errors can use .writableStream()', testEarlyErrorWebConvertor, 'writableStream');
+test('child_process.spawn() early errors can use .transformStream()', testEarlyErrorWebConvertor, 'transformStream');
+
 const testEarlyErrorStream = async (t, getStreamProperty, options) => {
 	const subprocess = getEarlyErrorSubprocess(options);
 	const stream = getStreamProperty(subprocess);