X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2Fthread-worker.ts;h=638de2ba2f9c8f090360c528cd1169cfcd52e823;hb=eceaf36f69c97b0376194a849e77f6ea15fb8f7b;hp=4d4c1c3479f14970db523ff96d873f0b4cdac79f;hpb=984dc9c8adc66d4ed982e76391db751dd685f0b2;p=poolifier.git diff --git a/src/worker/thread-worker.ts b/src/worker/thread-worker.ts index 4d4c1c34..638de2ba 100644 --- a/src/worker/thread-worker.ts +++ b/src/worker/thread-worker.ts @@ -4,10 +4,10 @@ import { parentPort, threadId } from 'node:worker_threads' -import type { MessageValue } from '../utility-types' -import { AbstractWorker } from './abstract-worker' -import type { WorkerOptions } from './worker-options' -import type { TaskFunctions, WorkerFunction } from './worker-functions' +import type { MessageValue } from '../utility-types.js' +import { AbstractWorker } from './abstract-worker.js' +import type { WorkerOptions } from './worker-options.js' +import type { TaskFunction, TaskFunctions } from './task-functions.js' /** * A thread worker used by a poolifier `ThreadPool`. @@ -28,9 +28,10 @@ export class ThreadWorker< Response = unknown > extends AbstractWorker { /** - * Message port used to communicate with the main thread. + * Message port used to communicate with the main worker. */ - private port!: MessagePort + private port?: MessagePort + /** * Constructs a new poolifier thread worker. * @@ -38,36 +39,37 @@ export class ThreadWorker< * @param opts - Options for the worker. */ public constructor ( - taskFunctions: - | WorkerFunction - | TaskFunctions, + taskFunctions: TaskFunction | TaskFunctions, opts: WorkerOptions = {} ) { - super( - 'worker-thread-pool:poolifier', - isMainThread, - parentPort as MessagePort, - taskFunctions, - opts - ) + super(isMainThread, parentPort, taskFunctions, opts) } /** @inheritDoc */ protected handleReadyMessage (message: MessageValue): void { if ( - !this.isMain && message.workerId === this.id && - message.ready != null && + message.ready === false && message.port != null ) { - this.port = message.port - this.port.on('message', this.messageListener.bind(this)) - this.sendToMainWorker({ ready: true, workerId: this.id }) + try { + this.port = message.port + this.port.on('message', this.messageListener.bind(this)) + this.sendToMainWorker({ + ready: true, + taskFunctionNames: this.listTaskFunctionNames() + }) + } catch { + this.sendToMainWorker({ + ready: false, + taskFunctionNames: this.listTaskFunctionNames() + }) + } } } /** @inheritDoc */ - protected handleKillMessage (message: MessageValue): void { + protected handleKillMessage (message: MessageValue): void { super.handleKillMessage(message) this.port?.unref() this.port?.close() @@ -79,12 +81,20 @@ export class ThreadWorker< } /** @inheritDoc */ - protected sendToMainWorker (message: MessageValue): void { - this.port.postMessage(message) + protected readonly sendToMainWorker = ( + message: MessageValue + ): void => { + this.port?.postMessage({ + ...message, + workerId: this.id + } satisfies MessageValue) } - /** @inheritDoc */ - protected handleError (e: Error | string): string { - return e as string + /** + * @inheritDoc + * @override + */ + protected handleError (error: Error | string): string { + return error as string } }