knowledge/node_modules/bullmq/dist/esm/classes/child.js

import { fork } from 'child_process';
import { createServer } from 'net';
import { Worker } from 'worker_threads';
import { ChildCommand, ParentCommand } from '../enums';
import { EventEmitter } from 'events';
/**
 * @see https://nodejs.org/api/process.html#process_exit_codes
 */
const exitCodesErrors = {
    1: 'Uncaught Fatal Exception',
    2: 'Unused',
    3: 'Internal JavaScript Parse Error',
    4: 'Internal JavaScript Evaluation Failure',
    5: 'Fatal Error',
    6: 'Non-function Internal Exception Handler',
    7: 'Internal Exception Handler Run-Time Failure',
    8: 'Unused',
    9: 'Invalid Argument',
    10: 'Internal JavaScript Run-Time Failure',
    12: 'Invalid Debug Argument',
    13: 'Unfinished Top-Level Await',
};
/**
 * Child class
 *
 * This class is used to create a child process or worker thread, and allows using
 * isolated processes or threads for processing jobs.
 *
 */
export class Child extends EventEmitter {
    constructor(mainFile, processFile, opts = {
        useWorkerThreads: false,
    }) {
        super();
        this.mainFile = mainFile;
        this.processFile = processFile;
        this.opts = opts;
        this._exitCode = null;
        this._signalCode = null;
        this._killed = false;
    }
    get pid() {
        if (this.childProcess) {
            return this.childProcess.pid;
        }
        else if (this.worker) {
            // Worker threads pids can become negative when they are terminated
            // so we need to use the absolute value to index the retained object
            return Math.abs(this.worker.threadId);
        }
        else {
            throw new Error('No child process or worker thread');
        }
    }
    get exitCode() {
        return this._exitCode;
    }
    get signalCode() {
        return this._signalCode;
    }
    get killed() {
        if (this.childProcess) {
            return this.childProcess.killed;
        }
        return this._killed;
    }
    async init() {
        const execArgv = await convertExecArgv(process.execArgv);
        let parent;
        if (this.opts.useWorkerThreads) {
            this.worker = parent = new Worker(this.mainFile, Object.assign({ execArgv, stdin: true, stdout: true, stderr: true }, (this.opts.workerThreadsOptions
                ? this.opts.workerThreadsOptions
                : {})));
        }
        else {
            this.childProcess = parent = fork(this.mainFile, [], Object.assign({ execArgv, stdio: 'pipe' }, (this.opts.workerForkOptions ? this.opts.workerForkOptions : {})));
        }
        parent.on('exit', (exitCode, signalCode) => {
            this._exitCode = exitCode;
            // Coerce to null if undefined for backwards compatibility
            signalCode = typeof signalCode === 'undefined' ? null : signalCode;
            this._signalCode = signalCode;
            this._killed = true;
            this.emit('exit', exitCode, signalCode);
            // Clean all listeners, we do not expect any more events after "exit"
            parent.removeAllListeners();
            this.removeAllListeners();
        });
        parent.on('error', (...args) => this.emit('error', ...args));
        parent.on('message', (...args) => this.emit('message', ...args));
        parent.on('close', (...args) => this.emit('close', ...args));
        parent.stdout.pipe(process.stdout);
        parent.stderr.pipe(process.stderr);
        await this.initChild();
    }
    async send(msg) {
        return new Promise((resolve, reject) => {
            if (this.childProcess) {
                this.childProcess.send(msg, (err) => {
                    if (err) {
                        reject(err);
                    }
                    else {
                        resolve();
                    }
                });
            }
            else if (this.worker) {
                resolve(this.worker.postMessage(msg));
            }
            else {
                resolve();
            }
        });
    }
    killProcess(signal = 'SIGKILL') {
        if (this.childProcess) {
            this.childProcess.kill(signal);
        }
        else if (this.worker) {
            this.worker.terminate();
        }
    }
    async kill(signal = 'SIGKILL', timeoutMs) {
        if (this.hasProcessExited()) {
            return;
        }
        const onExit = onExitOnce(this.childProcess || this.worker);
        this.killProcess(signal);
        if (timeoutMs !== undefined && (timeoutMs === 0 || isFinite(timeoutMs))) {
            const timeoutHandle = setTimeout(() => {
                if (!this.hasProcessExited()) {
                    this.killProcess('SIGKILL');
                }
            }, timeoutMs);
            await onExit;
            clearTimeout(timeoutHandle);
        }
        await onExit;
    }
    async initChild() {
        const onComplete = new Promise((resolve, reject) => {
            const onMessageHandler = (msg) => {
                if (!Object.values(ParentCommand).includes(msg.cmd)) {
                    return;
                }
                if (msg.cmd === ParentCommand.InitCompleted) {
                    resolve();
                }
                else if (msg.cmd === ParentCommand.InitFailed) {
                    const err = new Error();
                    err.stack = msg.err.stack;
                    err.message = msg.err.message;
                    reject(err);
                }
                this.off('message', onMessageHandler);
                this.off('close', onCloseHandler);
            };
            const onCloseHandler = (code, signal) => {
                if (code > 128) {
                    code -= 128;
                }
                const msg = exitCodesErrors[code] || `Unknown exit code ${code}`;
                reject(new Error(`Error initializing child: ${msg} and signal ${signal}`));
                this.off('message', onMessageHandler);
                this.off('close', onCloseHandler);
            };
            this.on('message', onMessageHandler);
            this.on('close', onCloseHandler);
        });
        await this.send({
            cmd: ChildCommand.Init,
            value: this.processFile,
        });
        await onComplete;
    }
    hasProcessExited() {
        return !!(this.exitCode !== null || this.signalCode);
    }
}
function onExitOnce(child) {
    return new Promise(resolve => {
        child.once('exit', () => resolve());
    });
}
const getFreePort = async () => {
    return new Promise(resolve => {
        const server = createServer();
        server.listen(0, () => {
            const { port } = server.address();
            server.close(() => resolve(port));
        });
    });
};
const convertExecArgv = async (execArgv) => {
    const standard = [];
    const convertedArgs = [];
    for (let i = 0; i < execArgv.length; i++) {
        const arg = execArgv[i];
        if (arg.indexOf('--inspect') === -1) {
            standard.push(arg);
        }
        else {
            const argName = arg.split('=')[0];
            const port = await getFreePort();
            convertedArgs.push(`${argName}=${port}`);
        }
    }
    return standard.concat(convertedArgs);
};
//# sourceMappingURL=child.js.map