From 7206a786463b23ecbfb6cd1e96dbd8f683fae4e1 Mon Sep 17 00:00:00 2001 From: Sindre Sorhus Date: Sun, 5 Jul 2026 17:35:42 +0200 Subject: [PATCH] Expose the destination subprocess methods on `.pipe()` The return value of `subprocess.pipe()` now forwards the destination subprocess' iteration (`Symbol.asyncIterator`, `.iterable()`), stream conversion (`.all`, `.readable()`) and IPC (`.sendMessage()`, `.getOneMessage()`, `.getEachMessage()`) methods, so the piped output can be consumed directly. `.writable()` and `.duplex()` are not forwarded, since they write to the destination's `stdin`, which is already piped from the source. Fixes #1211 --- docs/api.md | 2 + docs/pipe.md | 10 + lib/ipc/get-each.js | 41 ++- lib/ipc/get-one.js | 27 +- lib/pipe/setup.js | 178 ++++++++++- test-d/pipe.test-d.ts | 30 ++ test/fixtures/stdin-distinct-both.js | 7 + test/pipe/methods.js | 435 +++++++++++++++++++++++++++ test/pipe/sequence.js | 80 ++++- types/pipe.d.ts | 24 +- types/subprocess/subprocess.d.ts | 54 ++-- 11 files changed, 829 insertions(+), 59 deletions(-) create mode 100755 test/fixtures/stdin-distinct-both.js create mode 100644 test/pipe/methods.js diff --git a/docs/api.md b/docs/api.md index 4b1affa92e..17e4b2dc4b 100644 --- a/docs/api.md +++ b/docs/api.md @@ -253,6 +253,8 @@ _Returns_: [`Promise`](#result) This follows the same syntax as [`execa(file, arguments?, options?)`](#execafile-arguments-options) except both [regular options](#options-1) and [pipe-specific options](#pipeoptions) can be specified. +Like a subprocess, the return value can be [iterated](lines.md#progressive-splitting), [converted to a stream](streams.md#converting-a-subprocess-to-a-stream), or used for [IPC](ipc.md) with the destination subprocess. + [More info.](pipe.md#array-syntax) ### subprocess.pipe\`command\` diff --git a/docs/pipe.md b/docs/pipe.md index 2d80a6a9a5..4a8205edf0 100644 --- a/docs/pipe.md +++ b/docs/pipe.md @@ -66,6 +66,16 @@ const sourceResult = destinationResult.pipedFrom[0]; console.log(sourceResult.stdout); // Full output of `npm run build` ``` +## Iterate, stream and IPC + +Just like a regular subprocess, the value returned by `subprocess.pipe()` can be [iterated](lines.md#progressive-splitting) over its output lines, [converted to a readable stream](streams.md#converting-a-subprocess-to-a-stream), or used to exchange [IPC messages](ipc.md) with the destination subprocess. + +```js +for await (const line of execa`npm run build`.pipe`sort`) { + console.log(line); +} +``` + ## Errors When either subprocess fails, `subprocess.pipe()` is rejected with that subprocess' error. If the destination subprocess fails, [`error.pipedFrom`](api.md#resultpipedfrom) includes the source subprocess' result, which is useful for debugging. diff --git a/lib/ipc/get-each.js b/lib/ipc/get-each.js index f134fc12cd..ffcd6138b1 100644 --- a/lib/ipc/get-each.js +++ b/lib/ipc/get-each.js @@ -3,18 +3,24 @@ import {validateIpcMethod, disconnect, getStrictResponseError} from './validatio import {getIpcEmitter, isConnected} from './forward.js'; import {addReference, removeReference} from './reference.js'; +export const internalGetEachMessageOptions = Symbol('internalGetEachMessageOptions'); + // Like `[sub]process.on('message')` but promise-based -export const getEachMessage = ({anyProcess, channel, isSubprocess, ipc}, {reference = true} = {}) => loopOnMessages({ - anyProcess, - channel, - isSubprocess, - ipc, - shouldAwait: !isSubprocess, - reference, -}); +export const getEachMessage = (subprocessInfo, options = {}) => { + const {reference = true} = options; + // Internal callers can stop the IPC listener without exposing cancellation or pipe-awaiting options as public API. + const {signal, shouldAwait = !subprocessInfo.isSubprocess} = options[internalGetEachMessageOptions] ?? {}; + + return loopOnMessages({ + ...subprocessInfo, + shouldAwait, + reference, + signal, + }); +}; // Same but used internally -export const loopOnMessages = ({anyProcess, channel, isSubprocess, ipc, shouldAwait, reference}) => { +export const loopOnMessages = ({anyProcess, channel, isSubprocess, ipc, shouldAwait, reference, signal}) => { validateIpcMethod({ methodName: 'getEachMessage', isSubprocess, @@ -26,6 +32,7 @@ export const loopOnMessages = ({anyProcess, channel, isSubprocess, ipc, shouldAw const ipcEmitter = getIpcEmitter(anyProcess, channel, isSubprocess); const controller = new AbortController(); const state = {}; + stopOnAbort(signal, controller); stopOnDisconnect(anyProcess, ipcEmitter, controller); abortOnStrictError({ ipcEmitter, @@ -45,6 +52,22 @@ export const loopOnMessages = ({anyProcess, channel, isSubprocess, ipc, shouldAw }); }; +const stopOnAbort = (signal, controller) => { + if (signal === undefined) { + return; + } + + if (signal.aborted) { + controller.abort(); + return; + } + + // Reuse the iterator controller for listener cleanup when iteration ends before the internal signal aborts. + signal.addEventListener('abort', () => { + controller.abort(); + }, {once: true, signal: controller.signal}); +}; + const stopOnDisconnect = async (anyProcess, ipcEmitter, controller) => { try { await once(ipcEmitter, 'disconnect', {signal: controller.signal}); diff --git a/lib/ipc/get-one.js b/lib/ipc/get-one.js index 976a8fe191..8082517790 100644 --- a/lib/ipc/get-one.js +++ b/lib/ipc/get-one.js @@ -8,8 +8,13 @@ import { import {getIpcEmitter, isConnected} from './forward.js'; import {addReference, removeReference} from './reference.js'; +export const internalGetOneMessageOptions = Symbol('internalGetOneMessageOptions'); + // Like `[sub]process.once('message')` but promise-based -export const getOneMessage = ({anyProcess, channel, isSubprocess, ipc}, {reference = true, filter} = {}) => { +export const getOneMessage = ({anyProcess, channel, isSubprocess, ipc}, options = {}) => { + const {reference = true, filter} = options; + const {signal} = options[internalGetOneMessageOptions] ?? {}; + validateIpcMethod({ methodName: 'getOneMessage', isSubprocess, @@ -23,13 +28,16 @@ export const getOneMessage = ({anyProcess, channel, isSubprocess, ipc}, {referen isSubprocess, filter, reference, + signal, }); }; -const getOneMessageAsync = async ({anyProcess, channel, isSubprocess, filter, reference}) => { +const getOneMessageAsync = async ({anyProcess, channel, isSubprocess, filter, reference, signal}) => { addReference(channel, reference); const ipcEmitter = getIpcEmitter(anyProcess, channel, isSubprocess); const controller = new AbortController(); + stopOnAbort(signal, controller); + try { return await Promise.race([ getMessage(ipcEmitter, filter, controller), @@ -45,6 +53,21 @@ const getOneMessageAsync = async ({anyProcess, channel, isSubprocess, filter, re } }; +const stopOnAbort = (signal, controller) => { + if (signal === undefined) { + return; + } + + if (signal.aborted) { + controller.abort(); + return; + } + + signal.addEventListener('abort', () => { + controller.abort(); + }, {once: true, signal: controller.signal}); +}; + const getMessage = async (ipcEmitter, filter, {signal}) => { if (filter === undefined) { const [message] = await once(ipcEmitter, 'message', {signal}); diff --git a/lib/pipe/setup.js b/lib/pipe/setup.js index bf1a87b503..6160f9267c 100644 --- a/lib/pipe/setup.js +++ b/lib/pipe/setup.js @@ -1,4 +1,10 @@ import isPlainObject from 'is-plain-obj'; +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 {internalGetOneMessageOptions} from '../ipc/get-one.js'; +import {internalGetEachMessageOptions} from '../ipc/get-each.js'; import {normalizePipeArguments} from './pipe-arguments.js'; import {handlePipeArgumentsError} from './throw.js'; import {waitForBothSubprocesses} from './sequence.js'; @@ -15,16 +21,162 @@ export const pipeToSubprocess = (sourceInfo, ...pipeArguments) => { } const {destination, ...normalizedInfo} = normalizePipeArguments(sourceInfo, ...pipeArguments); - const promise = handlePipePromise({...normalizedInfo, destination}); + const pipeFailureController = new AbortController(); + const promise = handlePipePromise({...normalizedInfo, destination, pipeFailureController}); promise.pipe = pipeToSubprocess.bind(undefined, { ...sourceInfo, source: destination, sourcePromise: promise, boundOptions: {}, }); + forwardDestinationMethods(promise, destination, pipeFailureController.signal); return promise; }; +/* +The return value of `.pipe()` exposes the destination subprocess' output, but its iteration and stream conversion methods must await the whole pipe so source failures are propagated too. The destination is `undefined` when `.pipe()` was passed invalid arguments, in which case the promise rejects and there is nothing to forward. +*/ +const forwardDestinationMethods = (promise, destination, pipeFailureSignal) => { + if (destination === undefined) { + return; + } + + forwardReadableMethods(promise, destination); + forwardIpcMethods(promise, destination, pipeFailureSignal); +}; + +const forwardReadableMethods = (promise, destination) => { + const subprocessOptions = SUBPROCESS_OPTIONS.get(destination); + SUBPROCESS_OPTIONS.set(promise, subprocessOptions); + promise.stdio = destination.stdio; + promise.all = destination.all; + + const {options: {encoding}} = subprocessOptions; + const concurrentStreams = initializeConcurrentStreams(); + promise[Symbol.asyncIterator] = createIterable.bind(undefined, promise, encoding, {}); + promise.iterable = createIterable.bind(undefined, promise, encoding); + promise.readable = createPipeReadable.bind(undefined, promise, { + subprocess: promise, + concurrentStreams, + encoding, + }); + forwardAll(promise, destination); +}; + +const forwardAll = (promise, destination) => { + if (destination.all === undefined) { + promise.all = undefined; + return; + } + + Object.defineProperty(promise, 'all', { + get() { + setAllProperty(promise, destination.all); + const all = promise.readable({from: 'all'}); + setAllProperty(promise, all); + return all; + }, + enumerable: true, + configurable: true, + }); +}; + +const setAllProperty = (promise, value) => { + Object.defineProperty(promise, 'all', { + value, + writable: true, + enumerable: true, + configurable: true, + }); +}; + +const createPipeReadable = (promise, readableOptions, ...arguments_) => { + const readable = createReadable(readableOptions, ...arguments_); + destroyOnPipeFailure(promise, readable); + return readable; +}; + +const destroyOnPipeFailure = async (promise, readable) => { + try { + await promise; + } catch (error) { + readable.destroy(error); + } +}; + +const forwardIpcMethods = (promise, destination, pipeFailureSignal) => { + promise.sendMessage = destination.sendMessage; + promise.getOneMessage = getOnePipeMessage.bind(undefined, destination, pipeFailureSignal); + promise.getEachMessage = getEachPipeMessage.bind(undefined, promise, destination, pipeFailureSignal); +}; + +const getOnePipeMessage = (destination, pipeFailureSignal, ...arguments_) => { + const controller = new AbortController(); + const messagePromise = destination.getOneMessage(...addPipeOptions(arguments_, controller.signal, internalGetOneMessageOptions)); + return waitForOnePipeMessage(pipeFailureSignal, messagePromise, controller); +}; + +const waitForOnePipeMessage = async (pipeFailureSignal, messagePromise, controller) => { + try { + return await Promise.race([messagePromise, getSignalRejection(pipeFailureSignal, controller.signal)]); + } finally { + controller.abort(); + } +}; + +const getSignalRejection = (signal, listenerSignal) => new Promise((_, reject) => { + if (signal.aborted) { + reject(signal.reason); + return; + } + + signal.addEventListener('abort', () => { + reject(signal.reason); + }, {once: true, signal: listenerSignal}); +}); + +const getEachPipeMessage = (promise, destination, pipeFailureSignal, ...arguments_) => { + const controller = new AbortController(); + // Create the destination iterator before awaiting the pipe so option validation stays synchronous. + const iterator = destination.getEachMessage(...addPipeOptions(arguments_, controller.signal, internalGetEachMessageOptions)); + abortOnSignal(pipeFailureSignal, controller); + return iterateOnPipeMessages(promise, iterator, controller); +}; + +const iterateOnPipeMessages = async function * (promise, iterator, controller) { + try { + yield * iterator; + } finally { + controller.abort(); + await promise; + } +}; + +const addPipeOptions = (arguments_, signal, internalOptionsSymbol) => { + if (arguments_[0] === null) { + // Preserve the public validation error from `getEachMessage(null)` instead of masking it with a pipe failure. + return arguments_; + } + + const [options] = arguments_; + // The returned pipe promise is awaited by the forwarded IPC iterator, so the destination IPC iterator must not also await the destination subprocess on close. + return [{...options, [internalOptionsSymbol]: {signal, shouldAwait: false}}]; +}; + +const abortOnSignal = (signal, controller) => { + if (signal.aborted) { + controller.abort(); + return; + } + + // Interrupt pending IPC reads when the pipe rejects, including `unpipeSignal` cancellation while the destination keeps running. + signal.addEventListener('abort', () => { + controller.abort(); + }, {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. + // Asynchronous logic when piping subprocesses const handlePipePromise = async ({ sourcePromise, @@ -37,19 +189,20 @@ const handlePipePromise = async ({ unpipeSignal, fileDescriptors, startTime, + pipeFailureController, }) => { - const subprocessPromises = getSubprocessPromises(sourcePromise, destination); - handlePipeArgumentsError({ - sourceStream, - sourceError, - destinationStream, - destinationError, - fileDescriptors, - sourceOptions, - startTime, - }); const maxListenersController = new AbortController(); try { + const subprocessPromises = getSubprocessPromises(sourcePromise, destination); + handlePipeArgumentsError({ + sourceStream, + sourceError, + destinationStream, + destinationError, + fileDescriptors, + sourceOptions, + startTime, + }); const mergedStream = pipeSubprocessStream(sourceStream, destinationStream, maxListenersController); return await Promise.race([ waitForBothSubprocesses(subprocessPromises), @@ -61,6 +214,9 @@ const handlePipePromise = async ({ startTime, }), ]); + } catch (error) { + pipeFailureController.abort(error); + throw error; } finally { maxListenersController.abort(); } diff --git a/test-d/pipe.test-d.ts b/test-d/pipe.test-d.ts index 1ce42a7ecc..7d39d835f9 100644 --- a/test-d/pipe.test-d.ts +++ b/test-d/pipe.test-d.ts @@ -1,10 +1,12 @@ import {createWriteStream} from 'node:fs'; +import type {Readable} from 'node:stream'; import {expectType, expectNotType, expectError} from 'tsd'; import { execa, execaSync, $, type Result, + type Message, } from '../index.js'; const fileUrl = new URL('file:///test'); @@ -252,3 +254,31 @@ expectType(ignoreShortcutScriptPipeResult.stdout); const unicornsResult = execaSync('unicorns'); expectError(unicornsResult.pipe); + +// The `.pipe()` return value forwards the destination subprocess' iteration, stream conversion and IPC methods. +for await (const pipeLine of subprocess.pipe`stdin`) { + expectType(pipeLine); +} + +for await (const pipeLine of subprocess.pipe`stdin`.iterable()) { + expectType(pipeLine); +} + +for await (const pipeLine of subprocess.pipe(bufferSubprocess)) { + expectType(pipeLine); +} + +expectType(subprocess.pipe`stdin`.readable()); + +expectType(subprocess.pipe({all: true})`stdin`.all); +expectType(subprocess.pipe`stdin`.all); + +expectType(subprocess.pipe`stdin`.sendMessage); +const ipcPipeResult = subprocess.pipe({ipc: true})`stdin`; +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. +expectError(subprocess.pipe`stdin`.writable()); +expectError(subprocess.pipe`stdin`.duplex()); diff --git a/test/fixtures/stdin-distinct-both.js b/test/fixtures/stdin-distinct-both.js new file mode 100755 index 0000000000..409ee14ac6 --- /dev/null +++ b/test/fixtures/stdin-distinct-both.js @@ -0,0 +1,7 @@ +#!/usr/bin/env node +import process from 'node:process'; +import {text} from 'node:stream/consumers'; + +const input = await text(process.stdin); +process.stdout.write(`${input}:stdout`); +process.stderr.write(`${input}:stderr`); diff --git a/test/pipe/methods.js b/test/pipe/methods.js new file mode 100644 index 0000000000..e5825d2736 --- /dev/null +++ b/test/pipe/methods.js @@ -0,0 +1,435 @@ +import {setTimeout} from 'node:timers/promises'; +import {Readable} from 'node:stream'; +import {text} from 'node:stream/consumers'; +import test from 'ava'; +import {execa} from '../../index.js'; +import {setFixtureDirectory} from '../helpers/fixtures-directory.js'; +import {simpleFull, noNewlinesFull, noNewlinesChunks} from '../helpers/lines.js'; +import {foobarString, foobarArray} from '../helpers/input.js'; + +setFixtureDirectory(); + +const timeoutSymbol = Symbol('timeout'); + +// `stdin.js` pipes its stdin to its stdout, so the piped output equals the source output +const pipeSimple = () => execa('noop-fd.js', ['1', simpleFull]).pipe('stdin.js'); +const pipeAll = () => execa('noop-fd.js', ['1', simpleFull]).pipe('stdin.js', {all: true}); +const pipeDistinctBoth = () => execa('noop-fd.js', ['1', simpleFull]).pipe('stdin-distinct-both.js'); + +const assertSettles = async (t, promise, timeout = 1000) => { + const result = await Promise.race([ + promise, + // Several pipe cancellation tests are specifically checking that a pending reader does not hang. + setTimeout(timeout, timeoutSymbol, {ref: false}), + ]); + + t.not(result, timeoutSymbol); + return result; +}; + +const cleanupSubprocesses = async (...subprocesses) => { + for (const subprocess of subprocesses) { + subprocess.kill(); + } + + await Promise.allSettled(subprocesses); +}; + +const getPipeMessages = async piped => { + const messages = []; + for await (const message of piped.getEachMessage()) { + messages.push(message); + } + + return messages; +}; + +test('The .pipe() return value can be iterated', async t => { + const lines = []; + for await (const line of pipeSimple()) { + lines.push(line); + } + + t.deepEqual(lines, noNewlinesChunks); +}); + +test('The .pipe() return value has an .iterable() method', async t => { + const lines = []; + for await (const line of pipeSimple().iterable()) { + lines.push(line); + } + + t.deepEqual(lines, noNewlinesChunks); +}); + +test('The .pipe() return value has a .readable() method', async t => { + t.is(await text(pipeSimple().readable()), simpleFull); +}); + +test('The .pipe() return value .readable() can be called multiple times', async t => { + const piped = pipeSimple(); + const [output, secondOutput] = await Promise.all([ + text(piped.readable()), + text(piped.readable()), + ]); + + t.is(output, simpleFull); + t.is(secondOutput, simpleFull); +}); + +test('The .pipe() return value .readable() keeps conversion options', async t => { + const chunks = []; + for await (const chunk of pipeSimple().readable({binary: false, preserveNewlines: false})) { + chunks.push(chunk); + } + + t.deepEqual(chunks, noNewlinesChunks); + t.is(chunks.join(''), noNewlinesFull); +}); + +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); + await piped; +}); + +test('The .pipe() return value does not have .all unless destination uses the "all" option', async t => { + const piped = pipeSimple(); + const descriptor = Object.getOwnPropertyDescriptor(piped, 'all'); + + t.is(piped.all, undefined); + t.is(descriptor.value, undefined); + t.is(descriptor.get, undefined); + await piped; +}); + +test('The .pipe() return value has an .all property', async t => { + const piped = pipeAll(); + t.true(piped.all instanceof Readable); + t.is(await text(piped.all), simpleFull); + await piped; +}); + +test('The .pipe() return value .all is created lazily and cached', async t => { + const piped = pipeAll(); + const descriptor = Object.getOwnPropertyDescriptor(piped, 'all'); + + t.is(typeof descriptor.get, 'function'); + const {all} = piped; + t.true(all instanceof Readable); + t.is(piped.all, all); + t.is(Object.getOwnPropertyDescriptor(piped, 'all').get, undefined); + t.is(await text(all), simpleFull); + await piped; +}); + +test('The .pipe() return value .readable() can read destination stderr', async t => { + const piped = pipeDistinctBoth(); + + t.is(await text(piped.readable({from: 'stderr'})), `${simpleFull}:stderr`); + await piped; +}); + +test('The .pipe() return value .iterable() can read destination stderr', async t => { + const piped = pipeDistinctBoth(); + const lines = []; + for await (const line of piped.iterable({from: 'stderr'})) { + lines.push(line); + } + + t.deepEqual(lines, ['aaa', 'bbb', 'ccc:stderr']); + await piped; +}); + +test('The .pipe() return value has IPC methods', async t => { + const piped = execa('empty.js').pipe('ipc-send-twice.js', {ipc: true}); + t.deepEqual(await getPipeMessages(piped), foobarArray); + const {ipcOutput} = await piped; + t.deepEqual(ipcOutput, foobarArray); +}); + +test('The .pipe() return value .getEachMessage() can be called twice at the same time', async t => { + const piped = execa('empty.js').pipe('ipc-send-twice.js', {ipc: true}); + + t.deepEqual( + await Promise.all([getPipeMessages(piped), getPipeMessages(piped)]), + [foobarArray, foobarArray], + ); + + const {ipcOutput} = await piped; + t.deepEqual(ipcOutput, foobarArray); +}); + +test('The .pipe() return value .getEachMessage() preserves setup errors', async t => { + const piped = execa('empty.js').pipe('ipc-send-twice.js', {ipc: true}); + + t.throws(() => { + piped.getEachMessage(null); + }, {instanceOf: TypeError}); + + await piped; +}); + +test('The .pipe() return value .getEachMessage() waits for source failure', async t => { + const piped = execa('fail.js').pipe('ipc-send-twice.js', {ipc: true}); + + await t.throwsAsync(async () => { + for await (const message of piped.getEachMessage()) { + t.true(foobarArray.includes(message)); + } + }, {message: /Command failed with exit code 2/}); +}); + +test('The .pipe() return value .getEachMessage() waits for pipe cancellation', async t => { + const abortController = new AbortController(); + const source = execa('noop-repeat.js'); + const destination = execa('ipc-send-forever.js', {ipc: true}); + t.teardown(async () => { + await cleanupSubprocesses(source, destination); + }); + + const piped = source.pipe(destination, {unpipeSignal: abortController.signal}); + const iterator = piped.getEachMessage(); + t.deepEqual(await iterator.next(), {done: false, value: foobarString}); + + const readerPromise = t.throwsAsync(iterator.next(), {message: /Pipe canceled/}); + abortController.abort(); + await assertSettles(t, readerPromise); +}); + +test('The .pipe() return value .getEachMessage() cancels every pending reader', async t => { + const abortController = new AbortController(); + const source = execa('noop-repeat.js'); + const destination = execa('ipc-send-forever.js', {ipc: true}); + t.teardown(async () => { + await cleanupSubprocesses(source, destination); + }); + + const piped = source.pipe(destination, {unpipeSignal: abortController.signal}); + const iterators = [ + piped.getEachMessage(), + piped.getEachMessage(), + ]; + + t.deepEqual( + await Promise.all(iterators.map(async iterator => iterator.next())), + [ + {done: false, value: foobarString}, + {done: false, value: foobarString}, + ], + ); + + const readerPromises = iterators.map(iterator => t.throwsAsync(iterator.next(), {message: /Pipe canceled/})); + abortController.abort(); + await Promise.all(readerPromises.map(async readerPromise => assertSettles(t, readerPromise))); +}); + +const assertCanceledPipeMessageReader = async (t, getReader, abortBeforePipe = false) => { + const abortController = new AbortController(); + if (abortBeforePipe) { + abortController.abort(); + } + + const source = execa('noop-repeat.js'); + const destination = execa('ipc-get.js', {ipc: true}); + t.teardown(async () => { + await cleanupSubprocesses(source, destination); + }); + + const piped = source.pipe(destination, {unpipeSignal: abortController.signal}); + const readerPromise = t.throwsAsync(getReader(piped), {message: /Pipe canceled/}); + + if (!abortBeforePipe) { + abortController.abort(); + } + + await assertSettles(t, readerPromise); + await t.throwsAsync(piped, {message: /Pipe canceled/}); +}; + +test('The .pipe() return value .getEachMessage() waits for pipe cancellation before IPC messages', async t => { + await assertCanceledPipeMessageReader(t, piped => piped.getEachMessage().next()); +}); + +test('The .pipe() return value .getEachMessage() waits for already canceled pipes', async t => { + await assertCanceledPipeMessageReader(t, piped => piped.getEachMessage().next(), true); +}); + +test('The .pipe() return value .getOneMessage() waits for pipe cancellation before IPC messages', async t => { + await assertCanceledPipeMessageReader(t, piped => piped.getOneMessage()); +}); + +test('The .pipe() return value .getOneMessage() waits for already canceled pipes', async t => { + await assertCanceledPipeMessageReader(t, piped => piped.getOneMessage(), true); +}); + +test('The .pipe() return value .getEachMessage() waits for the pipe when the iterator is returned', async t => { + const source = execa('empty.js'); + const destination = execa('ipc-send-forever.js', {ipc: true}); + t.teardown(async () => { + await cleanupSubprocesses(source, destination); + }); + + const piped = source.pipe(destination); + const iterator = piped.getEachMessage(); + + t.deepEqual(await iterator.next(), {done: false, value: foobarString}); + const returnPromise = t.throwsAsync(iterator.return()); + destination.kill(); + const error = await returnPromise; + t.true(error.isTerminated); + await Promise.allSettled([source, destination, piped]); +}); + +test('The .pipe() return value .getEachMessage() propagates source failure when the iterator is returned', async t => { + const source = execa('ipc-echo-fail.js', {ipc: true}); + const piped = source.pipe('ipc-send-wait-print.js', {ipc: true}); + const iterator = piped.getEachMessage(); + + t.deepEqual(await iterator.next(), {done: false, value: foobarString}); + const returnPromise = t.throwsAsync(iterator.return(), {message: /Command failed with exit code 1/}); + await source.sendMessage(foobarString); + await returnPromise; +}); + +test('The .pipe() return value .getEachMessage() propagates destination failure', async t => { + const piped = execa('empty.js').pipe('ipc-send-fail.js', {ipc: true}); + + await t.throwsAsync(async () => { + for await (const message of piped.getEachMessage()) { + t.is(message, foobarString); + } + }, {message: /Command failed with exit code 1/}); + + const {ipcOutput} = await t.throwsAsync(piped); + t.deepEqual(ipcOutput, [foobarString]); +}); + +test('The .pipe() return value .getEachMessage() waits for the pipe when the consumer throws', async t => { + const piped = execa('empty.js').pipe('ipc-send-wait-print.js', {ipc: true}); + const cause = new Error(foobarString); + + t.is(await t.throwsAsync(async () => { + // eslint-disable-next-line no-unreachable-loop + for await (const message of piped.getEachMessage()) { + t.is(message, foobarString); + throw cause; + } + }), cause); + + const {ipcOutput, stdout} = await piped; + t.deepEqual(ipcOutput, [foobarString]); + t.is(stdout, '.'); +}); + +test('The .pipe() return value .getEachMessage() validates IPC immediately', async t => { + const source = execa('stdin.js'); + const destination = execa('stdin.js'); + t.teardown(async () => { + await cleanupSubprocesses(source, destination); + }); + + const piped = source.pipe(destination); + const {message} = t.throws(() => piped.getEachMessage()); + + t.true(message.includes('subprocess.getEachMessage() can only be used')); + source.kill(); + destination.kill(); + await Promise.allSettled([source, destination, piped]); +}); + +test('The .pipe() return value can exchange IPC messages', async t => { + const piped = execa('empty.js').pipe('ipc-echo.js', {ipc: true}); + await piped.sendMessage(foobarString); + t.is(await piped.getOneMessage(), foobarString); + await piped; +}); + +test('A chained .pipe() return value can be iterated', async t => { + const lines = []; + for await (const line of execa('noop-fd.js', ['1', simpleFull]).pipe('stdin.js').pipe('stdin.js')) { + lines.push(line); + } + + t.deepEqual(lines, noNewlinesChunks); +}); + +test('The .pipe() return value still awaits both subprocesses', async t => { + await t.throwsAsync(execa('fail.js').pipe('stdin.js'), {message: /Command failed with exit code 2/}); +}); + +test('The .pipe() return value iteration waits for source failure', async t => { + const piped = execa('fail.js').pipe('stdin.js'); + + await t.throwsAsync(async () => { + for await (const line of piped) { + t.fail(`Unexpected line: ${line}`); + } + }, {message: /Command failed with exit code 2/}); +}); + +test('The .pipe() return value .readable() waits for source failure', async t => { + const piped = execa('fail.js').pipe('stdin.js'); + + await t.throwsAsync(text(piped.readable()), {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}); + + await t.throwsAsync(text(piped.all), {message: /Command failed with exit code 2/}); +}); + +const assertCanceledPipeReader = async (t, getReader, abortBeforePipe = false) => { + const abortController = new AbortController(); + if (abortBeforePipe) { + abortController.abort(); + } + + const source = execa('stdin.js'); + const destination = execa('stdin.js', {all: true}); + t.teardown(async () => { + await cleanupSubprocesses(source, destination); + }); + + const piped = source.pipe(destination, {unpipeSignal: abortController.signal}); + const readerPromise = t.throwsAsync(getReader(piped), {message: /Pipe canceled/}); + + if (!abortBeforePipe) { + abortController.abort(); + } + + await assertSettles(t, readerPromise, 200); +}; + +test('The .pipe() return value .readable() waits for pipe cancellation', async t => { + await assertCanceledPipeReader(t, piped => text(piped.readable())); +}); + +test('The .pipe() return value .all waits for pipe cancellation', async t => { + await assertCanceledPipeReader(t, piped => text(piped.all)); +}); + +test('The .pipe() return value .readable() waits for already canceled pipes', async t => { + await assertCanceledPipeReader(t, piped => text(piped.readable()), true); +}); + +test('The .pipe() return value .all waits for already canceled pipes', async t => { + await assertCanceledPipeReader(t, piped => text(piped.all), true); +}); + +test('Multiple sources piped to one destination each expose the destination methods', async t => { + const destination = execa('stdin.js'); + const piped = [ + execa('noop-fd.js', ['1', foobarString]).pipe(destination), + execa('noop-fd.js', ['1', foobarString]).pipe(destination), + ]; + + for (const pipePromise of piped) { + t.is(typeof pipePromise.iterable, 'function'); + t.is(typeof pipePromise[Symbol.asyncIterator], 'function'); + } + + await Promise.all(piped); +}); diff --git a/test/pipe/sequence.js b/test/pipe/sequence.js index fef27e8487..3266ea15b2 100644 --- a/test/pipe/sequence.js +++ b/test/pipe/sequence.js @@ -1,6 +1,7 @@ import {once} from 'node:events'; import process from 'node:process'; import {PassThrough} from 'node:stream'; +import {setTimeout} from 'node:timers/promises'; import test from 'ava'; import {execa} from '../../index.js'; import {setFixtureDirectory} from '../helpers/fixtures-directory.js'; @@ -11,6 +12,38 @@ import {prematureClose} from '../helpers/stdio.js'; setFixtureDirectory(); const isLinux = process.platform === 'linux'; +const timeoutSymbol = Symbol('timeout'); + +const assertUnhandledRejection = async (t, pipePromise, unhandledRejectionPromise) => { + const result = await Promise.race([ + unhandledRejectionPromise, + setTimeout(500, timeoutSymbol), + ]); + + t.not(result, timeoutSymbol); + + if (result === timeoutSymbol) { + return; + } + + const [reason, unhandledPromise] = result; + t.is(unhandledPromise, pipePromise); + t.regex(reason.message, /Command failed with exit code 2/); +}; + +const assertUnhandledPipePromise = async (t, {destinationOptions = {}, inspectPipePromise} = {}) => { + const source = execa('fail.js'); + const destination = execa('fail.js', destinationOptions); + const pipePromise = source.pipe(destination); + const unhandledRejectionPromise = once(process, 'unhandledRejection'); + await inspectPipePromise?.(pipePromise); + await assertUnhandledRejection(t, pipePromise, unhandledRejectionPromise); + await Promise.all([ + t.throwsAsync(source), + t.throwsAsync(destination), + ]); + await t.throwsAsync(pipePromise); +}; test('Source stream abort -> destination success', async t => { const source = execa('noop-repeat.js'); @@ -267,15 +300,48 @@ test('Does not need to await individual promises', async t => { await t.throwsAsync(source.pipe(destination)); }); -test('Need to await .pipe() return value', async t => { +test.serial('Need to await .pipe() return value', async t => { + await assertUnhandledPipePromise(t); +}); + +test.serial('Need to await .pipe() return value, "all" option', async t => { + await assertUnhandledPipePromise(t, {destinationOptions: {all: true}}); +}); + +test.serial('Need to await .pipe() return value after inspecting lazy .all', async t => { + await assertUnhandledPipePromise(t, { + destinationOptions: {all: true}, + inspectPipePromise(pipePromise) { + const descriptor = Object.getOwnPropertyDescriptor(pipePromise, 'all'); + t.is(typeof descriptor.get, 'function'); + }, + }); +}); + +test.serial('Need to await .pipe() return value after getOneMessage()', async t => { const source = execa('fail.js'); - const destination = execa('fail.js'); + const destination = execa('ipc-send-twice.js', {ipc: true}); const pipePromise = source.pipe(destination); - await Promise.all([ - once(process, 'unhandledRejection'), - t.throwsAsync(source), - t.throwsAsync(destination), - ]); + const unhandledRejectionPromise = once(process, 'unhandledRejection'); + + t.is(await pipePromise.getOneMessage(), 'foo'); + await assertUnhandledRejection(t, pipePromise, unhandledRejectionPromise); + await t.throwsAsync(source); + await destination; + await t.throwsAsync(pipePromise); +}); + +test.serial('Need to await .pipe() return value after paused getEachMessage()', async t => { + const source = execa('fail.js'); + const destination = execa('ipc-send-twice.js', {ipc: true}); + const pipePromise = source.pipe(destination); + const unhandledRejectionPromise = once(process, 'unhandledRejection'); + const iterator = pipePromise.getEachMessage(); + + t.deepEqual(await iterator.next(), {done: false, value: 'foo'}); + await assertUnhandledRejection(t, pipePromise, unhandledRejectionPromise); + await t.throwsAsync(source); + await destination; await t.throwsAsync(pipePromise); }); diff --git a/types/pipe.d.ts b/types/pipe.d.ts index 6ad078a6f6..a742366060 100644 --- a/types/pipe.d.ts +++ b/types/pipe.d.ts @@ -1,7 +1,8 @@ import type {Options} from './arguments/options.js'; import type {Result} from './return/result.js'; import type {FromOption, ToOption} from './arguments/fd-options.js'; -import type {ResultPromise} from './subprocess/subprocess.js'; +import type {ResultPromise, SubprocessResultMethods} from './subprocess/subprocess.js'; +import type {IpcMethods, HasIpc} from './ipc.js'; import type {TemplateExpression} from './methods/template.js'; // `subprocess.pipe()` options @@ -24,37 +25,48 @@ type PipeOptions = { readonly unpipeSignal?: AbortSignal; }; +// Methods forwarded from the destination subprocess to the return value of `subprocess.pipe()`, so its output can be iterated, converted to a stream, or used for IPC. +type PipeResultMethods = + & SubprocessResultMethods + & IpcMethods, OptionsType['serialization']>; + +// Same as `PipeResultMethods`, but when the destination is another `execa()` call, so its own option types are kept. +// The base `Options` is only used to compute the property names, which do not depend on the specific options. +type PipeResultMethodsFrom = Pick>; + // `subprocess.pipe()` export type PipableSubprocess = { /** [Pipe](https://nodejs.org/api/stream.html#readablepipedestination-options) the subprocess' `stdout` to a second Execa subprocess' `stdin`. This resolves with that second subprocess' result. If either subprocess is rejected, this is rejected with that subprocess' error instead. This follows the same syntax as `execa(file, arguments?, options?)` except both regular options and pipe-specific options can be specified. + + Like a subprocess, the return value can be [iterated](https://github.com/sindresorhus/execa/blob/main/docs/lines.md#progressive-splitting), [converted to a stream](https://github.com/sindresorhus/execa/blob/main/docs/streams.md#converting-a-subprocess-to-a-stream), or used for [IPC](https://github.com/sindresorhus/execa/blob/main/docs/ipc.md) with the destination subprocess. */ pipe( file: string | URL, arguments?: readonly string[], options?: OptionsType, - ): Promise> & PipableSubprocess; + ): Promise> & PipableSubprocess & PipeResultMethods; pipe( file: string | URL, options?: OptionsType, - ): Promise> & PipableSubprocess; + ): Promise> & PipableSubprocess & PipeResultMethods; /** Like `subprocess.pipe(file, arguments?, options?)` but using a `command` template string instead. This follows the same syntax as `$`. */ pipe(templates: TemplateStringsArray, ...expressions: readonly TemplateExpression[]): - Promise> & PipableSubprocess; + Promise> & PipableSubprocess & PipeResultMethods<{}>; pipe(options: OptionsType): (templates: TemplateStringsArray, ...expressions: readonly TemplateExpression[]) - => Promise> & PipableSubprocess; + => Promise> & PipableSubprocess & PipeResultMethods; /** Like `subprocess.pipe(file, arguments?, options?)` but using the return value of another `execa()` call instead. */ pipe(destination: Destination, options?: PipeOptions): - Promise> & PipableSubprocess; + Promise> & PipableSubprocess & PipeResultMethodsFrom; }; export {}; diff --git a/types/subprocess/subprocess.d.ts b/types/subprocess/subprocess.d.ts index ec054ef89b..de81d664b8 100644 --- a/types/subprocess/subprocess.d.ts +++ b/types/subprocess/subprocess.d.ts @@ -15,9 +15,39 @@ import type {SubprocessStdioStream} from './stdout.js'; import type {SubprocessStdioArray} from './stdio.js'; 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. +export type SubprocessResultMethods = { + /** + Stream combining/interleaving `subprocess.stdout` and `subprocess.stderr`. + + This requires the `all` option to be `true`. + + This is `undefined` if `stdout` and `stderr` options are set to `'inherit'`, `'ignore'`, `Writable` or `integer`, or if the `buffer` option is `false`. + */ + all: SubprocessAll; + + /** + Subprocesses are [async iterables](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Symbol/asyncIterator). They iterate over each output line. + */ + [Symbol.asyncIterator](): SubprocessAsyncIterable; + + /** + Same as `subprocess[Symbol.asyncIterator]` except options can be provided. + */ + iterable(readableOptions?: IterableOptions): SubprocessAsyncIterable; + + /** + Converts the subprocess to a readable stream. + */ + readable(readableOptions?: ReadableOptions): Readable; +}; + type ExecaCustomSubprocess = & IpcMethods, OptionsType['serialization']> & PipableSubprocess + & SubprocessResultMethods & { /** Process identifier ([PID](https://en.wikipedia.org/wiki/Process_identifier)). @@ -47,15 +77,6 @@ type ExecaCustomSubprocess = */ stderr: SubprocessStdioStream<'2', OptionsType>; - /** - Stream combining/interleaving `subprocess.stdout` and `subprocess.stderr`. - - This requires the `all` option to be `true`. - - This is `undefined` if `stdout` and `stderr` options are set to `'inherit'`, `'ignore'`, `Writable` or `integer`, or if the `buffer` option is `false`. - */ - all: SubprocessAll; - /** The subprocess `stdin`, `stdout`, `stderr` and other files descriptors as an array of streams. @@ -75,21 +96,6 @@ type ExecaCustomSubprocess = kill(signal?: keyof SignalConstants | number, error?: Error): boolean; kill(error?: Error): boolean; - /** - Subprocesses are [async iterables](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Symbol/asyncIterator). They iterate over each output line. - */ - [Symbol.asyncIterator](): SubprocessAsyncIterable; - - /** - Same as `subprocess[Symbol.asyncIterator]` except options can be provided. - */ - iterable(readableOptions?: IterableOptions): SubprocessAsyncIterable; - - /** - Converts the subprocess to a readable stream. - */ - readable(readableOptions?: ReadableOptions): Readable; - /** Converts the subprocess to a writable stream. */