X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2Fabstract-worker.ts;h=df263b6d14ebba49768dbaea88a7e6e924f1a78d;hb=82888165d01dba6a57f5047d4a4c27c6da6025f3;hp=73eec1c6999d2aae78fa7e0a70038d30d6993997;hpb=d4aeae5aa9e260c8c2f6d28f3133de368552c108;p=poolifier.git diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts index 73eec1c6..df263b6d 100644 --- a/src/worker/abstract-worker.ts +++ b/src/worker/abstract-worker.ts @@ -1,11 +1,18 @@ import { AsyncResource } from 'node:async_hooks' import type { Worker } from 'node:cluster' import type { MessagePort } from 'node:worker_threads' -import type { MessageValue } from '../utility-types' +import { + type MessageValue, + type TaskFunctions, + type WorkerAsyncFunction, + type WorkerFunction, + type WorkerSyncFunction +} from '../utility-types' import { EMPTY_FUNCTION } from '../utils' import type { KillBehavior, WorkerOptions } from './worker-options' import { KillBehaviors } from './worker-options' +const DEFAULT_FUNCTION_NAME = 'default' const DEFAULT_MAX_INACTIVE_TIME = 60000 const DEFAULT_KILL_BEHAVIOR: KillBehavior = KillBehaviors.SOFT @@ -21,6 +28,10 @@ export abstract class AbstractWorker< Data = unknown, Response = unknown > extends AsyncResource { + /** + * Task function(s) processed by the worker when the pool's `execution` function is invoked. + */ + protected taskFunctions!: Map> /** * Timestamp of the last task processed by this worker. */ @@ -34,14 +45,16 @@ export abstract class AbstractWorker< * * @param type - The type of async event. * @param isMain - Whether this is the main worker or not. - * @param fn - Function processed by the worker when the pool's `execution` function is invoked. + * @param taskFunctions - Task function(s) processed by the worker when the pool's `execution` function is invoked. The first function is the default function. * @param mainWorker - Reference to main worker. * @param opts - Options for the worker. */ public constructor ( type: string, protected readonly isMain: boolean, - fn: (data: Data) => Response | Promise, + taskFunctions: + | WorkerFunction + | TaskFunctions, protected mainWorker: MainWorker | undefined | null, protected readonly opts: WorkerOptions = { /** @@ -57,7 +70,7 @@ export abstract class AbstractWorker< ) { super(type) this.checkWorkerOptions(this.opts) - this.checkFunctionInput(fn) + this.checkTaskFunctions(taskFunctions) if (!this.isMain) { this.lastTaskTimestamp = performance.now() this.aliveInterval = setInterval( @@ -67,30 +80,75 @@ export abstract class AbstractWorker< this.checkAlive.bind(this)() } - this.mainWorker?.on( - 'message', - (message: MessageValue) => { - this.messageListener(message, fn) + this.mainWorker?.on('message', this.messageListener.bind(this)) + } + + private checkWorkerOptions (opts: WorkerOptions): void { + this.opts.killBehavior = opts.killBehavior ?? DEFAULT_KILL_BEHAVIOR + this.opts.maxInactiveTime = + opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME + this.opts.async = opts.async ?? false + } + + /** + * Checks if the `taskFunctions` parameter is passed to the constructor. + * + * @param taskFunctions - The task function(s) parameter that should be checked. + */ + private checkTaskFunctions ( + taskFunctions: + | WorkerFunction + | TaskFunctions + ): void { + if (taskFunctions == null) { + throw new Error('taskFunctions parameter is mandatory') + } + if ( + typeof taskFunctions !== 'function' && + typeof taskFunctions !== 'object' + ) { + throw new Error('taskFunctions parameter is not a function or an object') + } + if ( + typeof taskFunctions === 'object' && + taskFunctions.constructor !== Object && + Object.prototype.toString.call(taskFunctions) !== '[object Object]' + ) { + throw new Error('taskFunctions parameter is not an object literal') + } + this.taskFunctions = new Map>() + if (typeof taskFunctions !== 'function') { + let firstEntry = true + for (const [name, fn] of Object.entries(taskFunctions)) { + if (typeof fn !== 'function') { + throw new Error( + 'A taskFunctions parameter object value is not a function' + ) + } + this.taskFunctions.set(name, fn.bind(this)) + if (firstEntry) { + this.taskFunctions.set(DEFAULT_FUNCTION_NAME, fn.bind(this)) + firstEntry = false + } } - ) + } else { + this.taskFunctions.set(DEFAULT_FUNCTION_NAME, taskFunctions.bind(this)) + } } /** * Worker message listener. * * @param message - Message received. - * @param fn - Function processed by the worker when the pool's `execution` function is invoked. */ - protected messageListener ( - message: MessageValue, - fn: (data: Data) => Response | Promise - ): void { + protected messageListener (message: MessageValue): void { if (message.id != null && message.data != null) { // Task message received - if (this.opts.async === true) { + const fn = this.getTaskFunction(message.name) + if (fn?.constructor.name === 'AsyncFunction') { this.runInAsyncScope(this.runAsync.bind(this), this, fn, message) } else { - this.runInAsyncScope(this.run.bind(this), this, fn, message) + this.runInAsyncScope(this.runSync.bind(this), this, fn, message) } } else if (message.parent != null) { // Main worker reference message received @@ -102,32 +160,6 @@ export abstract class AbstractWorker< } } - private checkWorkerOptions (opts: WorkerOptions): void { - this.opts.killBehavior = opts.killBehavior ?? DEFAULT_KILL_BEHAVIOR - this.opts.maxInactiveTime = - opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME - this.opts.async = opts.async ?? false - } - - /** - * Checks if the `fn` parameter is passed to the constructor. - * - * @param fn - The function that should be defined. - */ - private checkFunctionInput ( - fn: (data: Data) => Response | Promise - ): void { - if (fn == null) throw new Error('fn parameter is mandatory') - if (typeof fn !== 'function') { - throw new TypeError('fn parameter is not a function') - } - if (fn.constructor.name === 'AsyncFunction' && this.opts.async === false) { - throw new Error( - 'fn parameter is an async function, please set the async option to true' - ) - } - } - /** * Returns the main worker. * @@ -175,8 +207,8 @@ export abstract class AbstractWorker< * @param fn - Function that will be executed. * @param message - Input data for the given function. */ - protected run ( - fn: (data?: Data) => Response, + protected runSync ( + fn: WorkerSyncFunction, message: MessageValue ): void { try { @@ -203,7 +235,7 @@ export abstract class AbstractWorker< * @param message - Input data for the given function. */ protected runAsync ( - fn: (data?: Data) => Promise, + fn: WorkerAsyncFunction, message: MessageValue ): void { const startTimestamp = performance.now() @@ -226,4 +258,18 @@ export abstract class AbstractWorker< }) .catch(EMPTY_FUNCTION) } + + /** + * Gets the task function in the given scope. + * + * @param name - Name of the function that will be returned. + */ + private getTaskFunction (name?: string): WorkerFunction { + name = name ?? DEFAULT_FUNCTION_NAME + const fn = this.taskFunctions.get(name) + if (fn == null) { + throw new Error(`Task function "${name}" not found`) + } + return fn + } }