X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;ds=inline;f=src%2Fworker%2Fabstract-worker.ts;h=ebbefc75ba4cc8c4e09bcc3073e372085466c181;hb=refs%2Fheads%2Fmaster;hp=7803ce2eb836b003159760ce408f97fb3acedd50;hpb=6e5d7052fe741b50e68f8614b33d3754be41415f;p=poolifier.git diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts index 7803ce2eb..f3c36bceb 100644 --- a/src/worker/abstract-worker.ts +++ b/src/worker/abstract-worker.ts @@ -1,7 +1,8 @@ import type { Worker } from 'node:cluster' -import { performance } from 'node:perf_hooks' import type { MessagePort } from 'node:worker_threads' +import { performance } from 'node:perf_hooks' + import type { MessageValue, Task, @@ -9,13 +10,6 @@ import type { TaskPerformance, WorkerStatistics, } from '../utility-types.js' -import { - buildTaskFunctionProperties, - DEFAULT_TASK_NAME, - EMPTY_FUNCTION, - isAsyncFunction, - isPlainObject, -} from '../utils.js' import type { TaskAsyncFunction, TaskFunction, @@ -24,6 +18,15 @@ import type { TaskFunctions, TaskSyncFunction, } from './task-functions.js' + +import { + buildTaskFunctionProperties, + DEFAULT_TASK_NAME, + EMPTY_FUNCTION, + isAsyncFunction, + isPlainObject, +} from '../utils.js' +import { AbortError } from './abort-error.js' import { checkTaskFunctionName, checkValidTaskFunctionObjectEntry, @@ -32,21 +35,21 @@ import { import { KillBehaviors, type WorkerOptions } from './worker-options.js' const DEFAULT_MAX_INACTIVE_TIME = 60000 -const DEFAULT_WORKER_OPTIONS: WorkerOptions = { +const DEFAULT_WORKER_OPTIONS: Readonly = Object.freeze({ /** * The kill behavior option on this worker or its default value. */ killBehavior: KillBehaviors.SOFT, + /** + * The function to call when the worker is killed. + */ + killHandler: EMPTY_FUNCTION, /** * 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, - /** - * The function to call when the worker is killed. - */ - killHandler: EMPTY_FUNCTION, -} +}) /** * Base class that implements some shared logic for all poolifier workers. @@ -55,42 +58,51 @@ const DEFAULT_WORKER_OPTIONS: WorkerOptions = { * @typeParam Response - Type of response the worker sends back to the main worker. This can only be structured-cloneable data. */ export abstract class AbstractWorker< - MainWorker extends Worker | MessagePort, + MainWorker extends MessagePort | Worker, Data = unknown, Response = unknown > { /** - * Worker id. + * Handler id of the `activeInterval` worker activity check. */ - protected abstract id: number + protected activeInterval?: NodeJS.Timeout /** - * Task function object(s) processed by the worker when the pool's `execution` function is invoked. + * Worker id. */ - protected taskFunctions!: Map> + protected abstract readonly id: number /** * Timestamp of the last task processed by this worker. */ protected lastTaskTimestamp!: number + /** * Performance statistics computation requirements. */ protected statistics?: WorkerStatistics + /** - * Handler id of the `activeInterval` worker activity check. + * Task abort functions processed by the worker when task operation 'abort' is received. */ - // eslint-disable-next-line no-undef - protected activeInterval?: NodeJS.Timeout + protected taskAbortFunctions: Map< + `${string}-${string}-${string}-${string}-${string}`, + () => void + > + + /** + * Task function object(s) processed by the worker when the pool's `execute` method is invoked. + */ + protected taskFunctions!: Map> /** * Constructs a new poolifier worker. * @param isMain - Whether this is the main worker or not. * @param mainWorker - Reference to main worker. - * @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 taskFunctions - Task function(s) processed by the worker when the pool's `execute` method is invoked. The first function is the default function. * @param opts - Options for the worker. */ public constructor ( protected readonly isMain: boolean | undefined, - private readonly mainWorker: MainWorker | undefined | null, + private readonly mainWorker: MainWorker | null | undefined, taskFunctions: TaskFunction | TaskFunctions, protected opts: WorkerOptions = DEFAULT_WORKER_OPTIONS ) { @@ -98,6 +110,10 @@ export abstract class AbstractWorker< throw new Error('isMain parameter is mandatory') } this.checkTaskFunctions(taskFunctions) + this.taskAbortFunctions = new Map< + `${string}-${string}-${string}-${string}-${string}`, + () => void + >() this.checkWorkerOptions(this.opts) if (!this.isMain) { // Should be once() but Node.js on windows has a bug that prevents it from working @@ -105,76 +121,6 @@ export abstract class AbstractWorker< } } - private checkWorkerOptions (opts: WorkerOptions): void { - checkValidWorkerOptions(opts) - this.opts = { ...DEFAULT_WORKER_OPTIONS, ...opts } - } - - /** - * Checks if the `taskFunctions` parameter is passed to the constructor and valid. - * @param taskFunctions - The task function(s) parameter that should be checked. - */ - private checkTaskFunctions ( - taskFunctions: - | TaskFunction - | TaskFunctions - | undefined - ): void { - if (taskFunctions == null) { - throw new Error('taskFunctions parameter is mandatory') - } - this.taskFunctions = new Map>() - if (typeof taskFunctions === 'function') { - const fnObj = { taskFunction: taskFunctions.bind(this) } - this.taskFunctions.set(DEFAULT_TASK_NAME, fnObj) - this.taskFunctions.set( - typeof taskFunctions.name === 'string' && - taskFunctions.name.trim().length > 0 - ? taskFunctions.name - : 'fn1', - fnObj - ) - } else if (isPlainObject(taskFunctions)) { - let firstEntry = true - for (let [name, fnObj] of Object.entries(taskFunctions)) { - if (typeof fnObj === 'function') { - fnObj = { taskFunction: fnObj } satisfies TaskFunctionObject< - Data, - Response - > - } - checkValidTaskFunctionObjectEntry(name, fnObj) - fnObj.taskFunction = fnObj.taskFunction.bind(this) - if (firstEntry) { - this.taskFunctions.set(DEFAULT_TASK_NAME, fnObj) - firstEntry = false - } - this.taskFunctions.set(name, fnObj) - } - if (firstEntry) { - throw new Error('taskFunctions parameter object is empty') - } - } else { - throw new TypeError( - 'taskFunctions parameter is not a function or a plain object' - ) - } - } - - /** - * Checks if the worker has a task function with the given name. - * @param name - The name of the task function to check. - * @returns Whether the worker has a task function with the given name or not. - */ - public hasTaskFunction (name: string): TaskFunctionOperationResult { - try { - checkTaskFunctionName(name) - } catch (error) { - return { status: false, error: error as Error } - } - return { status: this.taskFunctions.has(name) } - } - /** * Adds a task function to the worker. * If a task function with the same name already exists, it is replaced. @@ -208,37 +154,22 @@ export abstract class AbstractWorker< this.sendTaskFunctionsPropertiesToMainWorker() return { status: true } } catch (error) { - return { status: false, error: error as Error } + return { error: error as Error, status: false } } } /** - * Removes a task function from the worker. - * @param name - The name of the task function to remove. - * @returns Whether the task function existed and was removed or not. + * Checks if the worker has a task function with the given name. + * @param name - The name of the task function to check. + * @returns Whether the worker has a task function with the given name or not. */ - public removeTaskFunction (name: string): TaskFunctionOperationResult { + public hasTaskFunction (name: string): TaskFunctionOperationResult { try { checkTaskFunctionName(name) - if (name === DEFAULT_TASK_NAME) { - throw new Error( - 'Cannot remove the task function with the default reserved name' - ) - } - if ( - this.taskFunctions.get(name) === - this.taskFunctions.get(DEFAULT_TASK_NAME) - ) { - throw new Error( - 'Cannot remove the task function used as the default task function' - ) - } - const deleteStatus = this.taskFunctions.delete(name) - this.sendTaskFunctionsPropertiesToMainWorker() - return { status: deleteStatus } } catch (error) { - return { status: false, error: error as Error } + return { error: error as Error, status: false } } + return { status: this.taskFunctions.has(name) } } /** @@ -276,6 +207,35 @@ export abstract class AbstractWorker< ] } + /** + * Removes a task function from the worker. + * @param name - The name of the task function to remove. + * @returns Whether the task function existed and was removed or not. + */ + public removeTaskFunction (name: string): TaskFunctionOperationResult { + try { + checkTaskFunctionName(name) + if (name === DEFAULT_TASK_NAME) { + throw new Error( + 'Cannot remove the task function with the default reserved name' + ) + } + if ( + this.taskFunctions.get(name) === + this.taskFunctions.get(DEFAULT_TASK_NAME) + ) { + throw new Error( + 'Cannot remove the task function used as the default task function' + ) + } + const deleteStatus = this.taskFunctions.delete(name) + this.sendTaskFunctionsPropertiesToMainWorker() + return { status: deleteStatus } + } catch (error) { + return { error: error as Error, status: false } + } + } + /** * Sets the default task function to use in the worker. * @param name - The name of the task function to use as default task function. @@ -299,52 +259,69 @@ export abstract class AbstractWorker< this.sendTaskFunctionsPropertiesToMainWorker() return { status: true } } catch (error) { - return { status: false, error: error as Error } + return { error: error as Error, status: false } } } /** - * Handles the ready message sent by the main worker. - * @param message - The ready message. + * Returns the main worker. + * @returns Reference to the main worker. + * @throws {@link https://nodejs.org/api/errors.html#class-error} If the main worker is not set. */ - protected abstract handleReadyMessage (message: MessageValue): void + protected getMainWorker (): MainWorker { + if (this.mainWorker == null) { + throw new Error('Main worker not set') + } + return this.mainWorker + } /** - * Worker message listener. - * @param message - The received message. + * Handles a worker error. + * @param error - The error raised by the worker. + * @returns The worker error object. */ - protected messageListener (message: MessageValue): void { - this.checkMessageWorkerId(message) - const { - statistics, - checkActive, - taskFunctionOperation, - taskId, - data, - kill, - } = message - if (statistics != null) { - // Statistics message received - this.statistics = statistics - } else if (checkActive != null) { - // Check active message received - checkActive ? this.startCheckActive() : this.stopCheckActive() - } else if (taskFunctionOperation != null) { - // Task function operation message received - this.handleTaskFunctionOperationMessage(message) - } else if (taskId != null && data != null) { - // Task message received - this.run(message) - } else if (kill === true) { - // Kill message received - this.handleKillMessage(message) + protected abstract handleError (error: Error): { + aborted: boolean + error?: Error + message: string + stack?: string + } + + /** + * Handles a kill message sent by the main worker. + * @param message - The kill message. + */ + protected handleKillMessage (message: MessageValue): void { + this.stopCheckActive() + if (isAsyncFunction(this.opts.killHandler)) { + ;(this.opts.killHandler as () => Promise)() + .then(() => { + this.sendToMainWorker({ kill: 'success' }) + return undefined + }) + .catch(() => { + this.sendToMainWorker({ kill: 'failure' }) + }) + } else { + try { + ;(this.opts.killHandler as (() => void) | undefined)?.() + this.sendToMainWorker({ kill: 'success' }) + } catch { + this.sendToMainWorker({ kill: 'failure' }) + } } } + /** + * Handles the ready message sent by the main worker. + * @param message - The ready message. + */ + protected abstract handleReadyMessage (message: MessageValue): void + protected handleTaskFunctionOperationMessage ( message: MessageValue ): void { - const { taskFunctionOperation, taskFunctionProperties, taskFunction } = + const { taskFunction, taskFunctionOperation, taskFunctionProperties } = message if (taskFunctionProperties == null) { throw new Error( @@ -354,11 +331,15 @@ export abstract class AbstractWorker< let response: TaskFunctionOperationResult switch (taskFunctionOperation) { case 'add': + if (typeof taskFunction !== 'string') { + throw new Error( + `Cannot handle task function operation ${taskFunctionOperation} message without task function` + ) + } response = this.addTaskFunction(taskFunctionProperties.name, { - // eslint-disable-next-line @typescript-eslint/no-implied-eval, no-new-func + // eslint-disable-next-line @typescript-eslint/no-implied-eval, no-new-func, @typescript-eslint/no-unsafe-call taskFunction: new Function( - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - `return ${taskFunction!}` + `return ${taskFunction}` )() as TaskFunction, ...(taskFunctionProperties.priority != null && { priority: taskFunctionProperties.priority, @@ -368,162 +349,101 @@ export abstract class AbstractWorker< }), }) break - case 'remove': - response = this.removeTaskFunction(taskFunctionProperties.name) - break case 'default': response = this.setDefaultTaskFunction(taskFunctionProperties.name) break + case 'remove': + response = this.removeTaskFunction(taskFunctionProperties.name) + break default: - response = { status: false, error: new Error('Unknown task operation') } + response = { + error: new Error('Unknown task operation'), + status: false, + } break } + const { error, status } = response this.sendToMainWorker({ taskFunctionOperation, - taskFunctionOperationStatus: response.status, + taskFunctionOperationStatus: status, taskFunctionProperties, - ...(!response.status && - response.error != null && { + ...(!status && + error != null && { workerError: { name: taskFunctionProperties.name, - message: this.handleError(response.error as Error | string), + ...this.handleError(error), }, }), }) } /** - * Handles a kill message sent by the main worker. - * @param message - The kill message. + * Worker message listener. + * @param message - The received message. */ - protected handleKillMessage (message: MessageValue): void { - this.stopCheckActive() - if (isAsyncFunction(this.opts.killHandler)) { - ;(this.opts.killHandler() as Promise) - .then(() => { - this.sendToMainWorker({ kill: 'success' }) - return undefined - }) - .catch(() => { - this.sendToMainWorker({ kill: 'failure' }) - }) - } else { - try { - ;(this.opts.killHandler as (() => void) | undefined)?.() - this.sendToMainWorker({ kill: 'success' }) - } catch { - this.sendToMainWorker({ kill: 'failure' }) + protected messageListener (message: MessageValue): void { + this.checkMessageWorkerId(message) + const { + checkActive, + data, + kill, + statistics, + taskFunctionOperation, + taskId, + taskOperation, + } = message + if (statistics != null) { + // Statistics message received + this.statistics = statistics + } else if (checkActive != null) { + // Check active message received + checkActive ? this.startCheckActive() : this.stopCheckActive() + } else if (taskFunctionOperation != null) { + // Task function operation message received + this.handleTaskFunctionOperationMessage(message) + } else if (taskId != null && data != null) { + // Task message received + this.run(message) + } else if (taskOperation === 'abort' && taskId != null) { + // Abort task operation message received + if (this.taskAbortFunctions.has(taskId)) { + this.taskAbortFunctions.get(taskId)?.() } + } else if (kill === true) { + // Kill message received + this.handleKillMessage(message) } } - /** - * Check if the message worker id is set and matches the worker id. - * @param message - The message to check. - * @throws {@link https://nodejs.org/api/errors.html#class-error} If the message worker id is not set or does not match the worker id. - */ - private checkMessageWorkerId (message: MessageValue): void { - if (message.workerId == null) { - throw new Error('Message worker id is not set') - } else if (message.workerId !== this.id) { - throw new Error( - `Message worker id ${message.workerId.toString()} does not match the worker id ${this.id.toString()}` - ) - } - } - - /** - * Starts the worker check active interval. - */ - private startCheckActive (): void { - this.lastTaskTimestamp = performance.now() - this.activeInterval = setInterval( - this.checkActive.bind(this), - (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME) / 2 - ) - } - - /** - * Stops the worker check active interval. - */ - private stopCheckActive (): void { - if (this.activeInterval != null) { - clearInterval(this.activeInterval) - delete this.activeInterval - } - } - - /** - * Checks if the worker should be terminated, because its living too long. - */ - private checkActive (): void { - if ( - performance.now() - this.lastTaskTimestamp > - (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME) - ) { - this.sendToMainWorker({ kill: this.opts.killBehavior }) - } - } - - /** - * Returns the main worker. - * @returns Reference to the main worker. - * @throws {@link https://nodejs.org/api/errors.html#class-error} If the main worker is not set. - */ - protected getMainWorker (): MainWorker { - if (this.mainWorker == null) { - throw new Error('Main worker not set') - } - return this.mainWorker - } - - /** - * Sends a message to main worker. - * @param message - The response message. - */ - protected abstract sendToMainWorker ( - message: MessageValue - ): void - - /** - * Sends task functions properties to the main worker. - */ - protected sendTaskFunctionsPropertiesToMainWorker (): void { - this.sendToMainWorker({ - taskFunctionsProperties: this.listTaskFunctionsProperties(), - }) - } - - /** - * Handles an error and convert it to a string so it can be sent back to the main worker. - * @param error - The error raised by the worker. - * @returns The error message. - */ - protected handleError (error: Error | string): string { - return error instanceof Error ? error.message : error - } - /** * Runs the given task. * @param task - The task to execute. */ protected readonly run = (task: Task): void => { - const { name, taskId, data } = task + const { abortable, data, name, taskId } = task const taskFunctionName = name ?? DEFAULT_TASK_NAME if (!this.taskFunctions.has(taskFunctionName)) { this.sendToMainWorker({ + taskId, workerError: { - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - name: name!, - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - message: `Task function '${name!}' not found`, data, + name, + ...this.handleError( + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + new Error(`Task function '${name!}' not found`) + ), }, - taskId, }) return } - const fn = this.taskFunctions.get(taskFunctionName)?.taskFunction + let fn: TaskFunction + if (abortable === true) { + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + fn = this.getAbortableTaskFunction(taskFunctionName, taskId!) + } else { + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + fn = this.taskFunctions.get(taskFunctionName)!.taskFunction + } if (isAsyncFunction(fn)) { this.runAsync(fn as TaskAsyncFunction, task) } else { @@ -531,40 +451,6 @@ export abstract class AbstractWorker< } } - /** - * Runs the given task function synchronously. - * @param fn - Task function that will be executed. - * @param task - Input data for the task function. - */ - protected readonly runSync = ( - fn: TaskSyncFunction, - task: Task - ): void => { - const { name, taskId, data } = task - try { - let taskPerformance = this.beginTaskPerformance(name) - const res = fn(data) - taskPerformance = this.endTaskPerformance(taskPerformance) - this.sendToMainWorker({ - data: res, - taskPerformance, - taskId, - }) - } catch (error) { - this.sendToMainWorker({ - workerError: { - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - name: name!, - message: this.handleError(error), - data, - }, - taskId, - }) - } finally { - this.updateLastTaskTimestamp() - } - } - /** * Runs the given task function asynchronously. * @param fn - Task function that will be executed. @@ -574,35 +460,92 @@ export abstract class AbstractWorker< fn: TaskAsyncFunction, task: Task ): void => { - const { name, taskId, data } = task + const { abortable, data, name, taskId } = task let taskPerformance = this.beginTaskPerformance(name) fn(data) .then(res => { taskPerformance = this.endTaskPerformance(taskPerformance) this.sendToMainWorker({ data: res, - taskPerformance, taskId, + taskPerformance, }) return undefined }) .catch((error: unknown) => { this.sendToMainWorker({ + taskId, workerError: { - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - name: name!, - message: this.handleError(error as Error | string), data, + name, + ...this.handleError(error as Error), }, - taskId, }) }) .finally(() => { this.updateLastTaskTimestamp() + if (abortable === true) { + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + this.taskAbortFunctions.delete(taskId!) + } }) .catch(EMPTY_FUNCTION) } + /** + * Runs the given task function synchronously. + * @param fn - Task function that will be executed. + * @param task - Input data for the task function. + */ + protected readonly runSync = ( + fn: TaskSyncFunction, + task: Task + ): void => { + const { abortable, data, name, taskId } = task + try { + let taskPerformance = this.beginTaskPerformance(name) + const res = fn(data) + taskPerformance = this.endTaskPerformance(taskPerformance) + this.sendToMainWorker({ + data: res, + taskId, + taskPerformance, + }) + } catch (error) { + this.sendToMainWorker({ + taskId, + workerError: { + data, + name, + ...this.handleError(error as Error), + }, + }) + } finally { + this.updateLastTaskTimestamp() + if (abortable === true) { + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + this.taskAbortFunctions.delete(taskId!) + } + } + } + + /** + * Sends task functions properties to the main worker. + */ + protected sendTaskFunctionsPropertiesToMainWorker (): void { + this.sendToMainWorker({ + taskFunctionsProperties: this.listTaskFunctionsProperties(), + }) + } + + /** + * Sends a message to main worker. + * @param message - The response message. + */ + protected abstract sendToMainWorker ( + message: MessageValue + ): void + private beginTaskPerformance (name?: string): TaskPerformance { if (this.statistics == null) { throw new Error('Performance statistics computation requirements not set') @@ -616,6 +559,90 @@ export abstract class AbstractWorker< } } + /** + * Checks if the worker should be terminated, because its living too long. + */ + private checkActive (): void { + if ( + performance.now() - this.lastTaskTimestamp > + (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME) + ) { + this.sendToMainWorker({ kill: this.opts.killBehavior }) + } + } + + /** + * Check if the message worker id is set and matches the worker id. + * @param message - The message to check. + * @throws {@link https://nodejs.org/api/errors.html#class-error} If the message worker id is not set or does not match the worker id. + */ + private checkMessageWorkerId (message: MessageValue): void { + if (message.workerId == null) { + throw new Error('Message worker id is not set') + } + if (message.workerId !== this.id) { + throw new Error( + `Message worker id ${message.workerId.toString()} does not match the worker id ${this.id.toString()}` + ) + } + } + + /** + * Checks if the `taskFunctions` parameter is passed to the constructor and valid. + * @param taskFunctions - The task function(s) parameter that should be checked. + */ + private checkTaskFunctions ( + taskFunctions: + | TaskFunction + | TaskFunctions + | undefined + ): void { + if (taskFunctions == null) { + throw new Error('taskFunctions parameter is mandatory') + } + this.taskFunctions = new Map>() + if (typeof taskFunctions === 'function') { + const fnObj = { taskFunction: taskFunctions.bind(this) } + this.taskFunctions.set(DEFAULT_TASK_NAME, fnObj) + this.taskFunctions.set( + typeof taskFunctions.name === 'string' && + taskFunctions.name.trim().length > 0 + ? taskFunctions.name + : 'fn1', + fnObj + ) + } else if (isPlainObject(taskFunctions)) { + let firstEntry = true + for (let [name, fnObj] of Object.entries(taskFunctions)) { + if (typeof fnObj === 'function') { + fnObj = { taskFunction: fnObj } satisfies TaskFunctionObject< + Data, + Response + > + } + checkValidTaskFunctionObjectEntry(name, fnObj) + fnObj.taskFunction = fnObj.taskFunction.bind(this) + if (firstEntry) { + this.taskFunctions.set(DEFAULT_TASK_NAME, fnObj) + firstEntry = false + } + this.taskFunctions.set(name, fnObj) + } + if (firstEntry) { + throw new Error('taskFunctions parameter object is empty') + } + } else { + throw new TypeError( + 'taskFunctions parameter is not a function or a plain object' + ) + } + } + + private checkWorkerOptions (opts: WorkerOptions): void { + checkValidWorkerOptions(opts) + this.opts = { ...DEFAULT_WORKER_OPTIONS, ...opts } + } + private endTaskPerformance ( taskPerformance: TaskPerformance ): TaskPerformance { @@ -633,6 +660,57 @@ export abstract class AbstractWorker< } } + /** + * Gets abortable task function. + * An abortable promise is built to permit the task to be aborted. + * @param name - The name of the task. + * @param taskId - The task id. + * @returns The abortable task function. + */ + private getAbortableTaskFunction ( + name: string, + taskId: `${string}-${string}-${string}-${string}-${string}` + ): TaskAsyncFunction { + return async (data?: Data): Promise => + await new Promise( + (resolve, reject: (reason?: unknown) => void) => { + this.taskAbortFunctions.set(taskId, () => { + reject(new AbortError(`Task '${name}' id '${taskId}' aborted`)) + }) + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + const taskFunction = this.taskFunctions.get(name)!.taskFunction + if (isAsyncFunction(taskFunction)) { + ;(taskFunction as TaskAsyncFunction)(data) + .then(resolve) + .catch(reject) + } else { + resolve((taskFunction as TaskSyncFunction)(data)) + } + } + ) + } + + /** + * Starts the worker check active interval. + */ + private startCheckActive (): void { + this.lastTaskTimestamp = performance.now() + this.activeInterval = setInterval( + this.checkActive.bind(this), + (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME) / 2 + ) + } + + /** + * Stops the worker check active interval. + */ + private stopCheckActive (): void { + if (this.activeInterval != null) { + clearInterval(this.activeInterval) + delete this.activeInterval + } + } + private updateLastTaskTimestamp (): void { if (this.activeInterval != null) { this.lastTaskTimestamp = performance.now()