13468958302

Committed 22 Feb 2025 03:02AM UTC coverage: 98.18%. Remained the same

Build # 13468958302

Build Type

push

github

Committed by

web-flow

Commit Message

deps: bump esbuild from 0.23.1 to 0.25.0 (#528)

Run Details

552 of 572 branches covered (96.5%)

Branch coverage included in aggregate %.

797 of 802 relevant lines covered (99.38%)

461.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.26

/src/command.ts

import {
    ChildProcess as BaseChildProcess,
    MessageOptions,
    SendHandle,
    SpawnOptions,
} from 'child_process';
import * as Rx from 'rxjs';
import { EventEmitter, Writable } from 'stream';

/**
 * Identifier for a command; if string, it's the command's name, if number, it's the index.
 */
export type CommandIdentifier = string | number;

export interface CommandInfo {
    /**
     * Command's name.
     */
    name: string;

    /**
     * Which command line the command has.
     */
    command: string;

    /**
     * Which environment variables should the spawned process have.
     */
    env?: Record<string, unknown>;

    /**
     * The current working directory of the process when spawned.
     */
    cwd?: string;

    /**
     * Color to use on prefix of the command.
     */
    prefixColor?: string;

    /**
     * Whether sending of messages to/from this command (also known as "inter-process communication")
     * should be enabled, and using which file descriptor number.
     *
     * If set, must be > 2.
     */
    ipc?: number;

    /**
     * Output command in raw format.
     */
    raw?: boolean;
}

export interface CloseEvent {
    command: CommandInfo;

    /**
     * The command's index among all commands ran.
     */
    index: number;

    /**
     * Whether the command exited because it was killed.
     */
    killed: boolean;

    /**
     * The exit code or signal for the command.
     */
    exitCode: string | number;

    timings: {
        startDate: Date;
        endDate: Date;
        durationSeconds: number;
    };
}

export interface TimerEvent {
    startDate: Date;
    endDate?: Date;
}

export interface MessageEvent {
    message: object;
    handle?: SendHandle;
}

interface OutgoingMessageEvent extends MessageEvent {
    options?: MessageOptions;
    onSent(error?: unknown): void;
}

/**
 * Subtype of NodeJS's child_process including only what's actually needed for a command to work.
 */
export type ChildProcess = EventEmitter &
    Pick<BaseChildProcess, 'pid' | 'stdin' | 'stdout' | 'stderr' | 'send'>;

/**
 * Interface for a function that must kill the process with `pid`, optionally sending `signal` to it.
 */
export type KillProcess = (pid: number, signal?: string) => void;

/**
 * Interface for a function that spawns a command and returns its child process instance.
 */
export type SpawnCommand = (command: string, options: SpawnOptions) => ChildProcess;

/**
 * The state of a command.
 *
 * - `stopped`: command was never started
 * - `started`: command is currently running
 * - `errored`: command failed spawning
 * - `exited`: command is not running anymore, e.g. it received a close event
 */
type CommandState = 'stopped' | 'started' | 'errored' | 'exited';

export class Command implements CommandInfo {
    private readonly killProcess: KillProcess;
    private readonly spawn: SpawnCommand;
    private readonly spawnOpts: SpawnOptions;
    readonly index: number;

    /** @inheritdoc */
    readonly name: string;

    /** @inheritdoc */
    readonly command: string;

    /** @inheritdoc */
    readonly prefixColor?: string;

    /** @inheritdoc */
    readonly env: Record<string, unknown>;

    /** @inheritdoc */
    readonly cwd?: string;

    /** @inheritdoc */
    readonly ipc?: number;

    readonly close = new Rx.Subject<CloseEvent>();
    readonly error = new Rx.Subject<unknown>();
    readonly stdout = new Rx.Subject<Buffer>();
    readonly stderr = new Rx.Subject<Buffer>();
    readonly timer = new Rx.Subject<TimerEvent>();
    readonly messages = {
        incoming: new Rx.Subject<MessageEvent>(),
        outgoing: new Rx.ReplaySubject<OutgoingMessageEvent>(),
    };

    process?: ChildProcess;

    // TODO: Should exit/error/stdio subscriptions be added here?
    private subscriptions: readonly Rx.Subscription[] = [];
    stdin?: Writable;
    pid?: number;
    killed = false;
    exited = false;

    state: CommandState = 'stopped';

    constructor(
        { index, name, command, prefixColor, env, cwd, ipc }: CommandInfo & { index: number },
        spawnOpts: SpawnOptions,
        spawn: SpawnCommand,
        killProcess: KillProcess,
    ) {
        this.index = index;
        this.name = name;
        this.command = command;
        this.prefixColor = prefixColor;
        this.env = env || {};
        this.cwd = cwd;
        this.ipc = ipc;
        this.killProcess = killProcess;
        this.spawn = spawn;
        this.spawnOpts = spawnOpts;
    }

    /**
     * Starts this command, piping output, error and close events onto the corresponding observables.
     */
    start() {
        const child = this.spawn(this.command, this.spawnOpts);
        this.state = 'started';
        this.process = child;
        this.pid = child.pid;
        const startDate = new Date(Date.now());
        const highResStartTime = process.hrtime();
        this.timer.next({ startDate });

        this.subscriptions = [...this.maybeSetupIPC(child)];
        Rx.fromEvent(child, 'error').subscribe((event) => {
            this.cleanUp();
            const endDate = new Date(Date.now());
            this.timer.next({ startDate, endDate });
            this.error.next(event);
            this.state = 'errored';
        });
        Rx.fromEvent(child, 'close')
            .pipe(Rx.map((event) => event as [number | null, NodeJS.Signals | null]))
            .subscribe(([exitCode, signal]) => {
                this.cleanUp();

                // Don't override error event
                if (this.state !== 'errored') {
                    this.state = 'exited';
                }

                const endDate = new Date(Date.now());
                this.timer.next({ startDate, endDate });
                const [durationSeconds, durationNanoSeconds] = process.hrtime(highResStartTime);
                this.close.next({
                    command: this,
                    index: this.index,
                    exitCode: exitCode ?? String(signal),
                    killed: this.killed,
                    timings: {
                        startDate,
                        endDate,
                        durationSeconds: durationSeconds + durationNanoSeconds / 1e9,
                    },
                });
            });
        child.stdout &&
            pipeTo(
                Rx.fromEvent(child.stdout, 'data').pipe(Rx.map((event) => event as Buffer)),
                this.stdout,
            );
        child.stderr &&
            pipeTo(
                Rx.fromEvent(child.stderr, 'data').pipe(Rx.map((event) => event as Buffer)),
                this.stderr,
            );
        this.stdin = child.stdin || undefined;
    }

    private maybeSetupIPC(child: ChildProcess) {
        if (!this.ipc) {
            return [];
        }

        return [
            pipeTo(
                Rx.fromEvent(child, 'message').pipe(
                    Rx.map((event) => {
                        const [message, handle] = event as [object, SendHandle | undefined];
                        return { message, handle };
                    }),
                ),
                this.messages.incoming,
            ),
            this.messages.outgoing.subscribe((message) => {
                if (!child.send) {
                    return message.onSent(new Error('Command does not have an IPC channel'));
                }

                child.send(message.message, message.handle, message.options, (error) => {
                    message.onSent(error);
                });
            }),
        ];
    }

    /**
     * Sends a message to the underlying process once it starts.
     *
     * @throws  If the command doesn't have an IPC channel enabled
     * @returns Promise that resolves when the message is sent,
     *          or rejects if it fails to deliver the message.
     */
    send(message: object, handle?: SendHandle, options?: MessageOptions): Promise<void> {
        if (this.ipc == null) {
            throw new Error('Command IPC is disabled');
        }
        return new Promise((resolve, reject) => {
            this.messages.outgoing.next({
                message,
                handle,
                options,
                onSent(error) {
                    error ? reject(error) : resolve();
                },
            });
        });
    }

    /**
     * Kills this command, optionally specifying a signal to send to it.
     */
    kill(code?: string) {
        if (Command.canKill(this)) {
            this.killed = true;
            this.killProcess(this.pid, code);
        }
    }

    private cleanUp() {
        this.subscriptions?.forEach((sub) => sub.unsubscribe());
        this.messages.outgoing = new Rx.ReplaySubject();
        this.process = undefined;
    }

    /**
     * Detects whether a command can be killed.
     *
     * Also works as a type guard on the input `command`.
     */
    static canKill(command: Command): command is Command & { pid: number; process: ChildProcess } {
        return !!command.pid && !!command.process;
    }
}

/**
 * Pipes all events emitted by `stream` into `subject`.
 */
function pipeTo<T>(stream: Rx.Observable<T>, subject: Rx.Subject<T>) {
    return stream.subscribe((event) => subject.next(event));
}

1	import {
2	ChildProcess as BaseChildProcess,
3	MessageOptions,
4	SendHandle,
5	SpawnOptions,
6	} from 'child_process';
7	import * as Rx from 'rxjs';	135✔
8	import { EventEmitter, Writable } from 'stream';
9
10	/**
11	* Identifier for a command; if string, it's the command's name, if number, it's the index.
12	*/
13	export type CommandIdentifier = string \| number;
14
15	export interface CommandInfo {
16	/**
17	* Command's name.
18	*/
19	name: string;
20
21	/**
22	* Which command line the command has.
23	*/
24	command: string;
25
26	/**
27	* Which environment variables should the spawned process have.
28	*/
29	env?: Record<string, unknown>;
30
31	/**
32	* The current working directory of the process when spawned.
33	*/
34	cwd?: string;
35
36	/**
37	* Color to use on prefix of the command.
38	*/
39	prefixColor?: string;
40
41	/**
42	* Whether sending of messages to/from this command (also known as "inter-process communication")
43	* should be enabled, and using which file descriptor number.
44	*
45	* If set, must be > 2.
46	*/
47	ipc?: number;
48
49	/**
50	* Output command in raw format.
51	*/
52	raw?: boolean;
53	}
54
55	export interface CloseEvent {
56	command: CommandInfo;
57
58	/**
59	* The command's index among all commands ran.
60	*/
61	index: number;
62
63	/**
64	* Whether the command exited because it was killed.
65	*/
66	killed: boolean;
67
68	/**
69	* The exit code or signal for the command.
70	*/
71	exitCode: string \| number;
72
73	timings: {
74	startDate: Date;
75	endDate: Date;
76	durationSeconds: number;
77	};
78	}
79
80	export interface TimerEvent {
81	startDate: Date;
82	endDate?: Date;
83	}
84
85	export interface MessageEvent {
86	message: object;
87	handle?: SendHandle;
88	}
89
90	interface OutgoingMessageEvent extends MessageEvent {
91	options?: MessageOptions;
92	onSent(error?: unknown): void;
93	}
94
95	/**
96	* Subtype of NodeJS's child_process including only what's actually needed for a command to work.
97	*/
98	export type ChildProcess = EventEmitter &
99	Pick<BaseChildProcess, 'pid' \| 'stdin' \| 'stdout' \| 'stderr' \| 'send'>;
100
101	/**
102	* Interface for a function that must kill the process with `pid`, optionally sending `signal` to it.
103	*/
104	export type KillProcess = (pid: number, signal?: string) => void;
105
106	/**
107	* Interface for a function that spawns a command and returns its child process instance.
108	*/
109	export type SpawnCommand = (command: string, options: SpawnOptions) => ChildProcess;
110
111	/**
112	* The state of a command.
113	*
114	* - `stopped`: command was never started
115	* - `started`: command is currently running
116	* - `errored`: command failed spawning
117	* - `exited`: command is not running anymore, e.g. it received a close event
118	*/
119	type CommandState = 'stopped' \| 'started' \| 'errored' \| 'exited';
120
121	export class Command implements CommandInfo {	1,017✔
122	private readonly killProcess: KillProcess;	1,443✔
123	private readonly spawn: SpawnCommand;	1,443✔
124	private readonly spawnOpts: SpawnOptions;	1,443✔
125	readonly index: number;	1,443✔
126
127	/** @inheritdoc */
128	readonly name: string;	1,443✔
129
130	/** @inheritdoc */
131	readonly command: string;	1,443✔
132
133	/** @inheritdoc */
134	readonly prefixColor?: string;	1,443✔
135
136	/** @inheritdoc */
137	readonly env: Record<string, unknown>;	1,443✔
138
139	/** @inheritdoc */
140	readonly cwd?: string;	1,443✔
141
142	/** @inheritdoc */
143	readonly ipc?: number;	1,443✔
144
145	readonly close = new Rx.Subject<CloseEvent>();	4,329✔
146	readonly error = new Rx.Subject<unknown>();	4,329✔
147	readonly stdout = new Rx.Subject<Buffer>();	4,329✔
148	readonly stderr = new Rx.Subject<Buffer>();	4,329✔
149	readonly timer = new Rx.Subject<TimerEvent>();	4,329✔
150	readonly messages = {	4,329✔
151	incoming: new Rx.Subject<MessageEvent>(),
152	outgoing: new Rx.ReplaySubject<OutgoingMessageEvent>(),
153	};
154
155	process?: ChildProcess;	1,443✔
156
157	// TODO: Should exit/error/stdio subscriptions be added here?
158	private subscriptions: readonly Rx.Subscription[] = [];	4,329✔
159	stdin?: Writable;	1,443✔
160	pid?: number;	1,443✔
161	killed = false;	4,329✔
162	exited = false;	4,329✔
163
164	state: CommandState = 'stopped';	4,329✔
165
166	constructor(
167	{ index, name, command, prefixColor, env, cwd, ipc }: CommandInfo & { index: number },
168	spawnOpts: SpawnOptions,
169	spawn: SpawnCommand,
170	killProcess: KillProcess,
171	) {
172	this.index = index;	4,329✔
173	this.name = name;	4,329✔
174	this.command = command;	4,329✔
175	this.prefixColor = prefixColor;	4,329✔
176	this.env = env \|\| {};	4,329✔
177	this.cwd = cwd;	4,329✔
178	this.ipc = ipc;	4,329✔
179	this.killProcess = killProcess;	4,329✔
180	this.spawn = spawn;	4,329✔
181	this.spawnOpts = spawnOpts;	4,329✔
182	}
183
184	/**
185	* Starts this command, piping output, error and close events onto the corresponding observables.
186	*/
187	start() {
188	const child = this.spawn(this.command, this.spawnOpts);	657✔
189	this.state = 'started';	657✔
190	this.process = child;	657✔
191	this.pid = child.pid;	657✔
192	const startDate = new Date(Date.now());	657✔
193	const highResStartTime = process.hrtime();	657✔
194	this.timer.next({ startDate });	657✔
195
196	this.subscriptions = [...this.maybeSetupIPC(child)];	657✔
197	Rx.fromEvent(child, 'error').subscribe((event) => {	657✔
198	this.cleanUp();	45✔
199	const endDate = new Date(Date.now());	45✔
200	this.timer.next({ startDate, endDate });	45✔
201	this.error.next(event);	45✔
202	this.state = 'errored';	45✔
203	});
204	Rx.fromEvent(child, 'close')	657✔
205	.pipe(Rx.map((event) => event as [number \| null, NodeJS.Signals \| null]))	180✔
206	.subscribe(([exitCode, signal]) => {
207	this.cleanUp();	180✔
208
209	// Don't override error event
210	if (this.state !== 'errored') {	180✔
211	this.state = 'exited';	171✔
212	}
213
214	const endDate = new Date(Date.now());	180✔
215	this.timer.next({ startDate, endDate });	180✔
216	const [durationSeconds, durationNanoSeconds] = process.hrtime(highResStartTime);	180✔
217	this.close.next({	180✔
218	command: this,
219	index: this.index,
220	exitCode: exitCode ?? String(signal),	309✔
221	killed: this.killed,
222	timings: {
223	startDate,
224	endDate,
225	durationSeconds: durationSeconds + durationNanoSeconds / 1e9,
226	},
227	});
228	});
229	child.stdout &&	657✔
230	pipeTo(
231	Rx.fromEvent(child.stdout, 'data').pipe(Rx.map((event) => event as Buffer)),	9✔
232	this.stdout,
233	);
234	child.stderr &&	657✔
235	pipeTo(
236	Rx.fromEvent(child.stderr, 'data').pipe(Rx.map((event) => event as Buffer)),	9✔
237	this.stderr,
238	);
239	this.stdin = child.stdin \|\| undefined;	657!
240	}
241
242	private maybeSetupIPC(child: ChildProcess) {
243	if (!this.ipc) {	657✔
244	return [];	612✔
245	}
246
247	return [	45✔
248	pipeTo(
249	Rx.fromEvent(child, 'message').pipe(
250	Rx.map((event) => {
251	const [message, handle] = event as [object, SendHandle \| undefined];	18✔
252	return { message, handle };	18✔
253	}),
254	),
255	this.messages.incoming,
256	),
257	this.messages.outgoing.subscribe((message) => {
258	if (!child.send) {	54✔
259	return message.onSent(new Error('Command does not have an IPC channel'));	9✔
260	}
261
262	child.send(message.message, message.handle, message.options, (error) => {	45✔
263	message.onSent(error);	18✔
264	});
265	}),
266	];
267	}
268
269	/**
270	* Sends a message to the underlying process once it starts.
271	*
272	* @throws If the command doesn't have an IPC channel enabled
273	* @returns Promise that resolves when the message is sent,
274	* or rejects if it fails to deliver the message.
275	*/
276	send(message: object, handle?: SendHandle, options?: MessageOptions): Promise<void> {
277	if (this.ipc == null) {	54✔
278	throw new Error('Command IPC is disabled');	9✔
279	}
280	return new Promise((resolve, reject) => {	45✔
281	this.messages.outgoing.next({	45✔
282	message,
283	handle,
284	options,
285	onSent(error) {
286	error ? reject(error) : resolve();	18✔
287	},
288	});
289	});
290	}
291
292	/**
293	* Kills this command, optionally specifying a signal to send to it.
294	*/
295	kill(code?: string) {
296	if (Command.canKill(this)) {	36✔
297	this.killed = true;	27✔
298	this.killProcess(this.pid, code);	27✔
299	}
300	}
301
302	private cleanUp() {
303	this.subscriptions?.forEach((sub) => sub.unsubscribe());	225!
304	this.messages.outgoing = new Rx.ReplaySubject();	225✔
305	this.process = undefined;	225✔
306	}
307
308	/**
309	* Detects whether a command can be killed.
310	*
311	* Also works as a type guard on the input `command`.
312	*/
313	static canKill(command: Command): command is Command & { pid: number; process: ChildProcess } {
314	return !!command.pid && !!command.process;	63✔
315	}
316	}
317
318	/**
319	* Pipes all events emitted by `stream` into `subject`.
320	*/
321	function pipeTo<T>(stream: Rx.Observable<T>, subject: Rx.Subject<T>) {
322	return stream.subscribe((event) => subject.next(event));	1,359✔
323	}

open-cli-tools / concurrently / 13468958302

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous