diff --git a/docs/api.md b/docs/api.md index 17e4b2dc4b..a8ef473d83 100644 --- a/docs/api.md +++ b/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)\ diff --git a/docs/streams.md b/docs/streams.md index 1faaf0e4dc..9b0c6868f6 100644 --- a/docs/streams.md +++ b/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(); +``` +
[**Next**: 📞 Inter-process communication](ipc.md)\ diff --git a/lib/convert/add.js b/lib/convert/add.js index 699aa2bacd..c9acb18892 100644 --- a/lib/convert/add.js +++ b/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, {}); }; diff --git a/lib/convert/web.js b/lib/convert/web.js new file mode 100644 index 0000000000..958408ef39 --- /dev/null +++ b/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)); diff --git a/lib/pipe/setup.js b/lib/pipe/setup.js index 6160f9267c..1fa18db7bb 100644 --- a/lib/pipe/setup.js +++ b/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 ({ diff --git a/lib/return/early-error.js b/lib/return/early-error.js index 0c968b4cc4..a52bbc250f 100644 --- a/lib/return/early-error.js +++ b/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); diff --git a/test-d/convert/readable-stream.test-d.ts b/test-d/convert/readable-stream.test-d.ts new file mode 100644 index 0000000000..699b39a954 --- /dev/null +++ b/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(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'})); diff --git a/test-d/convert/transform-stream.test-d.ts b/test-d/convert/transform-stream.test-d.ts new file mode 100644 index 0000000000..a6badd44ec --- /dev/null +++ b/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(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'})); diff --git a/test-d/convert/writable-stream.test-d.ts b/test-d/convert/writable-stream.test-d.ts new file mode 100644 index 0000000000..79d97b7600 --- /dev/null +++ b/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(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'}); diff --git a/test-d/pipe.test-d.ts b/test-d/pipe.test-d.ts index 7d39d835f9..e95ea800e2 100644 --- a/test-d/pipe.test-d.ts +++ b/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(subprocess.pipe`stdin`.readable()); +expectType(subprocess.pipe`stdin`.readableStream()); expectType(subprocess.pipe({all: true})`stdin`.all); expectType(subprocess.pipe`stdin`.all); @@ -279,6 +281,8 @@ expectType>(ipcPipeResult.sendMessage('message')); expectType>>(ipcPipeResult.getOneMessage()); expectType>>(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()); diff --git a/test/convert/web.js b/test/convert/web.js new file mode 100644 index 0000000000..6251e2c1cf --- /dev/null +++ b/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}); +}); diff --git a/test/pipe/methods.js b/test/pipe/methods.js index e5825d2736..1242abdd3c 100644 --- a/test/pipe/methods.js +++ b/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}); diff --git a/test/return/early-error.js b/test/return/early-error.js index bf6513b9b7..29d3271a73 100644 --- a/test/return/early-error.js +++ b/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); diff --git a/types/subprocess/subprocess.d.ts b/types/subprocess/subprocess.d.ts index de81d664b8..5df10d43a3 100644 --- a/types/subprocess/subprocess.d.ts +++ b/types/subprocess/subprocess.d.ts @@ -1,6 +1,7 @@ import type {ChildProcess} from 'node:child_process'; import type {SignalConstants} from 'node:os'; import type {Readable, Writable, Duplex} from 'node:stream'; +import type {ReadableStream, WritableStream, ReadableWritablePair} from 'node:stream/web'; import type {Options} from '../arguments/options.js'; import type {Result} from '../return/result.js'; import type {PipableSubprocess} from '../pipe.js'; @@ -17,7 +18,7 @@ import type {SubprocessAll} from './all.js'; // Read-side iteration, stream conversion and `all` methods. // These are shared between a subprocess and the return value of `subprocess.pipe()`, which forwards them from its destination subprocess. -// `writable()` and `duplex()` are not included: they write to `stdin`, which the pipe already feeds from its source. +// `writable()`, `duplex()`, `writableStream()` and `transformStream()` are not included: they write to `stdin`, which the pipe already feeds from its source. export type SubprocessResultMethods = { /** Stream combining/interleaving `subprocess.stdout` and `subprocess.stderr`. @@ -42,6 +43,11 @@ export type SubprocessResultMethods = { Converts the subprocess to a readable stream. */ readable(readableOptions?: ReadableOptions): Readable; + + /** + Converts the subprocess to a readable [web stream](https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream). + */ + readableStream(readableOptions?: ReadableOptions): ReadableStream; }; type ExecaCustomSubprocess = @@ -105,6 +111,16 @@ type ExecaCustomSubprocess = Converts the subprocess to a duplex stream. */ duplex(duplexOptions?: DuplexOptions): Duplex; + + /** + Converts the subprocess to a writable [web stream](https://developer.mozilla.org/en-US/docs/Web/API/WritableStream). + */ + writableStream(writableOptions?: WritableOptions): WritableStream; + + /** + Converts the subprocess to a [`{readable, writable}`](https://developer.mozilla.org/en-US/docs/Web/API/TransformStream) pair of web streams. + */ + transformStream(duplexOptions?: DuplexOptions): ReadableWritablePair; }; /**