X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2Fabstract-worker.ts;h=ff31fc7cb1879e19435c7942ca34c80b1a5a2051;hb=3c5dc3fbade52c4f8e707279cd3089c57210e241;hp=381e4d9a7357ec70ed425fa4ab9d3ae990a775a7;hpb=440dd7d77da6e027b3b9543504addcf805f76c1d;p=poolifier.git diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts index 381e4d9a..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,16 +318,20 @@ export abstract class AbstractWorker< * @param message - The received message. */ protected messageListener (message: MessageValue): void { - if (message.workerId === this.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() - } else if (message.id != null && message.data != null) { + message.checkActive ? this.startCheckActive() : this.stopCheckActive() + } else if (message.taskId != null && message.data != null) { // Task message received this.run(message) } else if (message.kill === true) { @@ -318,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() + } + } } /** @@ -376,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. * @@ -393,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) @@ -422,7 +481,7 @@ export abstract class AbstractWorker< data: res, taskPerformance, workerId: this.id, - id: task.id + taskId: task.taskId }) } catch (e) { const errorMessage = this.handleError(e as Error | string) @@ -433,7 +492,7 @@ export abstract class AbstractWorker< data: task.data }, workerId: this.id, - id: task.id + taskId: task.taskId }) } finally { this.updateLastTaskTimestamp() @@ -452,17 +511,17 @@ 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, taskPerformance, workerId: this.id, - id: task.id + taskId: task.taskId }) return null }) - .catch(e => { + .catch((e) => { const errorMessage = this.handleError(e as Error | string) this.sendToMainWorker({ taskError: { @@ -471,7 +530,7 @@ export abstract class AbstractWorker< data: task.data }, workerId: this.id, - id: task.id + taskId: task.taskId }) }) .finally(() => { @@ -527,7 +586,7 @@ export abstract class AbstractWorker< } private updateLastTaskTimestamp (): void { - if (!this.isMain && this.activeInterval != null) { + if (this.activeInterval != null) { this.lastTaskTimestamp = performance.now() } }