X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;ds=sidebyside;f=src%2Fpools%2Fabstract-pool.ts;h=c3f293e78bc693fba6aea17e36c880968ee58443;hb=c9170223028c20b8e47f6ef43d33cc5b8b68e07e;hp=0c4b59c92fef071bbddd487d1490ed5d2f8db38b;hpb=4a6952ffec2bc4dba2d73bd747c003d0fe59fe7c;p=poolifier.git

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 0c4b59c9..c3f293e7 100644
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -2,22 +2,16 @@ import type {
   MessageValue,
   PromiseWorkerResponseWrapper
 } from '../utility-types'
+import { EMPTY_FUNCTION } from '../utils'
 import { isKillBehavior, KillBehaviors } from '../worker/worker-options'
 import type { IPoolInternal } from './pool-internal'
-import { PoolEmitter } from './pool-internal'
+import { PoolEmitter, PoolType } from './pool-internal'
 import type { WorkerChoiceStrategy } from './selection-strategies'
 import {
   WorkerChoiceStrategies,
   WorkerChoiceStrategyContext
 } from './selection-strategies'
 
-/**
- * An intentional empty function.
- */
-const EMPTY_FUNCTION: () => void = () => {
-  /* Intentionally empty */
-}
-
 /**
  * Callback invoked if the worker raised an error.
  */
@@ -87,6 +81,12 @@ export interface PoolOptions<Worker> {
    * The work choice strategy to use in this pool.
    */
   workerChoiceStrategy?: WorkerChoiceStrategy
+  /**
+   * Pool events emission.
+   *
+   * Default to true.
+   */
+  enableEvents?: boolean
 }
 
 /**
@@ -108,7 +108,10 @@ export abstract class AbstractPool<
   public readonly tasks: Map<Worker, number> = new Map<Worker, number>()
 
   /** @inheritdoc */
-  public readonly emitter: PoolEmitter
+  public readonly emitter?: PoolEmitter
+
+  /** @inheritdoc */
+  public readonly max?: number
 
   /**
    * The promise map.
@@ -156,13 +159,16 @@ export abstract class AbstractPool<
     }
     this.checkNumberOfWorkers(this.numberOfWorkers)
     this.checkFilePath(this.filePath)
+    this.checkPoolOptions(this.opts)
     this.setupHook()
 
     for (let i = 1; i <= this.numberOfWorkers; i++) {
       this.createAndSetupWorker()
     }
 
-    this.emitter = new PoolEmitter()
+    if (this.opts.enableEvents) {
+      this.emitter = new PoolEmitter()
+    }
     this.workerChoiceStrategyContext = new WorkerChoiceStrategyContext(
       this,
       () => {
@@ -179,7 +185,7 @@ export abstract class AbstractPool<
         })
         return workerCreated
       },
-      opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
+      this.opts.workerChoiceStrategy
     )
   }
 
@@ -202,14 +208,23 @@ export abstract class AbstractPool<
       throw new Error(
         'Cannot instantiate a pool with a negative number of workers'
       )
-    } else if (!this.dynamic && numberOfWorkers === 0) {
+    } else if (this.type === PoolType.FIXED && numberOfWorkers === 0) {
       throw new Error('Cannot instantiate a fixed pool with no worker')
     }
   }
 
+  private checkPoolOptions (opts: PoolOptions<Worker>): void {
+    this.opts.workerChoiceStrategy =
+      opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN
+    this.opts.enableEvents = opts.enableEvents ?? true
+  }
+
   /** @inheritdoc */
-  public get dynamic (): boolean {
-    return false
+  public abstract get type (): PoolType
+
+  /** @inheritdoc */
+  public get numberOfRunningTasks (): number {
+    return this.promiseMap.size
   }
 
   /** @inheritdoc */
@@ -222,11 +237,33 @@ export abstract class AbstractPool<
     )
   }
 
+  /** @inheritdoc */
+  public abstract get busy (): boolean
+
+  protected internalGetBusyStatus (): boolean {
+    return (
+      this.numberOfRunningTasks >= this.numberOfWorkers &&
+      this.findFreeTasksMapEntry() === false
+    )
+  }
+
+  /** @inheritdoc */
+  public findFreeTasksMapEntry (): [Worker, number] | false {
+    for (const [worker, numberOfTasks] of this.tasks) {
+      if (numberOfTasks === 0) {
+        // A worker is free, return the matching tasks map entry
+        return [worker, numberOfTasks]
+      }
+    }
+    return false
+  }
+
   /** @inheritdoc */
   public execute (data: Data): Promise<Response> {
     // Configure worker to handle message with the specified task
     const worker = this.chooseWorker()
     this.increaseWorkersTask(worker)
+    this.checkAndEmitBusy()
     const messageId = ++this.nextMessageId
     const res = this.internalExecute(worker, messageId)
     this.sendToWorker(worker, { data: data || ({} as Data), id: messageId })
@@ -259,7 +296,7 @@ export abstract class AbstractPool<
   protected abstract isMain (): boolean
 
   /**
-   * Increase the number of tasks that the given workers has done.
+   * Increase the number of tasks that the given workers has applied.
    *
    * @param worker Worker whose tasks are increased.
    */
@@ -268,7 +305,7 @@ export abstract class AbstractPool<
   }
 
   /**
-   * Decrease the number of tasks that the given workers has done.
+   * Decrease the number of tasks that the given workers has applied.
    *
    * @param worker Worker whose tasks are decreased.
    */
@@ -277,7 +314,7 @@ export abstract class AbstractPool<
   }
 
   /**
-   * Step the number of tasks that the given workers has done.
+   * Step the number of tasks that the given workers has applied.
    *
    * @param worker Worker whose tasks are set.
    * @param step Worker number of tasks step.
@@ -399,4 +436,10 @@ export abstract class AbstractPool<
       }
     }
   }
+
+  private checkAndEmitBusy (): void {
+    if (this.opts.enableEvents && this.busy) {
+      this.emitter?.emit('busy')
+    }
+  }
 }