X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2FWorkerSet.ts;h=270f57b1f100c8ab80c0e3fe56fb9bebdb3adf24;hb=628c30e510dd0f12be4432bd5eb0ff86790c4de0;hp=9efc88c1b3e76e53d5a58a09f2b06c1c3bb31436;hpb=810f4caf2bf0b12d5c429d1d21af62b1fe67f965;p=e-mobility-charging-stations-simulator.git diff --git a/src/worker/WorkerSet.ts b/src/worker/WorkerSet.ts index 9efc88c1..270f57b1 100644 --- a/src/worker/WorkerSet.ts +++ b/src/worker/WorkerSet.ts @@ -1,12 +1,11 @@ // Partial Copyright Jerome Benoit. 2021-2023. All Rights Reserved. import EventEmitterAsyncResource from 'node:events'; -import { Worker } from 'node:worker_threads'; +import { SHARE_ENV, Worker } from 'node:worker_threads'; import { WorkerAbstract } from './WorkerAbstract'; import { WorkerConstants } from './WorkerConstants'; import { - type MessageHandler, type SetInfo, type WorkerData, WorkerMessageEvents, @@ -14,28 +13,40 @@ import { type WorkerSetElement, WorkerSetEvents, } from './WorkerTypes'; -import { defaultErrorHandler, defaultExitHandler, sleep } from './WorkerUtils'; +import { sleep } from './WorkerUtils'; export class WorkerSet extends WorkerAbstract { public readonly emitter: EventEmitterAsyncResource; private readonly workerSet: Set; /** - * Create a new `WorkerSet`. + * Creates a new `WorkerSet`. * * @param workerScript - * @param workerOptions - */ constructor(workerScript: string, workerOptions?: WorkerOptions) { super(workerScript, workerOptions); + this.workerOptions.poolOptions = { + ...{ + enableEvents: true, + restartWorkerOnError: true, + }, + ...this.workerOptions.poolOptions, + }; this.workerSet = new Set(); - this.emitter = new EventEmitterAsyncResource(); + if (this.workerOptions?.poolOptions?.enableEvents) { + this.emitter = new EventEmitterAsyncResource(); + } } get info(): SetInfo { return { + version: WorkerConstants.version, + type: 'set', + worker: 'thread', size: this.size, - runningElements: [...this.workerSet].reduce( + elementsExecuting: [...this.workerSet].reduce( (accumulator, workerSetElement) => accumulator + workerSetElement.numberOfWorkerElements, 0 ), @@ -84,23 +95,38 @@ export class WorkerSet extends WorkerAbstract { } /** - * Add a new `WorkerSetElement`. + * Adds a new `WorkerSetElement`. */ private addWorkerSetElement(): WorkerSetElement { - const worker = new Worker(this.workerScript); + const worker = new Worker(this.workerScript, { + env: SHARE_ENV, + ...this.workerOptions.poolOptions.workerOptions, + }); worker.on( 'message', - (this.workerOptions?.messageHandler ?? WorkerConstants.EMPTY_FUNCTION).bind( - this - ) as MessageHandler + this.workerOptions?.poolOptions?.messageHandler ?? WorkerConstants.EMPTY_FUNCTION + ); + worker.on( + 'error', + this.workerOptions?.poolOptions?.errorHandler ?? WorkerConstants.EMPTY_FUNCTION ); - worker.on('error', defaultErrorHandler.bind(this) as (err: Error) => void); worker.on('error', (error) => { - this.emitter.emit(WorkerSetEvents.error, error); - this.addWorkerSetElement(); + if (this.emitter !== undefined) { + this.emitter.emit(WorkerSetEvents.error, error); + } + if (this.workerOptions?.poolOptions?.restartWorkerOnError) { + this.addWorkerSetElement(); + } }); - worker.on('exit', defaultExitHandler.bind(this) as (exitCode: number) => void); - worker.on('exit', () => this.workerSet.delete(this.getWorkerSetElementByWorker(worker))); + worker.on( + 'online', + this.workerOptions?.poolOptions?.onlineHandler ?? WorkerConstants.EMPTY_FUNCTION + ); + worker.on( + 'exit', + this.workerOptions?.poolOptions?.exitHandler ?? WorkerConstants.EMPTY_FUNCTION + ); + worker.once('exit', () => this.workerSet.delete(this.getWorkerSetElementByWorker(worker))); const workerSetElement: WorkerSetElement = { worker, numberOfWorkerElements: 0 }; this.workerSet.add(workerSetElement); return workerSetElement;