10578830901

Committed 27 Aug 2024 01:11PM UTC coverage: 89.276% (-0.1%) from 89.375%

Build # 10578830901

Build Type

Pull #2739

github

Committed by

web-flow

Commit Message

Merge 5c59ff929 into b8d888699

Pull Request Pull Request #2739: Fix/properly update data

Run Details

1168 of 1403 branches covered (83.25%)

Branch coverage included in aggregate %.

0 of 6 new or added lines in 1 file covered. (0.0%)

2 existing lines in 1 file now uncovered.

2262 of 2439 relevant lines covered (92.74%)

2959.04 hits per line

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

14.93

/src/classes/child-processor.ts

import { ParentCommand } from '../enums';
import { SandboxedJob } from '../interfaces';
import { JobJsonSandbox } from '../types';
import { errorToJSON } from '../utils';

enum ChildStatus {
  Idle,
  Started,
  Terminating,
  Errored,
}

/**
 * ChildProcessor
 *
 * This class acts as the interface between a child process and it parent process
 * so that jobs can be processed in different processes.
 *
 */
export class ChildProcessor {
  public status?: ChildStatus;
  public processor: any;
  public currentJobPromise: Promise<unknown> | undefined;

  constructor(private send: (msg: any) => Promise<void>) {}

  public async init(processorFile: string): Promise<void> {
    let processor;
    try {
      const { default: processorFn } = await import(processorFile);
      processor = processorFn;

      if (processor.default) {
        // support es2015 module.
        processor = processor.default;
      }

      if (typeof processor !== 'function') {
        throw new Error('No function is exported in processor file');
      }
    } catch (err) {
      this.status = ChildStatus.Errored;
      return this.send({
        cmd: ParentCommand.InitFailed,
        err: errorToJSON(err),
      });
    }

    const origProcessor = processor;
    processor = function (job: SandboxedJob, token?: string) {
      try {
        return Promise.resolve(origProcessor(job, token));
      } catch (err) {
        return Promise.reject(err);
      }
    };

    this.processor = processor;
    this.status = ChildStatus.Idle;
    await this.send({
      cmd: ParentCommand.InitCompleted,
    });
  }

  public async start(jobJson: JobJsonSandbox, token?: string): Promise<void> {
    if (this.status !== ChildStatus.Idle) {
      return this.send({
        cmd: ParentCommand.Error,
        err: errorToJSON(new Error('cannot start a not idling child process')),
      });
    }
    this.status = ChildStatus.Started;
    this.currentJobPromise = (async () => {
      try {
        const job = this.wrapJob(jobJson, this.send);
        const result = await this.processor(job, token);
        await this.send({
          cmd: ParentCommand.Completed,
          value: typeof result === 'undefined' ? null : result,
        });
      } catch (err) {
        await this.send({
          cmd: ParentCommand.Failed,
          value: errorToJSON(!(<Error>err).message ? new Error(<any>err) : err),
        });
      } finally {
        this.status = ChildStatus.Idle;
        this.currentJobPromise = undefined;
      }
    })();
  }

  public async stop(): Promise<void> {}

  async waitForCurrentJobAndExit(): Promise<void> {
    this.status = ChildStatus.Terminating;
    try {
      await this.currentJobPromise;
    } finally {
      process.exit(process.exitCode || 0);
    }
  }

  /**
   * Enhance the given job argument with some functions
   * that can be called from the sandboxed job processor.
   *
   * Note, the `job` argument is a JSON deserialized message
   * from the main node process to this forked child process,
   * the functions on the original job object are not in tact.
   * The wrapped job adds back some of those original functions.
   */
  protected wrapJob(
    job: JobJsonSandbox,
    send: (msg: any) => Promise<void>,
  ): SandboxedJob {
    const wrappedJob = {
      ...job,
      data: JSON.parse(job.data || '{}'),
      opts: job.opts,
      returnValue: JSON.parse(job.returnvalue || '{}'),
      /*
       * Emulate the real job `updateProgress` function, should works as `progress` function.
       */
      async updateProgress(progress: number | object) {
        // Locally store reference to new progress value
        // so that we can return it from this process synchronously.
        this.progress = progress;
        // Send message to update job progress.
        await send({
          cmd: ParentCommand.Progress,
          value: progress,
        });
      },
      /*
       * Emulate the real job `log` function.
       */
      log: async (row: any) => {
        await send({
          cmd: ParentCommand.Log,
          value: row,
        });
      },
      /*
       * Emulate the real job `moveToDelayed` function.
       */
      moveToDelayed: async (timestamp: number, token?: string) => {
        await send({
          cmd: ParentCommand.MoveToDelayed,
          value: { timestamp, token },
        });
      },
      /*
       * Emulate the real job `updateData` function.
       */
      updateData: async (data: any) => {
        await send({
          cmd: ParentCommand.Update,
          value: data,
        });
        wrappedJob.data = data;
      },
    };

    return wrappedJob;
  }
}

1	import { ParentCommand } from '../enums';	1✔
2	import { SandboxedJob } from '../interfaces';
3	import { JobJsonSandbox } from '../types';
4	import { errorToJSON } from '../utils';	1✔
5
6	enum ChildStatus {	1✔
7	Idle,	1✔
8	Started,	1✔
9	Terminating,	1✔
10	Errored,	1✔
11	}
12
13	/**
14	* ChildProcessor
15	*
16	* This class acts as the interface between a child process and it parent process
17	* so that jobs can be processed in different processes.
18	*
19	*/
20	export class ChildProcessor {	1✔
21	public status?: ChildStatus;
22	public processor: any;
23	public currentJobPromise: Promise<unknown> \| undefined;
24
25	constructor(private send: (msg: any) => Promise<void>) {}	×
26
27	public async init(processorFile: string): Promise<void> {
28	let processor;
29	try {	×
30	const { default: processorFn } = await import(processorFile);	×
31	processor = processorFn;	×
32
33	if (processor.default) {	×
34	// support es2015 module.
35	processor = processor.default;	×
36	}
37
38	if (typeof processor !== 'function') {	×
39	throw new Error('No function is exported in processor file');	×
40	}
41	} catch (err) {
42	this.status = ChildStatus.Errored;	×
43	return this.send({	×
44	cmd: ParentCommand.InitFailed,
45	err: errorToJSON(err),
46	});
47	}
48
49	const origProcessor = processor;	×
50	processor = function (job: SandboxedJob, token?: string) {	×
51	try {	×
52	return Promise.resolve(origProcessor(job, token));	×
53	} catch (err) {
54	return Promise.reject(err);	×
55	}
56	};
57
58	this.processor = processor;	×
59	this.status = ChildStatus.Idle;	×
60	await this.send({	×
61	cmd: ParentCommand.InitCompleted,
62	});
63	}
64
65	public async start(jobJson: JobJsonSandbox, token?: string): Promise<void> {
66	if (this.status !== ChildStatus.Idle) {	×
67	return this.send({	×
68	cmd: ParentCommand.Error,
69	err: errorToJSON(new Error('cannot start a not idling child process')),
70	});
71	}
72	this.status = ChildStatus.Started;	×
73	this.currentJobPromise = (async () => {	×
74	try {	×
75	const job = this.wrapJob(jobJson, this.send);	×
76	const result = await this.processor(job, token);	×
77	await this.send({	×
78	cmd: ParentCommand.Completed,
79	value: typeof result === 'undefined' ? null : result,	×
80	});
81	} catch (err) {
82	await this.send({	×
83	cmd: ParentCommand.Failed,
84	value: errorToJSON(!(<Error>err).message ? new Error(<any>err) : err),	×
85	});
86	} finally {
87	this.status = ChildStatus.Idle;	×
88	this.currentJobPromise = undefined;	×
89	}
90	})();
91	}
92
93	public async stop(): Promise<void> {}
94
95	async waitForCurrentJobAndExit(): Promise<void> {
96	this.status = ChildStatus.Terminating;	×
97	try {	×
98	await this.currentJobPromise;	×
99	} finally {
100	process.exit(process.exitCode \|\| 0);	×
101	}
102	}
103
104	/**
105	* Enhance the given job argument with some functions
106	* that can be called from the sandboxed job processor.
107	*
108	* Note, the `job` argument is a JSON deserialized message
109	* from the main node process to this forked child process,
110	* the functions on the original job object are not in tact.
111	* The wrapped job adds back some of those original functions.
112	*/
113	protected wrapJob(
114	job: JobJsonSandbox,
115	send: (msg: any) => Promise<void>,
116	): SandboxedJob {
NEW 117	const wrappedJob = {	×
118	...job,
119	data: JSON.parse(job.data \|\| '{}'),	×
120	opts: job.opts,
121	returnValue: JSON.parse(job.returnvalue \|\| '{}'),	×
122	/*
123	* Emulate the real job `updateProgress` function, should works as `progress` function.
124	*/
125	async updateProgress(progress: number \| object) {
126	// Locally store reference to new progress value
127	// so that we can return it from this process synchronously.
128	this.progress = progress;	×
129	// Send message to update job progress.
130	await send({	×
131	cmd: ParentCommand.Progress,
132	value: progress,
133	});
134	},
135	/*
136	* Emulate the real job `log` function.
137	*/
138	log: async (row: any) => {
NEW 139	await send({	×
140	cmd: ParentCommand.Log,
141	value: row,
142	});
143	},
144	/*
145	* Emulate the real job `moveToDelayed` function.
146	*/
147	moveToDelayed: async (timestamp: number, token?: string) => {
NEW 148	await send({	×
149	cmd: ParentCommand.MoveToDelayed,
150	value: { timestamp, token },
151	});
152	},
153	/*
154	* Emulate the real job `updateData` function.
155	*/
156	updateData: async (data: any) => {
NEW 157	await send({	×
158	cmd: ParentCommand.Update,
159	value: data,
160	});
NEW 161	wrappedJob.data = data;	×
162	},
163	};
164
NEW 165	return wrappedJob;	×
166	}
167	}

taskforcesh / bullmq / 10578830901

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