X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=6548eba30189c2491bc244eb6e7e168b395b14f0;hb=765457872257684691f93befb94175a2aaa83502;hp=41ce6aa019a5269a0a47e338934c093aa2a3ee6d;hpb=6c9d7004857091d1d46e65cb7a9871bb271a2516;p=poolifier.git

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 41ce6aa0..6548eba3 100644
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -76,9 +76,9 @@ export abstract class AbstractPool<
     this.checkPoolOptions(this.opts)
 
     this.chooseWorkerNode.bind(this)
-    this.internalExecute.bind(this)
+    this.executeTask.bind(this)
+    this.enqueueTask.bind(this)
     this.checkAndEmitEvents.bind(this)
-    this.sendToWorker.bind(this)
 
     this.setupHook()
 
@@ -235,18 +235,21 @@ export abstract class AbstractPool<
       data: data ?? ({} as Data),
       id: crypto.randomUUID()
     }
-    const res = this.internalExecute(workerNodeKey, workerNode, submittedTask)
-    let currentTask: Task<Data> = submittedTask
+    const res = new Promise<Response>((resolve, reject) => {
+      this.promiseResponseMap.set(submittedTask.id, {
+        resolve,
+        reject,
+        worker: workerNode.worker
+      })
+    })
     if (
       this.opts.enableTasksQueue === true &&
-      (this.busy || this.tasksQueueSize(workerNodeKey) > 0)
+      (this.busy || this.workerNodes[workerNodeKey].tasksUsage.running > 0)
     ) {
-      currentTask = this.enqueueDequeueTask(
-        workerNodeKey,
-        submittedTask
-      ) as Task<Data>
+      this.enqueueTask(workerNodeKey, submittedTask)
+    } else {
+      this.executeTask(workerNodeKey, submittedTask)
     }
-    this.sendToWorker(workerNode.worker, currentTask)
     this.checkAndEmitEvents()
     // eslint-disable-next-line @typescript-eslint/return-await
     return res
@@ -270,7 +273,7 @@ export abstract class AbstractPool<
   protected abstract destroyWorker (worker: Worker): void | Promise<void>
 
   /**
-   * Setup hook to run code before worker node are created in the abstract constructor.
+   * Setup hook to execute code before worker node are created in the abstract constructor.
    * Can be overridden
    *
    * @virtual
@@ -285,23 +288,23 @@ export abstract class AbstractPool<
   protected abstract isMain (): boolean
 
   /**
-   * Hook executed before the worker task promise resolution.
+   * Hook executed before the worker task execution.
    * Can be overridden.
    *
    * @param workerNodeKey - The worker node key.
    */
-  protected beforePromiseResponseHook (workerNodeKey: number): void {
+  protected beforeTaskExecutionHook (workerNodeKey: number): void {
     ++this.workerNodes[workerNodeKey].tasksUsage.running
   }
 
   /**
-   * Hook executed after the worker task promise resolution.
+   * Hook executed after the worker task execution.
    * Can be overridden.
    *
    * @param worker - The worker.
    * @param message - The received message.
    */
-  protected afterPromiseResponseHook (
+  protected afterTaskExecutionHook (
     worker: Worker,
     message: MessageValue<Response>
   ): void {
@@ -434,15 +437,15 @@ export abstract class AbstractPool<
           } else {
             promiseResponse.resolve(message.data as Response)
           }
-          this.afterPromiseResponseHook(promiseResponse.worker, message)
+          this.afterTaskExecutionHook(promiseResponse.worker, message)
           this.promiseResponseMap.delete(message.id)
           const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker)
           if (
             this.opts.enableTasksQueue === true &&
             this.tasksQueueSize(workerNodeKey) > 0
           ) {
-            this.sendToWorker(
-              promiseResponse.worker,
+            this.executeTask(
+              workerNodeKey,
               this.dequeueTask(workerNodeKey) as Task<Data>
             )
           }
@@ -451,21 +454,6 @@ export abstract class AbstractPool<
     }
   }
 
-  private async internalExecute (
-    workerNodeKey: number,
-    workerNode: WorkerNode<Worker, Data>,
-    task: Task<Data>
-  ): Promise<Response> {
-    this.beforePromiseResponseHook(workerNodeKey)
-    return await new Promise<Response>((resolve, reject) => {
-      this.promiseResponseMap.set(task.id, {
-        resolve,
-        reject,
-        worker: workerNode.worker
-      })
-    })
-  }
-
   private checkAndEmitEvents (): void {
     if (this.opts.enableEvents === true) {
       if (this.busy) {
@@ -558,12 +546,9 @@ export abstract class AbstractPool<
     this.workerChoiceStrategyContext.remove(workerNodeKey)
   }
 
-  private enqueueDequeueTask (
-    workerNodeKey: number,
-    task: Task<Data>
-  ): Task<Data> | undefined {
-    this.enqueueTask(workerNodeKey, task)
-    return this.dequeueTask(workerNodeKey)
+  private executeTask (workerNodeKey: number, task: Task<Data>): void {
+    this.beforeTaskExecutionHook(workerNodeKey)
+    this.sendToWorker(this.workerNodes[workerNodeKey].worker, task)
   }
 
   private enqueueTask (workerNodeKey: number, task: Task<Data>): void {
@@ -581,9 +566,8 @@ export abstract class AbstractPool<
   private flushTasksQueue (workerNodeKey: number): void {
     if (this.tasksQueueSize(workerNodeKey) > 0) {
       for (const task of this.workerNodes[workerNodeKey].tasksQueue) {
-        this.sendToWorker(this.workerNodes[workerNodeKey].worker, task)
+        this.executeTask(workerNodeKey, task)
       }
-      this.workerNodes[workerNodeKey].tasksQueue = []
     }
   }