init project
This commit is contained in:
root
+20
@@ -0,0 +1,20 @@
|
||||
import {aborted} from 'node:util';
|
||||
import {createNonCommandError} from './throw.js';
|
||||
|
||||
// When passing an `unpipeSignal` option, abort piping when the signal is aborted.
|
||||
// However, do not terminate the subprocesses.
|
||||
export const unpipeOnAbort = (unpipeSignal, unpipeContext) => unpipeSignal === undefined
|
||||
? []
|
||||
: [unpipeOnSignalAbort(unpipeSignal, unpipeContext)];
|
||||
|
||||
const unpipeOnSignalAbort = async (unpipeSignal, {sourceStream, mergedStream, fileDescriptors, sourceOptions, startTime}) => {
|
||||
await aborted(unpipeSignal, sourceStream);
|
||||
await mergedStream.remove(sourceStream);
|
||||
const error = new Error('Pipe canceled by `unpipeSignal` option.');
|
||||
throw createNonCommandError({
|
||||
error,
|
||||
fileDescriptors,
|
||||
sourceOptions,
|
||||
startTime,
|
||||
});
|
||||
};
|
||||
+91
@@ -0,0 +1,91 @@
|
||||
import {normalizeParameters} from '../methods/parameters.js';
|
||||
import {getStartTime} from '../return/duration.js';
|
||||
import {SUBPROCESS_OPTIONS, getToStream, getFromStream} from '../arguments/fd-options.js';
|
||||
import {isDenoExecPath} from '../arguments/file-url.js';
|
||||
|
||||
// Normalize and validate arguments passed to `source.pipe(destination)`
|
||||
export const normalizePipeArguments = ({source, sourcePromise, boundOptions, createNested}, ...pipeArguments) => {
|
||||
const startTime = getStartTime();
|
||||
const {
|
||||
destination,
|
||||
destinationStream,
|
||||
destinationError,
|
||||
from,
|
||||
unpipeSignal,
|
||||
} = getDestinationStream(boundOptions, createNested, pipeArguments);
|
||||
const {sourceStream, sourceError} = getSourceStream(source, from);
|
||||
const {options: sourceOptions, fileDescriptors} = SUBPROCESS_OPTIONS.get(source);
|
||||
return {
|
||||
sourcePromise,
|
||||
sourceStream,
|
||||
sourceOptions,
|
||||
sourceError,
|
||||
destination,
|
||||
destinationStream,
|
||||
destinationError,
|
||||
unpipeSignal,
|
||||
fileDescriptors,
|
||||
startTime,
|
||||
};
|
||||
};
|
||||
|
||||
const getDestinationStream = (boundOptions, createNested, pipeArguments) => {
|
||||
try {
|
||||
const {
|
||||
destination,
|
||||
pipeOptions: {from, to, unpipeSignal} = {},
|
||||
} = getDestination(boundOptions, createNested, ...pipeArguments);
|
||||
const destinationStream = getToStream(destination, to);
|
||||
return {
|
||||
destination,
|
||||
destinationStream,
|
||||
from,
|
||||
unpipeSignal,
|
||||
};
|
||||
} catch (error) {
|
||||
return {destinationError: error};
|
||||
}
|
||||
};
|
||||
|
||||
// Piping subprocesses can use three syntaxes:
|
||||
// - source.pipe('command', commandArguments, pipeOptionsOrDestinationOptions)
|
||||
// - source.pipe`command commandArgument` or source.pipe(pipeOptionsOrDestinationOptions)`command commandArgument`
|
||||
// - source.pipe(execa(...), pipeOptions)
|
||||
const getDestination = (boundOptions, createNested, firstArgument, ...pipeArguments) => {
|
||||
if (Array.isArray(firstArgument)) {
|
||||
const destination = createNested(mapDestinationArguments, boundOptions)(firstArgument, ...pipeArguments);
|
||||
return {destination, pipeOptions: boundOptions};
|
||||
}
|
||||
|
||||
if (typeof firstArgument === 'string' || firstArgument instanceof URL || isDenoExecPath(firstArgument)) {
|
||||
if (Object.keys(boundOptions).length > 0) {
|
||||
throw new TypeError('Please use .pipe("file", ..., options) or .pipe(execa("file", ..., options)) instead of .pipe(options)("file", ...).');
|
||||
}
|
||||
|
||||
const [rawFile, rawArguments, rawOptions] = normalizeParameters(firstArgument, ...pipeArguments);
|
||||
const destination = createNested(mapDestinationArguments)(rawFile, rawArguments, rawOptions);
|
||||
return {destination, pipeOptions: rawOptions};
|
||||
}
|
||||
|
||||
if (SUBPROCESS_OPTIONS.has(firstArgument)) {
|
||||
if (Object.keys(boundOptions).length > 0) {
|
||||
throw new TypeError('Please use .pipe(options)`command` or .pipe($(options)`command`) instead of .pipe(options)($`command`).');
|
||||
}
|
||||
|
||||
return {destination: firstArgument, pipeOptions: pipeArguments[0]};
|
||||
}
|
||||
|
||||
throw new TypeError(`The first argument must be a template string, an options object, or an Execa subprocess: ${firstArgument}`);
|
||||
};
|
||||
|
||||
// Force `stdin: 'pipe'` with the destination subprocess
|
||||
const mapDestinationArguments = ({options}) => ({options: {...options, stdin: 'pipe', piped: true}});
|
||||
|
||||
const getSourceStream = (source, from) => {
|
||||
try {
|
||||
const sourceStream = getFromStream(source, from);
|
||||
return {sourceStream};
|
||||
} catch (error) {
|
||||
return {sourceError: error};
|
||||
}
|
||||
};
|
||||
+24
@@ -0,0 +1,24 @@
|
||||
// Like Bash, we await both subprocesses. This is unlike some other shells which only await the destination subprocess.
|
||||
// Like Bash with the `pipefail` option, if either subprocess fails, the whole pipe fails.
|
||||
// Like Bash, if both subprocesses fail, we return the failure of the destination.
|
||||
// This ensures both subprocesses' errors are present, using `error.pipedFrom`.
|
||||
export const waitForBothSubprocesses = async subprocessPromises => {
|
||||
const [
|
||||
{status: sourceStatus, reason: sourceReason, value: sourceResult = sourceReason},
|
||||
{status: destinationStatus, reason: destinationReason, value: destinationResult = destinationReason},
|
||||
] = await subprocessPromises;
|
||||
|
||||
if (!destinationResult.pipedFrom.includes(sourceResult)) {
|
||||
destinationResult.pipedFrom.push(sourceResult);
|
||||
}
|
||||
|
||||
if (destinationStatus === 'rejected') {
|
||||
throw destinationResult;
|
||||
}
|
||||
|
||||
if (sourceStatus === 'rejected') {
|
||||
throw sourceResult;
|
||||
}
|
||||
|
||||
return destinationResult;
|
||||
};
|
||||
+230
@@ -0,0 +1,230 @@
|
||||
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 {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';
|
||||
import {handlePipeArgumentsError} from './throw.js';
|
||||
import {waitForBothSubprocesses} from './sequence.js';
|
||||
import {pipeSubprocessStream} from './streaming.js';
|
||||
import {unpipeOnAbort} from './abort.js';
|
||||
|
||||
// Pipe a subprocess' `stdout`/`stderr`/`stdio` into another subprocess' `stdin`
|
||||
export const pipeToSubprocess = (sourceInfo, ...pipeArguments) => {
|
||||
if (isPlainObject(pipeArguments[0])) {
|
||||
return pipeToSubprocess.bind(undefined, {
|
||||
...sourceInfo,
|
||||
boundOptions: {...sourceInfo.boundOptions, ...pipeArguments[0]},
|
||||
});
|
||||
}
|
||||
|
||||
const {destination, ...normalizedInfo} = normalizePipeArguments(sourceInfo, ...pipeArguments);
|
||||
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,
|
||||
});
|
||||
promise.readableStream = createReadableStream.bind(undefined, promise);
|
||||
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()`, `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 ({
|
||||
sourcePromise,
|
||||
sourceStream,
|
||||
sourceOptions,
|
||||
sourceError,
|
||||
destination,
|
||||
destinationStream,
|
||||
destinationError,
|
||||
unpipeSignal,
|
||||
fileDescriptors,
|
||||
startTime,
|
||||
pipeFailureController,
|
||||
}) => {
|
||||
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),
|
||||
...unpipeOnAbort(unpipeSignal, {
|
||||
sourceStream,
|
||||
mergedStream,
|
||||
sourceOptions,
|
||||
fileDescriptors,
|
||||
startTime,
|
||||
}),
|
||||
]);
|
||||
} catch (error) {
|
||||
pipeFailureController.abort(error);
|
||||
throw error;
|
||||
} finally {
|
||||
maxListenersController.abort();
|
||||
}
|
||||
};
|
||||
|
||||
// `.pipe()` awaits the subprocess promises.
|
||||
// When invalid arguments are passed to `.pipe()`, we throw an error, which prevents awaiting them.
|
||||
// We need to ensure this does not create unhandled rejections.
|
||||
const getSubprocessPromises = (sourcePromise, destination) => Promise.allSettled([sourcePromise, destination]);
|
||||
+51
@@ -0,0 +1,51 @@
|
||||
import {finished} from 'node:stream/promises';
|
||||
import mergeStreams from '@sindresorhus/merge-streams';
|
||||
import {incrementMaxListeners} from '../utils/max-listeners.js';
|
||||
import {pipeStreams} from '../io/pipeline.js';
|
||||
|
||||
// The piping behavior is like Bash.
|
||||
// In particular, when one subprocess exits, the other is not terminated by a signal.
|
||||
// Instead, its stdout (for the source) or stdin (for the destination) closes.
|
||||
// If the subprocess uses it, it will make it error with SIGPIPE or EPIPE (for the source) or end (for the destination).
|
||||
// If it does not use it, it will continue running.
|
||||
// This allows for subprocesses to gracefully exit and lower the coupling between subprocesses.
|
||||
export const pipeSubprocessStream = (sourceStream, destinationStream, maxListenersController) => {
|
||||
const mergedStream = MERGED_STREAMS.has(destinationStream)
|
||||
? pipeMoreSubprocessStream(sourceStream, destinationStream)
|
||||
: pipeFirstSubprocessStream(sourceStream, destinationStream);
|
||||
incrementMaxListeners(sourceStream, SOURCE_LISTENERS_PER_PIPE, maxListenersController.signal);
|
||||
incrementMaxListeners(destinationStream, DESTINATION_LISTENERS_PER_PIPE, maxListenersController.signal);
|
||||
cleanupMergedStreamsMap(destinationStream);
|
||||
return mergedStream;
|
||||
};
|
||||
|
||||
// We use `merge-streams` to allow for multiple sources to pipe to the same destination.
|
||||
const pipeFirstSubprocessStream = (sourceStream, destinationStream) => {
|
||||
const mergedStream = mergeStreams([sourceStream]);
|
||||
pipeStreams(mergedStream, destinationStream);
|
||||
MERGED_STREAMS.set(destinationStream, mergedStream);
|
||||
return mergedStream;
|
||||
};
|
||||
|
||||
const pipeMoreSubprocessStream = (sourceStream, destinationStream) => {
|
||||
const mergedStream = MERGED_STREAMS.get(destinationStream);
|
||||
mergedStream.add(sourceStream);
|
||||
return mergedStream;
|
||||
};
|
||||
|
||||
const cleanupMergedStreamsMap = async destinationStream => {
|
||||
try {
|
||||
await finished(destinationStream, {cleanup: true, readable: false, writable: true});
|
||||
} catch {}
|
||||
|
||||
MERGED_STREAMS.delete(destinationStream);
|
||||
};
|
||||
|
||||
const MERGED_STREAMS = new WeakMap();
|
||||
|
||||
// Number of listeners set up on `sourceStream` by each `sourceStream.pipe(destinationStream)`
|
||||
// Those are added by `merge-streams`
|
||||
const SOURCE_LISTENERS_PER_PIPE = 2;
|
||||
// Number of listeners set up on `destinationStream` by each `sourceStream.pipe(destinationStream)`
|
||||
// Those are added by `finished()` in `cleanupMergedStreamsMap()`
|
||||
const DESTINATION_LISTENERS_PER_PIPE = 1;
|
||||
+58
@@ -0,0 +1,58 @@
|
||||
import {makeEarlyError} from '../return/result.js';
|
||||
import {abortSourceStream, endDestinationStream} from '../io/pipeline.js';
|
||||
|
||||
// When passing invalid arguments to `source.pipe()`, throw asynchronously.
|
||||
// We also abort both subprocesses.
|
||||
export const handlePipeArgumentsError = ({
|
||||
sourceStream,
|
||||
sourceError,
|
||||
destinationStream,
|
||||
destinationError,
|
||||
fileDescriptors,
|
||||
sourceOptions,
|
||||
startTime,
|
||||
}) => {
|
||||
const error = getPipeArgumentsError({
|
||||
sourceStream,
|
||||
sourceError,
|
||||
destinationStream,
|
||||
destinationError,
|
||||
});
|
||||
if (error !== undefined) {
|
||||
throw createNonCommandError({
|
||||
error,
|
||||
fileDescriptors,
|
||||
sourceOptions,
|
||||
startTime,
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
const getPipeArgumentsError = ({sourceStream, sourceError, destinationStream, destinationError}) => {
|
||||
if (sourceError !== undefined && destinationError !== undefined) {
|
||||
return destinationError;
|
||||
}
|
||||
|
||||
if (destinationError !== undefined) {
|
||||
abortSourceStream(sourceStream);
|
||||
return destinationError;
|
||||
}
|
||||
|
||||
if (sourceError !== undefined) {
|
||||
endDestinationStream(destinationStream);
|
||||
return sourceError;
|
||||
}
|
||||
};
|
||||
|
||||
// Specific error return value when passing invalid arguments to `subprocess.pipe()` or when using `unpipeSignal`
|
||||
export const createNonCommandError = ({error, fileDescriptors, sourceOptions, startTime}) => makeEarlyError({
|
||||
error,
|
||||
command: PIPE_COMMAND_MESSAGE,
|
||||
escapedCommand: PIPE_COMMAND_MESSAGE,
|
||||
fileDescriptors,
|
||||
options: sourceOptions,
|
||||
startTime,
|
||||
isSync: false,
|
||||
});
|
||||
|
||||
const PIPE_COMMAND_MESSAGE = 'source.pipe(destination)';
|
||||
Reference in New Issue
Block a user