X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2Fabstract-worker.ts;h=950738b26f0a47f16f0f1eadd93be5f759a8024d;hb=310de0aa5d5b3887a4de4ef6a0b42e7b7a560987;hp=493ca0b293fc145b2eae812f0e0dd858b9b4e566;hpb=07588f306649b9df235aaf65de4842e99532ed6a;p=poolifier.git diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts index 493ca0b2..950738b2 100644 --- a/src/worker/abstract-worker.ts +++ b/src/worker/abstract-worker.ts @@ -103,8 +103,8 @@ export abstract class AbstractWorker< this.opts.killBehavior = opts.killBehavior ?? DEFAULT_KILL_BEHAVIOR this.opts.maxInactiveTime = opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME - delete this.opts.async this.opts.killHandler = opts.killHandler ?? EMPTY_FUNCTION + delete this.opts.async } /** @@ -137,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' @@ -170,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) } @@ -191,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' @@ -208,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 @@ -227,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' @@ -239,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 } /** @@ -264,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' @@ -300,22 +320,25 @@ export abstract class AbstractWorker< protected messageListener (message: MessageValue): void { if (this.isMain) { throw new Error('Cannot handle message to worker in main worker') + } else if (message.workerId == null) { + throw new Error('Message worker id is not set') } else if (message.workerId != null && message.workerId !== this.id) { - throw new Error('Message worker id does not match the worker 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 - 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) { - // Kill message received - this.handleKillMessage(message) - } + throw new Error( + `Message worker id ${message.workerId} does not match the worker id ${this.id}` + ) + } + if (message.statistics != null) { + // Statistics message received + this.statistics = message.statistics + } else if (message.checkActive != null) { + // Check active message received + 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) { + // Kill message received + this.handleKillMessage(message) } } @@ -328,12 +351,27 @@ export abstract class AbstractWorker< this.stopCheckActive() if (isAsyncFunction(this.opts.killHandler)) { (this.opts.killHandler?.() as Promise) - .then(() => this.emitDestroy()) + .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 { - // eslint-disable-next-line @typescript-eslint/no-invalid-void-type - this.opts.killHandler?.() as void - this.emitDestroy() + 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() + } } } @@ -391,6 +429,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. * @@ -426,26 +474,27 @@ export abstract class AbstractWorker< fn: TaskSyncFunction, task: Task ): void { + const { name, taskId, data } = task try { - let taskPerformance = this.beginTaskPerformance(task.name) - const res = fn(task.data) + let taskPerformance = this.beginTaskPerformance(name) + const res = fn(data) taskPerformance = this.endTaskPerformance(taskPerformance) this.sendToMainWorker({ data: res, taskPerformance, workerId: this.id, - taskId: task.taskId + taskId }) } catch (e) { const errorMessage = this.handleError(e as Error | string) this.sendToMainWorker({ taskError: { - name: task.name ?? DEFAULT_TASK_NAME, + name: name ?? DEFAULT_TASK_NAME, message: errorMessage, - data: task.data + data }, workerId: this.id, - taskId: task.taskId + taskId }) } finally { this.updateLastTaskTimestamp() @@ -462,15 +511,16 @@ export abstract class AbstractWorker< fn: TaskAsyncFunction, task: Task ): void { - let taskPerformance = this.beginTaskPerformance(task.name) - fn(task.data) + const { name, taskId, data } = task + let taskPerformance = this.beginTaskPerformance(name) + fn(data) .then((res) => { taskPerformance = this.endTaskPerformance(taskPerformance) this.sendToMainWorker({ data: res, taskPerformance, workerId: this.id, - taskId: task.taskId + taskId }) return null }) @@ -478,12 +528,12 @@ export abstract class AbstractWorker< const errorMessage = this.handleError(e as Error | string) this.sendToMainWorker({ taskError: { - name: task.name ?? DEFAULT_TASK_NAME, + name: name ?? DEFAULT_TASK_NAME, message: errorMessage, - data: task.data + data }, workerId: this.id, - taskId: task.taskId + taskId }) }) .finally(() => {