Apply dependencies update (#381)

[poolifier.git] / src / pools / abstract-pool.ts

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 54447c82ba2db2e9a88503d6e670f73022a143bc..e0aa1810c4d839f39d29eee350ae010a6945ffab 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -2,6 +2,7 @@ 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, PoolType } from './pool-internal'
@@ -11,13 +12,6 @@ import {
   WorkerChoiceStrategyContext
 } from './selection-strategies'
 
-/**
- * An intentional empty function.
- */
-const EMPTY_FUNCTION: () => void = () => {
-  /* Intentionally empty */
-}
-
 /**
  * Callback invoked if the worker raised an error.
  */
@@ -122,7 +116,7 @@ export abstract class AbstractPool<
   /**
    * The promise map.
    *
-   * - `key`: This is the message ID of each submitted task.
+   * - `key`: This is the message Id of each submitted task.
    * - `value`: An object that contains the worker, the resolve function and the reject function.
    *
    * When we receive a message from the worker we get a map entry and resolve/reject the promise based on the message.
@@ -133,7 +127,7 @@ export abstract class AbstractPool<
   > = new Map<number, PromiseWorkerResponseWrapper<Worker, Response>>()
 
   /**
-   * ID of the next message.
+   * Id of the next message.
    */
   protected nextMessageId: number = 0
 
@@ -179,14 +173,14 @@ export abstract class AbstractPool<
       this,
       () => {
         const workerCreated = this.createAndSetupWorker()
-        this.registerWorkerMessageListener(workerCreated, message => {
+        this.registerWorkerMessageListener(workerCreated, async message => {
           const tasksInProgress = this.tasks.get(workerCreated)
           if (
             isKillBehavior(KillBehaviors.HARD, message.kill) ||
             tasksInProgress === 0
           ) {
             // Kill received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime)
-            void this.destroyWorker(workerCreated)
+            await this.destroyWorker(workerCreated)
           }
         })
         return workerCreated
@@ -268,10 +262,9 @@ export abstract class AbstractPool<
   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.checkAndEmitBusy()
     this.sendToWorker(worker, { data: data || ({} as Data), id: messageId })
     return res
   }
@@ -382,6 +375,7 @@ export abstract class AbstractPool<
     worker: Worker,
     messageId: number
   ): Promise<Response> {
+    this.increaseWorkersTask(worker)
     return new Promise<Response>((resolve, reject) => {
       this.promiseMap.set(messageId, { resolve, reject, worker })
     })