X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2Fabstract-worker.ts;h=ff31fc7cb1879e19435c7942ca34c80b1a5a2051;hb=3c5dc3fbade52c4f8e707279cd3089c57210e241;hp=64627efb8823198c6cb09c81fbdded3bef046f01;hpb=bc4f7ae28319fdd0f43134a39a9dc57e84f42935;p=poolifier.git diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts index 64627efb..ff31fc7c 100644 --- a/src/worker/abstract-worker.ts +++ b/src/worker/abstract-worker.ts @@ -84,7 +84,11 @@ export abstract class AbstractWorker< * The maximum time to keep this worker active while idle. * The pool automatically checks and terminates this worker when the time expires. */ - maxInactiveTime: DEFAULT_MAX_INACTIVE_TIME + maxInactiveTime: DEFAULT_MAX_INACTIVE_TIME, + /** + * The function to call when the worker is killed. + */ + killHandler: EMPTY_FUNCTION } ) { super(type) @@ -99,6 +103,7 @@ export abstract class AbstractWorker< this.opts.killBehavior = opts.killBehavior ?? DEFAULT_KILL_BEHAVIOR this.opts.maxInactiveTime = opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME + this.opts.killHandler = opts.killHandler ?? EMPTY_FUNCTION delete this.opts.async } @@ -119,7 +124,7 @@ export abstract class AbstractWorker< this.taskFunctions.set(DEFAULT_TASK_NAME, boundFn) this.taskFunctions.set( typeof taskFunctions.name === 'string' && - taskFunctions.name.trim().length > 0 + taskFunctions.name.trim().length > 0 ? taskFunctions.name : 'fn1', boundFn @@ -132,6 +137,11 @@ export abstract class AbstractWorker< 'A taskFunctions parameter object key is not a string' ) } + if (typeof name === 'string' && name.trim().length === 0) { + throw new TypeError( + 'A taskFunctions parameter object key an empty string' + ) + } if (typeof fn !== 'function') { throw new TypeError( 'A taskFunctions parameter object value is not a function' @@ -165,6 +175,9 @@ export abstract class AbstractWorker< if (typeof name !== 'string') { throw new TypeError('name parameter is not a string') } + if (typeof name === 'string' && name.trim().length === 0) { + throw new TypeError('name parameter is an empty string') + } return this.taskFunctions.has(name) } @@ -186,6 +199,9 @@ export abstract class AbstractWorker< if (typeof name !== 'string') { throw new TypeError('name parameter is not a string') } + if (typeof name === 'string' && name.trim().length === 0) { + throw new TypeError('name parameter is an empty string') + } if (name === DEFAULT_TASK_NAME) { throw new Error( 'Cannot add a task function with the default reserved name' @@ -203,6 +219,7 @@ export abstract class AbstractWorker< this.taskFunctions.set(DEFAULT_TASK_NAME, boundFn) } this.taskFunctions.set(name, boundFn) + this.sendTaskFunctionsListToMainWorker() return true } catch { return false @@ -222,6 +239,9 @@ export abstract class AbstractWorker< if (typeof name !== 'string') { throw new TypeError('name parameter is not a string') } + if (typeof name === 'string' && name.trim().length === 0) { + throw new TypeError('name parameter is an empty string') + } if (name === DEFAULT_TASK_NAME) { throw new Error( 'Cannot remove the task function with the default reserved name' @@ -234,7 +254,9 @@ export abstract class AbstractWorker< 'Cannot remove the task function used as the default task function' ) } - return this.taskFunctions.delete(name) + const deleteStatus = this.taskFunctions.delete(name) + this.sendTaskFunctionsListToMainWorker() + return deleteStatus } /** @@ -259,6 +281,9 @@ export abstract class AbstractWorker< if (typeof name !== 'string') { throw new TypeError('name parameter is not a string') } + if (typeof name === 'string' && name.trim().length === 0) { + throw new TypeError('name parameter is an empty string') + } if (name === DEFAULT_TASK_NAME) { throw new Error( 'Cannot set the default task function reserved name as the default task function' @@ -293,17 +318,19 @@ export abstract class AbstractWorker< * @param message - The received message. */ protected messageListener (message: MessageValue): void { - if (message.workerId != null && message.workerId !== this.id) { - throw new Error('Message worker id does not match worker id') + if (this.isMain) { + throw new Error('Cannot handle message to worker in main worker') + } else if (message.workerId != null && message.workerId !== this.id) { + throw new Error( + `Message worker id ${message.workerId} does not match the worker id ${this.id}` + ) } else if (message.workerId === this.id) { if (message.statistics != null) { // Statistics message received this.statistics = message.statistics } else if (message.checkActive != null) { // Check active message received - !this.isMain && message.checkActive - ? this.startCheckActive() - : this.stopCheckActive() + message.checkActive ? this.startCheckActive() : this.stopCheckActive() } else if (message.taskId != null && message.data != null) { // Task message received this.run(message) @@ -320,8 +347,31 @@ export abstract class AbstractWorker< * @param message - The kill message. */ protected handleKillMessage (message: MessageValue): void { - !this.isMain && this.stopCheckActive() - this.emitDestroy() + this.stopCheckActive() + if (isAsyncFunction(this.opts.killHandler)) { + (this.opts.killHandler?.() as Promise) + .then(() => { + this.sendToMainWorker({ kill: 'success', workerId: this.id }) + return null + }) + .catch(() => { + this.sendToMainWorker({ kill: 'failure', workerId: this.id }) + }) + .finally(() => { + this.emitDestroy() + }) + .catch(EMPTY_FUNCTION) + } else { + try { + // eslint-disable-next-line @typescript-eslint/no-invalid-void-type + this.opts.killHandler?.() as void + this.sendToMainWorker({ kill: 'success', workerId: this.id }) + } catch { + this.sendToMainWorker({ kill: 'failure', workerId: this.id }) + } finally { + this.emitDestroy() + } + } } /** @@ -378,6 +428,16 @@ export abstract class AbstractWorker< message: MessageValue ): void + /** + * Sends the list of task function names to the main worker. + */ + protected sendTaskFunctionsListToMainWorker (): void { + this.sendToMainWorker({ + taskFunctions: this.listTaskFunctions(), + workerId: this.id + }) + } + /** * Handles an error and convert it to a string so it can be sent back to the main worker. * @@ -395,9 +455,6 @@ export abstract class AbstractWorker< * @throws {@link https://nodejs.org/api/errors.html#class-error} If the task function is not found. */ protected run (task: Task): void { - if (this.isMain) { - throw new Error('Cannot run a task in the main worker') - } const fn = this.getTaskFunction(task.name) if (isAsyncFunction(fn)) { this.runInAsyncScope(this.runAsync.bind(this), this, fn, task) @@ -454,7 +511,7 @@ export abstract class AbstractWorker< ): void { let taskPerformance = this.beginTaskPerformance(task.name) fn(task.data) - .then(res => { + .then((res) => { taskPerformance = this.endTaskPerformance(taskPerformance) this.sendToMainWorker({ data: res, @@ -464,7 +521,7 @@ export abstract class AbstractWorker< }) return null }) - .catch(e => { + .catch((e) => { const errorMessage = this.handleError(e as Error | string) this.sendToMainWorker({ taskError: { @@ -529,7 +586,7 @@ export abstract class AbstractWorker< } private updateLastTaskTimestamp (): void { - if (!this.isMain && this.activeInterval != null) { + if (this.activeInterval != null) { this.lastTaskTimestamp = performance.now() } }