X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2Fabstract-worker.ts;h=15259106cfe59e1e73d7ec3a613da27a262c62a8;hb=04212af7b1a1494ff58da58c9e5b3397d25f18b7;hp=311a56ad60e00684835e89ef0507d60f0c9d4432;hpb=9a38f99e676160c0bc7d28fe88f27b01fa31b5a1;p=poolifier.git diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts index 311a56ad..15259106 100644 --- a/src/worker/abstract-worker.ts +++ b/src/worker/abstract-worker.ts @@ -1,4 +1,3 @@ -import { AsyncResource } from 'node:async_hooks' import type { Worker } from 'node:cluster' import type { MessagePort } from 'node:worker_threads' import { performance } from 'node:perf_hooks' @@ -56,7 +55,7 @@ export abstract class AbstractWorker< MainWorker extends Worker | MessagePort, Data = unknown, Response = unknown -> extends AsyncResource { +> { /** * Worker id. */ @@ -87,19 +86,18 @@ export abstract class AbstractWorker< * @param opts - Options for the worker. */ public constructor ( - type: string, protected readonly isMain: boolean, private readonly mainWorker: MainWorker, taskFunctions: TaskFunction | TaskFunctions, protected opts: WorkerOptions = DEFAULT_WORKER_OPTIONS ) { - super(type) if (this.isMain == null) { throw new Error('isMain parameter is mandatory') } this.checkTaskFunctions(taskFunctions) this.checkWorkerOptions(this.opts) if (!this.isMain) { + // Should be once() but Node.js on windows has a bug that prevents it from working this.getMainWorker().on('message', this.handleReadyMessage.bind(this)) } } @@ -367,21 +365,17 @@ export abstract class AbstractWorker< * * @param message - The kill message. */ - protected handleKillMessage (message: MessageValue): void { + protected handleKillMessage (_message: MessageValue): void { this.stopCheckActive() if (isAsyncFunction(this.opts.killHandler)) { (this.opts.killHandler?.() as Promise) .then(() => { this.sendToMainWorker({ kill: 'success' }) - return null + return undefined }) .catch(() => { this.sendToMainWorker({ kill: 'failure' }) }) - .finally(() => { - this.emitDestroy() - }) - .catch(EMPTY_FUNCTION) } else { try { // eslint-disable-next-line @typescript-eslint/no-invalid-void-type @@ -389,8 +383,6 @@ export abstract class AbstractWorker< this.sendToMainWorker({ kill: 'success' }) } catch { this.sendToMainWorker({ kill: 'failure' }) - } finally { - this.emitDestroy() } } } @@ -404,7 +396,7 @@ export abstract class AbstractWorker< private checkMessageWorkerId (message: MessageValue): void { if (message.workerId == null) { throw new Error('Message worker id is not set') - } else if (message.workerId != null && message.workerId !== this.id) { + } else if (message.workerId !== this.id) { throw new Error( `Message worker id ${message.workerId} does not match the worker id ${this.id}` ) @@ -492,8 +484,8 @@ export abstract class AbstractWorker< */ protected run (task: Task): void { const { name, taskId, data } = task - const fn = this.taskFunctions.get(name ?? DEFAULT_TASK_NAME) - if (fn == null) { + const taskFunctionName = name ?? DEFAULT_TASK_NAME + if (!this.taskFunctions.has(taskFunctionName)) { this.sendToMainWorker({ workerError: { name: name as string, @@ -504,10 +496,11 @@ export abstract class AbstractWorker< }) return } + const fn = this.taskFunctions.get(taskFunctionName) if (isAsyncFunction(fn)) { - this.runInAsyncScope(this.runAsync.bind(this), this, fn, task) + this.runAsync(fn as TaskAsyncFunction, task) } else { - this.runInAsyncScope(this.runSync.bind(this), this, fn, task) + this.runSync(fn as TaskSyncFunction, task) } }