)
} else if (max === 0) {
throw new RangeError(
- 'Cannot instantiate a dynamic pool with a pool size equal to zero'
+ 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero'
)
} else if (min === max) {
throw new RangeError(
protected abstract get busy (): boolean
/**
- * Whether worker nodes are executing at least one task.
+ * Whether worker nodes are executing concurrently their tasks quota or not.
*
* @returns Worker nodes busyness boolean status.
*/
protected internalBusy (): boolean {
- return (
- this.workerNodes.findIndex(
- workerNode =>
- workerNode.info.ready && workerNode.usage.tasks.executing === 0
- ) === -1
- )
+ if (this.opts.enableTasksQueue === true) {
+ return (
+ this.workerNodes.findIndex(
+ workerNode =>
+ workerNode.info.ready &&
+ workerNode.usage.tasks.executing <
+ (this.opts.tasksQueueOptions?.concurrency as number)
+ ) === -1
+ )
+ } else {
+ return (
+ this.workerNodes.findIndex(
+ workerNode =>
+ workerNode.info.ready && workerNode.usage.tasks.executing === 0
+ ) === -1
+ )
+ }
}
/** @inheritDoc */
data: data ?? ({} as Data),
timestamp,
workerId: this.getWorkerInfo(workerNodeKey).id as number,
- id: randomUUID()
+ taskId: randomUUID()
}
- this.promiseResponseMap.set(task.id as string, {
+ this.promiseResponseMap.set(task.taskId as string, {
resolve,
reject,
workerNodeKey
if (message.ready != null) {
// Worker ready response received from worker
this.handleWorkerReadyResponse(message)
- } else if (message.id != null) {
+ } else if (message.taskId != null) {
// Task execution response received from worker
this.handleTaskExecutionResponse(message)
}
}
private handleTaskExecutionResponse (message: MessageValue<Response>): void {
- const promiseResponse = this.promiseResponseMap.get(message.id as string)
+ const promiseResponse = this.promiseResponseMap.get(
+ message.taskId as string
+ )
if (promiseResponse != null) {
if (message.taskError != null) {
this.emitter?.emit(PoolEvents.taskError, message.taskError)
}
const workerNodeKey = promiseResponse.workerNodeKey
this.afterTaskExecutionHook(workerNodeKey, message)
- this.promiseResponseMap.delete(message.id as string)
+ this.promiseResponseMap.delete(message.taskId as string)
if (
this.opts.enableTasksQueue === true &&
this.tasksQueueSize(workerNodeKey) > 0 &&