zhibo_2025/node_modules/execa/lib/io/output-async.js

import mergeStreams from '@sindresorhus/merge-streams';
import {isStandardStream} from '../utils/standard-stream.js';
import {incrementMaxListeners} from '../utils/max-listeners.js';
import {TRANSFORM_TYPES} from '../stdio/type.js';
import {pipeStreams} from './pipeline.js';

// Handle `input`, `inputFile`, `stdin`, `stdout` and `stderr` options, after spawning, in async mode
// When multiple input streams are used, we merge them to ensure the output stream ends only once each input stream has ended
export const pipeOutputAsync = (subprocess, fileDescriptors, controller) => {
	const pipeGroups = new Map();

	for (const [fdNumber, {stdioItems, direction}] of Object.entries(fileDescriptors)) {
		for (const {stream} of stdioItems.filter(({type}) => TRANSFORM_TYPES.has(type))) {
			pipeTransform(subprocess, stream, direction, fdNumber);
		}

		for (const {stream} of stdioItems.filter(({type}) => !TRANSFORM_TYPES.has(type))) {
			pipeStdioItem({
				subprocess,
				stream,
				direction,
				fdNumber,
				pipeGroups,
				controller,
			});
		}
	}

	for (const [outputStream, inputStreams] of pipeGroups.entries()) {
		const inputStream = inputStreams.length === 1 ? inputStreams[0] : mergeStreams(inputStreams);
		pipeStreams(inputStream, outputStream);
	}
};

// When using transforms, `subprocess.stdin|stdout|stderr|stdio` is directly mutated
const pipeTransform = (subprocess, stream, direction, fdNumber) => {
	if (direction === 'output') {
		pipeStreams(subprocess.stdio[fdNumber], stream);
	} else {
		pipeStreams(stream, subprocess.stdio[fdNumber]);
	}

	const streamProperty = SUBPROCESS_STREAM_PROPERTIES[fdNumber];
	if (streamProperty !== undefined) {
		subprocess[streamProperty] = stream;
	}

	subprocess.stdio[fdNumber] = stream;
};

const SUBPROCESS_STREAM_PROPERTIES = ['stdin', 'stdout', 'stderr'];

// Most `std*` option values involve piping `subprocess.std*` to a stream.
// The stream is either passed by the user or created internally.
const pipeStdioItem = ({subprocess, stream, direction, fdNumber, pipeGroups, controller}) => {
	if (stream === undefined) {
		return;
	}

	setStandardStreamMaxListeners(stream, controller);

	const [inputStream, outputStream] = direction === 'output'
		? [stream, subprocess.stdio[fdNumber]]
		: [subprocess.stdio[fdNumber], stream];
	const outputStreams = pipeGroups.get(inputStream) ?? [];
	pipeGroups.set(inputStream, [...outputStreams, outputStream]);
};

// Multiple subprocesses might be piping from/to `process.std*` at the same time.
// This is not necessarily an error and should not print a `maxListeners` warning.
const setStandardStreamMaxListeners = (stream, {signal}) => {
	if (isStandardStream(stream)) {
		incrementMaxListeners(stream, MAX_LISTENERS_INCREMENT, signal);
	}
};

// `source.pipe(destination)` adds at most 1 listener for each event.
// If `stdin` option is an array, the values might be combined with `merge-streams`.
// That library also listens for `source` end, which adds 1 more listener.
const MAX_LISTENERS_INCREMENT = 2;