-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'
MainWorker extends Worker | MessagePort,
Data = unknown,
Response = unknown
-> extends AsyncResource {
+> {
/**
* Worker id.
*/
/**
* Constructs a new poolifier worker.
*
- * @param type - The type of async event.
* @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 opts - Options for the worker.
*/
public constructor (
- type: string,
protected readonly isMain: boolean,
private readonly mainWorker: MainWorker,
taskFunctions: TaskFunction<Data, Response> | TaskFunctions<Data, Response>,
protected opts: WorkerOptions = DEFAULT_WORKER_OPTIONS
) {
- super(type)
if (this.isMain == null) {
throw new Error('isMain parameter is mandatory')
}
.catch(() => {
this.sendToMainWorker({ kill: 'failure' })
})
- .finally(() => {
- this.emitDestroy()
- })
- .catch(EMPTY_FUNCTION)
} else {
try {
// eslint-disable-next-line @typescript-eslint/no-invalid-void-type
this.sendToMainWorker({ kill: 'success' })
} catch {
this.sendToMainWorker({ kill: 'failure' })
- } finally {
- this.emitDestroy()
}
}
}
*
* @param task - The task to execute.
*/
- protected run (task: Task<Data>): void {
+ protected readonly run = (task: Task<Data>): 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,
})
return
}
+ const fn = this.taskFunctions.get(taskFunctionName)
if (isAsyncFunction(fn)) {
- this.runInAsyncScope(this.runAsync.bind(this), this, fn, task)
+ this.runAsync(fn as TaskAsyncFunction<Data, Response>, task)
} else {
- this.runInAsyncScope(this.runSync.bind(this), this, fn, task)
+ this.runSync(fn as TaskSyncFunction<Data, Response>, task)
}
}
* @param fn - Task function that will be executed.
* @param task - Input data for the task function.
*/
- protected runSync (
+ protected readonly runSync = (
fn: TaskSyncFunction<Data, Response>,
task: Task<Data>
- ): void {
+ ): void => {
const { name, taskId, data } = task
try {
let taskPerformance = this.beginTaskPerformance(name)
* @param fn - Task function that will be executed.
* @param task - Input data for the task function.
*/
- protected runAsync (
+ protected readonly runAsync = (
fn: TaskAsyncFunction<Data, Response>,
task: Task<Data>
- ): void {
+ ): void => {
const { name, taskId, data } = task
let taskPerformance = this.beginTaskPerformance(name)
fn(data)