Skip to content

Commit 29d9cfc

Browse files
sindresorhussindresorhus
authored
Add .readableStream(), .writableStream() and .transformStream() methods (#1254)
1 parent 522654a commit 29d9cfc

14 files changed

Lines changed: 297 additions & 4 deletions

File tree

‎docs/api.md‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -498,6 +498,33 @@ Converts the subprocess to a duplex stream.
498498

499499
[More info.](streams.md#converting-a-subprocess-to-a-stream)
500500

501+
### subprocess.readableStream(readableOptions?)
502+
503+
`readableOptions`: [`ReadableOptions`](#readableoptions)\
504+
_Returns_: [`ReadableStream`](https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream) web stream
505+
506+
Converts the subprocess to a readable web stream.
507+
508+
[More info.](streams.md#converting-a-subprocess-to-a-web-stream)
509+
510+
### subprocess.writableStream(writableOptions?)
511+
512+
`writableOptions`: [`WritableOptions`](#writableoptions)\
513+
_Returns_: [`WritableStream`](https://developer.mozilla.org/en-US/docs/Web/API/WritableStream) web stream
514+
515+
Converts the subprocess to a writable web stream.
516+
517+
[More info.](streams.md#converting-a-subprocess-to-a-web-stream)
518+
519+
### subprocess.transformStream(duplexOptions?)
520+
521+
`duplexOptions`: [`ReadableOptions | WritableOptions`](#readableoptions)\
522+
_Returns_: [`{readable: ReadableStream, writable: WritableStream}`](https://developer.mozilla.org/en-US/docs/Web/API/TransformStream)
523+
524+
Converts the subprocess to a [`{readable, writable}`](https://developer.mozilla.org/en-US/docs/Web/API/TransformStream) pair of web streams.
525+
526+
[More info.](streams.md#converting-a-subprocess-to-a-web-stream)
527+
501528
## Result
502529

503530
_TypeScript:_ [`Result`](typescript.md) or [`SyncResult`](typescript.md)\

‎docs/streams.md‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -118,6 +118,20 @@ When using [`subprocess.readable()`](api.md#subprocessreadablereadableoptions),
118118

119119
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.
120120

121+
## Converting a subprocess to a web stream
122+
123+
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.
124+
125+
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.
126+
127+
```js
128+
const readableStream = execa`npm run scaffold`.readableStream();
129+
130+
const writableStream = execa`npm run scaffold`.writableStream();
131+
132+
const {readable, writable} = execa`npm run scaffold`.transformStream();
133+
```
134+
121135
<hr>
122136

123137
[**Next**: 📞 Inter-process communication](ipc.md)\

‎lib/convert/add.js‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ import {initializeConcurrentStreams} from './concurrent.js';
22
import {createReadable} from './readable.js';
33
import {createWritable} from './writable.js';
44
import {createDuplex} from './duplex.js';
5+
import {createReadableStream, createWritableStream, createTransformStream} from './web.js';
56
import {createIterable} from './iterable.js';
67

78
// Add methods to convert the subprocess to a stream or iterable
@@ -10,6 +11,9 @@ export const addConvertedStreams = (subprocess, {encoding}) => {
1011
subprocess.readable = createReadable.bind(undefined, {subprocess, concurrentStreams, encoding});
1112
subprocess.writable = createWritable.bind(undefined, {subprocess, concurrentStreams});
1213
subprocess.duplex = createDuplex.bind(undefined, {subprocess, concurrentStreams, encoding});
14+
subprocess.readableStream = createReadableStream.bind(undefined, subprocess);
15+
subprocess.writableStream = createWritableStream.bind(undefined, subprocess);
16+
subprocess.transformStream = createTransformStream.bind(undefined, subprocess);
1317
subprocess.iterable = createIterable.bind(undefined, subprocess, encoding);
1418
subprocess[Symbol.asyncIterator] = createIterable.bind(undefined, subprocess, encoding, {});
1519
};

‎lib/convert/web.js‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
import {Readable, Writable, Duplex} from 'node:stream';
2+
3+
// Web stream versions of `subprocess.readable()`, `subprocess.writable()` and `subprocess.duplex()`.
4+
// They wrap those methods with `.toWeb()`, so all error and abort handling is reused as is.
5+
export const createReadableStream = (subprocess, readableOptions) => Readable.toWeb(subprocess.readable(readableOptions));
6+
export const createWritableStream = (subprocess, writableOptions) => Writable.toWeb(subprocess.writable(writableOptions));
7+
export const createTransformStream = (subprocess, duplexOptions) => Duplex.toWeb(subprocess.duplex(duplexOptions));

‎lib/pipe/setup.js‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ import {SUBPROCESS_OPTIONS} from '../arguments/fd-options.js';
33
import {initializeConcurrentStreams} from '../convert/concurrent.js';
44
import {createIterable} from '../convert/iterable.js';
55
import {createReadable} from '../convert/readable.js';
6+
import {createReadableStream} from '../convert/web.js';
67
import {internalGetOneMessageOptions} from '../ipc/get-one.js';
78
import {internalGetEachMessageOptions} from '../ipc/get-each.js';
89
import {normalizePipeArguments} from './pipe-arguments.js';
@@ -60,6 +61,7 @@ const forwardReadableMethods = (promise, destination) => {
6061
concurrentStreams,
6162
encoding,
6263
});
64+
promise.readableStream = createReadableStream.bind(undefined, promise);
6365
forwardAll(promise, destination);
6466
};
6567

@@ -175,7 +177,7 @@ const abortOnSignal = (signal, controller) => {
175177
}, {once: true, signal: controller.signal});
176178
};
177179

178-
// `writable()` and `duplex()` are intentionally not forwarded: they write to the destination's `stdin`, which is already being piped from the source.
180+
// `writable()`, `duplex()`, `writableStream()` and `transformStream()` are intentionally not forwarded: they write to the destination's `stdin`, which is already being piped from the source.
179181

180182
// Asynchronous logic when piping subprocesses
181183
const handlePipePromise = async ({

‎lib/return/early-error.js‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,14 @@ export const handleEarlyError = ({error, command, escapedCommand, fileDescriptor
1616

1717
const subprocess = new ChildProcess();
1818
createDummyStreams(subprocess, fileDescriptors);
19-
Object.assign(subprocess, {readable, writable, duplex});
19+
Object.assign(subprocess, {
20+
readable,
21+
writable,
22+
duplex,
23+
readableStream,
24+
writableStream,
25+
transformStream,
26+
});
2027

2128
const earlyError = makeEarlyError({
2229
error,
@@ -56,5 +63,8 @@ const createDummyStream = () => {
5663
const readable = () => new Readable({read() {}});
5764
const writable = () => new Writable({write() {}});
5865
const duplex = () => new Duplex({read() {}, write() {}});
66+
const readableStream = () => Readable.toWeb(readable());
67+
const writableStream = () => Writable.toWeb(writable());
68+
const transformStream = () => Duplex.toWeb(duplex());
5969

6070
const handleDummyPromise = async (error, verboseInfo, options) => handleResult(error, verboseInfo, options);

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

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
import type {ReadableStream} from 'node:stream/web';
2+
import {expectType, expectError} from 'tsd';
3+
import {execa} from '../../index.js';
4+
5+
const subprocess = execa('unicorns');
6+
7+
expectType<ReadableStream>(subprocess.readableStream());
8+
9+
subprocess.readableStream({from: 'stdout'});
10+
subprocess.readableStream({from: 'stderr'});
11+
subprocess.readableStream({from: 'all'});
12+
subprocess.readableStream({from: 'fd3'});
13+
expectError(subprocess.readableStream({from: 'fd3' as string}));
14+
expectError(subprocess.readableStream({from: 'stdin'}));
15+
expectError(subprocess.readableStream({from: 'fd'}));
16+
expectError(subprocess.readableStream({from: 'fdNotANumber'}));
17+
expectError(subprocess.readableStream({to: 'stdin'}));
18+
19+
subprocess.readableStream({binary: false});
20+
expectError(subprocess.readableStream({binary: 'false'}));
21+
22+
subprocess.readableStream({preserveNewlines: false});
23+
expectError(subprocess.readableStream({preserveNewlines: 'false'}));
24+
25+
expectError(subprocess.readableStream('stdout'));
26+
expectError(subprocess.readableStream({other: 'stdout'}));

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

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
import type {ReadableWritablePair} from 'node:stream/web';
2+
import {expectType, expectError} from 'tsd';
3+
import {execa} from '../../index.js';
4+
5+
const subprocess = execa('unicorns');
6+
7+
expectType<ReadableWritablePair>(subprocess.transformStream());
8+
9+
subprocess.transformStream({from: 'stdout'});
10+
subprocess.transformStream({from: 'stderr'});
11+
subprocess.transformStream({from: 'all'});
12+
subprocess.transformStream({from: 'fd3'});
13+
subprocess.transformStream({to: 'fd3'});
14+
subprocess.transformStream({from: 'stdout', to: 'stdin'});
15+
subprocess.transformStream({from: 'stdout', to: 'fd3'});
16+
expectError(subprocess.transformStream({from: 'stdout' as string}));
17+
expectError(subprocess.transformStream({to: 'fd3' as string}));
18+
expectError(subprocess.transformStream({from: 'stdin'}));
19+
expectError(subprocess.transformStream({from: 'stderr', to: 'stdout'}));
20+
expectError(subprocess.transformStream({from: 'fd'}));
21+
expectError(subprocess.transformStream({from: 'fdNotANumber'}));
22+
expectError(subprocess.transformStream({to: 'fd'}));
23+
expectError(subprocess.transformStream({to: 'fdNotANumber'}));
24+
25+
subprocess.transformStream({binary: false});
26+
expectError(subprocess.transformStream({binary: 'false'}));
27+
28+
subprocess.transformStream({preserveNewlines: false});
29+
expectError(subprocess.transformStream({preserveNewlines: 'false'}));
30+
31+
expectError(subprocess.transformStream('stdout'));
32+
expectError(subprocess.transformStream({other: 'stdout'}));

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

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
import type {WritableStream} from 'node:stream/web';
2+
import {expectType, expectError} from 'tsd';
3+
import {execa} from '../../index.js';
4+
5+
const subprocess = execa('unicorns');
6+
7+
expectType<WritableStream>(subprocess.writableStream());
8+
9+
subprocess.writableStream({to: 'stdin'});
10+
subprocess.writableStream({to: 'fd3'});
11+
expectError(subprocess.writableStream({to: 'fd3' as string}));
12+
expectError(subprocess.writableStream({to: 'stdout'}));
13+
expectError(subprocess.writableStream({to: 'fd'}));
14+
expectError(subprocess.writableStream({to: 'fdNotANumber'}));
15+
expectError(subprocess.writableStream({from: 'stdout'}));
16+
17+
expectError(subprocess.writableStream({binary: false}));
18+
19+
expectError(subprocess.writableStream({preserveNewlines: false}));
20+
21+
expectError(subprocess.writableStream('stdin'));
22+
expectError(subprocess.writableStream({other: 'stdin'}));
23+
24+
const inputPipeSubprocess = execa('unicorns', {stdio: ['pipe', 'pipe', 'pipe', {value: 'pipe', input: true}]});
25+
inputPipeSubprocess.writableStream({to: 'fd3'});

‎test-d/pipe.test-d.ts‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import {createWriteStream} from 'node:fs';
22
import type {Readable} from 'node:stream';
3+
import type {ReadableStream} from 'node:stream/web';
34
import {expectType, expectNotType, expectError} from 'tsd';
45
import {
56
execa,
@@ -269,6 +270,7 @@ for await (const pipeLine of subprocess.pipe(bufferSubprocess)) {
269270
}
270271

271272
expectType<Readable>(subprocess.pipe`stdin`.readable());
273+
expectType<ReadableStream>(subprocess.pipe`stdin`.readableStream());
272274

273275
expectType<Readable>(subprocess.pipe({all: true})`stdin`.all);
274276
expectType<undefined>(subprocess.pipe`stdin`.all);
@@ -279,6 +281,8 @@ expectType<Promise<void>>(ipcPipeResult.sendMessage('message'));
279281
expectType<Promise<Message<'advanced'>>>(ipcPipeResult.getOneMessage());
280282
expectType<AsyncIterableIterator<Message<'advanced'>>>(ipcPipeResult.getEachMessage());
281283

282-
// `writable()` and `duplex()` write to the destination's `stdin`, which is already piped from the source, so they are not forwarded.
284+
// `writable()`, `duplex()`, `writableStream()` and `transformStream()` write to the destination's `stdin`, which is already piped from the source, so they are not forwarded.
283285
expectError(subprocess.pipe`stdin`.writable());
284286
expectError(subprocess.pipe`stdin`.duplex());
287+
expectError(subprocess.pipe`stdin`.writableStream());
288+
expectError(subprocess.pipe`stdin`.transformStream());

0 commit comments

Comments
 (0)