* The start timestamp of the pool.
*/
private readonly startTimestamp
+ /**
+ * The task function names.
+ */
+ private taskFunctions!: string[]
/**
* Constructs a new poolifier pool.
* @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
*/
private checkMessageWorkerId (message: MessageValue<Response>): void {
- if (
+ if (message.workerId == null) {
+ throw new Error('Worker message received without worker id')
+ } else if (
message.workerId != null &&
this.getWorkerNodeKeyByWorkerId(message.workerId) === -1
) {
}
}
+ /** @inheritDoc */
+ public listTaskFunctions (): string[] {
+ if (this.taskFunctions != null) {
+ return this.taskFunctions
+ } else {
+ return []
+ }
+ }
+
/** @inheritDoc */
public async execute (
data?: Data,
if (name != null && typeof name !== 'string') {
reject(new TypeError('name argument must be a string'))
}
+ if (
+ name != null &&
+ typeof name === 'string' &&
+ name.trim().length === 0
+ ) {
+ reject(new TypeError('name argument must not be an empty string'))
+ }
+ if (
+ name != null &&
+ this.taskFunctions != null &&
+ !this.taskFunctions.includes(name)
+ ) {
+ reject(
+ new Error(`Task function '${name}' is not registered in the pool`)
+ )
+ }
if (transferList != null && !Array.isArray(transferList)) {
reject(new TypeError('transferList argument must be an array'))
}
await this.destroyWorkerNode(workerNodeKey)
})
)
+ this.emitter?.emit(PoolEvents.destroy)
}
protected async sendKillMessageToWorker (
protected createAndSetupWorkerNode (): number {
const worker = this.createWorker()
+ worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION)
worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION)
worker.on('error', (error) => {
this.redistributeQueuedTasks(workerNodeKey)
}
})
- worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION)
worker.on('exit', this.opts.exitHandler ?? EMPTY_FUNCTION)
worker.once('exit', () => {
this.removeWorkerNode(worker)
} else if (message.taskId != null) {
// Task execution response received from worker
this.handleTaskExecutionResponse(message)
+ } else if (message.taskFunctions != null) {
+ // Task functions message received from worker
+ this.taskFunctions = message.taskFunctions
}
}
}
private handleWorkerReadyResponse (message: MessageValue<Response>): void {
+ if (message.ready === false) {
+ throw new Error(`Worker ${message.workerId} failed to initialize`)
+ }
this.getWorkerInfo(
this.getWorkerNodeKeyByWorkerId(message.workerId)
).ready = message.ready as boolean