Expose the destination subprocess methods on .pipe()

docs/api.md

@@ -253,6 +253,8 @@ _Returns_: [`Promise<Result>`](#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\`

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.

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});

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});

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();
 	}