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'
import type {
* @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 NodeJS.Process | MessagePort,
+ MainWorker extends Worker | MessagePort,
Data = unknown,
Response = unknown
> extends AsyncResource {
+ /**
+ * Worker Id.
+ */
+ protected abstract id: number
/**
* Task function(s) processed by the worker when the pool's `execution` function is invoked.
*/
taskFunctions:
| WorkerFunction<Data, Response>
| TaskFunctions<Data, Response>,
- protected mainWorker: MainWorker,
+ protected readonly mainWorker: MainWorker,
protected readonly opts: WorkerOptions = {
/**
* The kill behavior option on this worker or its default value.
(this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME) / 2
)
this.checkAlive.bind(this)()
+ this.mainWorker?.on('message', this.messageListener.bind(this))
}
- this.mainWorker?.on('message', this.messageListener.bind(this))
}
private checkWorkerOptions (opts: WorkerOptions): void {
*
* @param message - Message received.
*/
- protected messageListener (
- message: MessageValue<Data, Data, MainWorker>
- ): void {
+ protected messageListener (message: MessageValue<Data, Data>): void {
if (message.id != null && message.data != null) {
// Task message received
const fn = this.getTaskFunction(message.name)
} else {
this.runInAsyncScope(this.runSync.bind(this), this, fn, message)
}
- } else if (message.parent != null) {
- // Main worker reference message received
- this.mainWorker = message.parent
} else if (message.statistics != null) {
// Statistics message received
this.statistics = message.statistics
this.sendToMainWorker({
data: res,
taskPerformance,
+ workerId: this.id,
id: message.id
})
} catch (e) {
message: err,
data: message.data
},
+ workerId: this.id,
id: message.id
})
} finally {
this.sendToMainWorker({
data: res,
taskPerformance,
+ workerId: this.id,
id: message.id
})
return null
message: err,
data: message.data
},
+ workerId: this.id,
id: message.id
})
})