Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 12 additions & 2 deletions docs/api.md
Original file line number Diff line number Diff line change
Expand Up @@ -211,7 +211,7 @@ This can only be called inside a subprocess. This requires the [`gracefulCancel`
## Return value

_TypeScript:_ [`ResultPromise`](typescript.md)\
_Type:_ `Promise<object> | Subprocess`
_Type:_ `Subprocess`

The return value of all [asynchronous methods](#methods) is both:
- the [subprocess](#subprocess).
Expand All @@ -223,7 +223,9 @@ The return value of all [asynchronous methods](#methods) is both:

_TypeScript:_ [`Subprocess`](typescript.md)

[`child_process` instance](https://nodejs.org/api/child_process.html#child_process_class_childprocess) with the following methods and properties.
Promise with the following Execa-specific methods and properties.

The underlying Node.js [`ChildProcess`](https://nodejs.org/api/child_process.html#class-childprocess) instance is available as [`subprocess.nodeChildProcess`](#subprocessnodechildprocess). This is an escape hatch for Node.js-specific APIs that Execa does not document, such as `.on()`, `.send()`, `.disconnect()`, `.ref()` and `.unref()`.

### subprocess\[Symbol.asyncIterator\]()

Expand Down Expand Up @@ -335,6 +337,14 @@ This is `undefined` if the subprocess failed to spawn.

[More info.](termination.md#inter-process-termination)

### subprocess.nodeChildProcess

_Type:_ [`ChildProcess`](https://nodejs.org/api/child_process.html#class-childprocess)

Underlying Node.js child process instance.

This is intended as an escape hatch for Node.js-specific APIs that Execa does not document.

### subprocess.sendMessage(message, sendMessageOptions)

`message`: [`Message`](ipc.md#message-type)\
Expand Down
4 changes: 2 additions & 2 deletions docs/execution.md
Original file line number Diff line number Diff line change
Expand Up @@ -127,7 +127,7 @@ await timedExeca`npm run test`;

### Subprocess

The subprocess is returned as soon as it is spawned. It is a [`child_process` instance](https://nodejs.org/api/child_process.html#child_process_class_childprocess) with [additional methods and properties](api.md#subprocess).
The subprocess is returned as soon as it is spawned. It is a `Promise` with [Execa-specific methods and properties](api.md#subprocess). The underlying Node.js [`ChildProcess`](https://nodejs.org/api/child_process.html#class-childprocess) is available as [`subprocess.nodeChildProcess`](api.md#subprocessnodechildprocess).

```js
const subprocess = execa`npm run build`;
Expand All @@ -136,7 +136,7 @@ console.log(subprocess.pid);

### Result

The subprocess is also a `Promise` that resolves with the [`result`](api.md#result).
The subprocess resolves with the [`result`](api.md#result).

```js
const {stdout} = await execa`npm run build`;
Expand Down
2 changes: 1 addition & 1 deletion lib/convert/duplex.js
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ export const createDuplex = ({subprocess, concurrentStreams, encoding}, {from, t
subprocess,
subprocessStdin,
});
onStdinFinished(subprocessStdin, duplex, subprocessStdout);
onStdinFinished(subprocessStdin, duplex, subprocessStdout, subprocess);
return duplex;
};

Expand Down
16 changes: 15 additions & 1 deletion lib/convert/readable.js
Original file line number Diff line number Diff line change
Expand Up @@ -96,10 +96,24 @@ export const onStdoutFinished = async ({subprocessStdout, onStdoutDataDone, read
}
} catch (error) {
await safeWaitForSubprocessStdin(subprocessStdin);
destroyOtherReadable(readable, error);
destroyOtherReadable(readable, await getPrematureCloseError(subprocess, error));
}
};

const getPrematureCloseError = async (subprocess, error) => {
if (error.code !== 'ERR_STREAM_PREMATURE_CLOSE') {
return error;
}

try {
await subprocess;
} catch (subprocessError) {
return subprocessError;
}

return error;
};

// When `readable` aborts/errors, do the same on `subprocess.stdout`
export const onReadableDestroy = async ({subprocessStdout, subprocess, waitReadableDestroy}, error) => {
if (!await waitForConcurrentStreams(waitReadableDestroy, subprocess)) {
Expand Down
25 changes: 22 additions & 3 deletions lib/convert/writable.js
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import {Writable} from 'node:stream';
import {callbackify} from 'node:util';
import {getToStream} from '../arguments/fd-options.js';
import {isStreamAbort} from '../resolve/wait-stream.js';
import {addConcurrentStream, waitForConcurrentStreams} from './concurrent.js';
import {
safeWaitForSubprocessStdout,
Expand All @@ -23,7 +24,7 @@ export const createWritable = ({subprocess, concurrentStreams}, {to} = {}) => {
highWaterMark: subprocessStdin.writableHighWaterMark,
objectMode: subprocessStdin.writableObjectMode,
});
onStdinFinished(subprocessStdin, writable);
onStdinFinished(subprocessStdin, writable, undefined, subprocess);
return writable;
};

Expand Down Expand Up @@ -66,18 +67,36 @@ const onWritableFinal = async (subprocessStdin, subprocess, waitWritableFinal) =
};

// When `subprocess.stdin` ends/aborts/errors, do the same on `writable`.
export const onStdinFinished = async (subprocessStdin, writable, subprocessStdout) => {
export const onStdinFinished = async (subprocessStdin, writable, subprocessStdout, subprocess) => {
try {
await waitForSubprocessStdin(subprocessStdin);
await subprocess;
if (writable.writable) {
writable.end();
}
} catch (error) {
await safeWaitForSubprocessStdout(subprocessStdout);
destroyOtherWritable(writable, error);
destroyOtherWritable(writable, await getSubprocessError(subprocess, error));
}
};

const getSubprocessError = async (subprocess, error) => {
if (!shouldUseSubprocessError(error)) {
return error;
}

try {
await subprocess;
} catch (subprocessError) {
return subprocessError;
}

return error;
};

const shouldUseSubprocessError = error => error === undefined
|| isStreamAbort(error);

// When `writable` aborts/errors, do the same on `subprocess.stdin`
export const onWritableDestroy = async ({subprocessStdin, subprocess, waitWritableFinal, waitWritableDestroy}, error) => {
await waitForConcurrentStreams(waitWritableFinal, subprocess);
Expand Down
2 changes: 1 addition & 1 deletion lib/ipc/forward.js
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,6 @@ const forwardEvents = ({ipcEmitter, anyProcess, channel, isSubprocess}) => {
export const isConnected = anyProcess => {
const ipcEmitter = IPC_EMITTERS.get(anyProcess);
return ipcEmitter === undefined
? anyProcess.channel !== null
? anyProcess.channel !== undefined && anyProcess.channel !== null
: ipcEmitter.connected;
};
7 changes: 4 additions & 3 deletions lib/ipc/get-each.js
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ export const getEachMessage = (subprocessInfo, options = {}) => {
};

// Same but used internally
export const loopOnMessages = ({anyProcess, channel, isSubprocess, ipc, shouldAwait, reference, signal}) => {
export const loopOnMessages = ({anyProcess, waitProcess = anyProcess, channel, isSubprocess, ipc, shouldAwait, reference, signal}) => {
validateIpcMethod({
methodName: 'getEachMessage',
isSubprocess,
Expand All @@ -42,6 +42,7 @@ export const loopOnMessages = ({anyProcess, channel, isSubprocess, ipc, shouldAw
});
return iterateOnMessages({
anyProcess,
waitProcess,
channel,
ipcEmitter,
isSubprocess,
Expand Down Expand Up @@ -83,7 +84,7 @@ const abortOnStrictError = async ({ipcEmitter, isSubprocess, controller, state})
} catch {}
};

const iterateOnMessages = async function * ({anyProcess, channel, ipcEmitter, isSubprocess, shouldAwait, controller, state, reference}) {
const iterateOnMessages = async function * ({anyProcess, waitProcess, channel, ipcEmitter, isSubprocess, shouldAwait, controller, state, reference}) {
try {
for await (const [message] of on(ipcEmitter, 'message', {signal: controller.signal})) {
throwIfStrictError(state);
Expand All @@ -100,7 +101,7 @@ const iterateOnMessages = async function * ({anyProcess, channel, ipcEmitter, is
}

if (shouldAwait) {
await anyProcess;
await waitProcess;
}
}
};
Expand Down
10 changes: 8 additions & 2 deletions lib/ipc/ipc-input.js
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import {serialize} from 'node:v8';
import {sendMessage} from './send.js';

// Validate the `ipcInput` option
export const validateIpcInputOption = ({ipcInput, ipc, serialization}) => {
Expand Down Expand Up @@ -35,10 +36,15 @@ const validateIpcInput = {
};

// When the `ipcInput` option is set, it is sent as an initial IPC message to the subprocess
export const sendIpcInput = async (subprocess, ipcInput) => {
export const sendIpcInput = async (subprocess, ipcInput, ipc) => {
if (ipcInput === undefined) {
return;
}

await subprocess.sendMessage(ipcInput);
await sendMessage({
anyProcess: subprocess,
channel: subprocess.channel,
isSubprocess: false,
ipc,
}, ipcInput);
};
7 changes: 4 additions & 3 deletions lib/ipc/methods.js
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,8 @@ import {getEachMessage} from './get-each.js';
import {getCancelSignal} from './graceful.js';

// Add promise-based IPC methods in current process
export const addIpcMethods = (subprocess, {ipc}) => {
Object.assign(subprocess, getIpcMethods(subprocess, false, ipc));
export const addIpcMethods = (target, subprocess, {ipc}) => {
Object.assign(target, getIpcMethods(subprocess, false, ipc, target));
};

// Get promise-based IPC in the subprocess
Expand All @@ -27,7 +27,7 @@ export const getIpcExport = () => {
};

// Retrieve the `ipc` shared by both the current process and the subprocess
const getIpcMethods = (anyProcess, isSubprocess, ipc) => ({
const getIpcMethods = (anyProcess, isSubprocess, ipc, waitProcess = anyProcess) => ({
sendMessage: sendMessage.bind(undefined, {
anyProcess,
channel: anyProcess.channel,
Expand All @@ -45,5 +45,6 @@ const getIpcMethods = (anyProcess, isSubprocess, ipc) => ({
channel: anyProcess.channel,
isSubprocess,
ipc,
waitProcess,
}),
});
12 changes: 9 additions & 3 deletions lib/ipc/outgoing.js
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
import {createDeferred} from '../utils/deferred.js';
import {getFdSpecificValue} from '../arguments/specific.js';
import {SUBPROCESS_OPTIONS} from '../arguments/fd-options.js';
import {validateStrictDeadlock} from './strict.js';

// When `sendMessage()` is ongoing, any `message` being received waits before being emitted.
Expand Down Expand Up @@ -35,13 +34,20 @@ export const waitForOutgoingMessages = async (anyProcess, ipcEmitter, wrappedMes
};

const OUTGOING_MESSAGES = new WeakMap();
const IPC_SUBPROCESS_OPTIONS = new WeakMap();

export const setIpcSubprocessOptions = (subprocess, options) => {
IPC_SUBPROCESS_OPTIONS.set(subprocess, options);
};

// Whether any `message` listener is setup
export const hasMessageListeners = (anyProcess, ipcEmitter) => ipcEmitter.listenerCount('message') > getMinListenerCount(anyProcess);

// When `buffer` is `false`, we set up a `message` listener that should be ignored.
// That listener is only meant to intercept `strict` acknowledgement responses.
const getMinListenerCount = anyProcess => SUBPROCESS_OPTIONS.has(anyProcess)
&& !getFdSpecificValue(SUBPROCESS_OPTIONS.get(anyProcess).options.buffer, 'ipc')
const getMinListenerCount = anyProcess => getOptions(anyProcess) !== undefined
&& !getFdSpecificValue(getOptions(anyProcess).buffer, 'ipc')
? 1
: 0;

const getOptions = anyProcess => IPC_SUBPROCESS_OPTIONS.get(anyProcess);
58 changes: 49 additions & 9 deletions lib/methods/main-async.js
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import {normalizeOptions} from '../arguments/options.js';
import {SUBPROCESS_OPTIONS} from '../arguments/fd-options.js';
import {concatenateShell} from '../arguments/shell.js';
import {addIpcMethods} from '../ipc/methods.js';
import {setIpcSubprocessOptions} from '../ipc/outgoing.js';
import {makeError, makeSuccessResult} from '../return/result.js';
import {handleResult} from '../return/reject.js';
import {handleEarlyError} from '../return/early-error.js';
Expand All @@ -24,7 +25,7 @@ import {mergePromise} from './promise.js';
// Main shared logic for all async methods: `execa()`, `$`, `execaNode()`
export const execaCoreAsync = (rawFile, rawArguments, rawOptions, createNested) => {
const {file, commandArguments, command, escapedCommand, startTime, verboseInfo, options, fileDescriptors} = handleAsyncArguments(rawFile, rawArguments, rawOptions);
const {subprocess, promise} = spawnSubprocessAsync({
const {subprocess: nodeChildProcess, promise, kill, all, convertedStreams} = spawnSubprocessAsync({
file,
commandArguments,
options,
Expand All @@ -34,13 +35,20 @@ export const execaCoreAsync = (rawFile, rawArguments, rawOptions, createNested)
escapedCommand,
fileDescriptors,
});
const subprocess = getSubprocessPromise({
promise,
nodeChildProcess,
kill,
all,
convertedStreams,
options,
});
subprocess.pipe = pipeToSubprocess.bind(undefined, {
source: subprocess,
sourcePromise: promise,
boundOptions: {},
createNested,
});
mergePromise(subprocess, promise);
SUBPROCESS_OPTIONS.set(subprocess, {options, fileDescriptors});
return subprocess;
};
Expand Down Expand Up @@ -94,23 +102,24 @@ const spawnSubprocessAsync = ({file, commandArguments, options, startTime, verbo

const originalStreams = [...subprocess.stdio];
pipeOutputAsync(subprocess, fileDescriptors, controller);
cleanupOnExit(subprocess, options, controller);
setIpcSubprocessOptions(subprocess, options);

const context = {};
const onInternalError = createDeferred();
subprocess.kill = subprocessKill.bind(undefined, {
const kill = subprocessKill.bind(undefined, {
kill: subprocess.kill.bind(subprocess),
options,
onInternalError,
context,
controller,
});
subprocess.all = makeAllStream(subprocess, options);
addConvertedStreams(subprocess, options);
addIpcMethods(subprocess, options);
cleanupOnExit(kill, options, controller);
const all = makeAllStream(subprocess, options);

const promise = handlePromise({
subprocess,
kill,
all,
options,
startTime,
verboseInfo,
Expand All @@ -122,11 +131,40 @@ const spawnSubprocessAsync = ({file, commandArguments, options, startTime, verbo
onInternalError,
controller,
});
return {subprocess, promise};
return {
subprocess,
promise,
kill,
all,
};
};

const getSubprocessPromise = ({promise, nodeChildProcess, kill, all, convertedStreams, options}) => {
const subprocess = mergePromise(promise, getSubprocessProperties(nodeChildProcess, all));
subprocess.kill = kill ?? subprocess.kill;
if (convertedStreams === undefined) {
addConvertedStreams(subprocess, options);
} else {
Object.assign(subprocess, convertedStreams);
}

addIpcMethods(subprocess, nodeChildProcess, options);
return subprocess;
};

const getSubprocessProperties = (nodeChildProcess, all) => ({
nodeChildProcess,
pid: nodeChildProcess.pid,
stdin: nodeChildProcess.stdin,
stdout: nodeChildProcess.stdout,
stderr: nodeChildProcess.stderr,
stdio: nodeChildProcess.stdio,
all,
kill: nodeChildProcess.kill.bind(nodeChildProcess),
});

// Asynchronous logic, as opposed to the previous logic which can be run synchronously, i.e. can be returned to user right away
const handlePromise = async ({subprocess, options, startTime, verboseInfo, fileDescriptors, originalStreams, command, escapedCommand, context, onInternalError, controller}) => {
const handlePromise = async ({subprocess, kill, all: allStream, options, startTime, verboseInfo, fileDescriptors, originalStreams, command, escapedCommand, context, onInternalError, controller}) => {
const [
errorInfo,
[exitCode, signal],
Expand All @@ -135,6 +173,8 @@ const handlePromise = async ({subprocess, options, startTime, verboseInfo, fileD
ipcOutput,
] = await waitForSubprocessResult({
subprocess,
kill,
all: allStream,
options,
context,
verboseInfo,
Expand Down
Loading
Loading