refactor: improve error reporting at task functions handling

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

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 3d10391c17748cdd63473c4afae7630665023008..9aeeebed772b37234c2a8f0e88c5cb45c65f0dce 100644 (file)
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -768,8 +768,8 @@ export abstract class AbstractPool<
     workerNodeKey: number,
     message: MessageValue<Data>
   ): Promise<boolean> {
-    const workerId = this.getWorkerInfo(workerNodeKey).id as number
     return await new Promise<boolean>((resolve, reject) => {
+      const workerId = this.getWorkerInfo(workerNodeKey).id as number
       this.registerWorkerMessageListener(workerNodeKey, message => {
         if (
           message.workerId === workerId &&
@@ -782,9 +782,11 @@ export abstract class AbstractPool<
         ) {
           reject(
             new Error(
-              `Task function operation ${
+              `Task function operation '${
                 message.taskFunctionOperation as string
-              } failed on worker ${message.workerId}`
+              }' failed on worker ${message.workerId} with error: '${
+                message.workerError?.message as string
+              }'`
             )
           )
         }
@@ -794,7 +796,7 @@ export abstract class AbstractPool<
   }
 
   private async sendTaskFunctionOperationToWorkers (
-    message: Omit<MessageValue<Data>, 'workerId'>
+    message: MessageValue<Data>
   ): Promise<boolean> {
     return await new Promise<boolean>((resolve, reject) => {
       const responsesReceived = new Array<MessageValue<Data | Response>>()
@@ -815,11 +817,18 @@ export abstract class AbstractPool<
                 message => message.taskFunctionOperationStatus === false
               )
             ) {
+              const errorResponse = responsesReceived.find(
+                response => response.taskFunctionOperationStatus === false
+              )
               reject(
                 new Error(
-                  `Task function operation ${
+                  `Task function operation '${
                     message.taskFunctionOperation as string
-                  } failed on worker ${message.workerId as number}`
+                  }' failed on worker ${
+                    errorResponse?.workerId as number
+                  } with error: '${
+                    errorResponse?.workerError?.message as string
+                  }'`
                 )
               )
             }
@@ -846,23 +855,40 @@ export abstract class AbstractPool<
   /** @inheritDoc */
   public async addTaskFunction (
     name: string,
-    taskFunction: TaskFunction<Data, Response>
+    fn: TaskFunction<Data, Response>
   ): Promise<boolean> {
-    this.taskFunctions.set(name, taskFunction)
-    return await this.sendTaskFunctionOperationToWorkers({
+    if (typeof name !== 'string') {
+      throw new TypeError('name argument must be a string')
+    }
+    if (typeof name === 'string' && name.trim().length === 0) {
+      throw new TypeError('name argument must not be an empty string')
+    }
+    if (typeof fn !== 'function') {
+      throw new TypeError('fn argument must be a function')
+    }
+    const opResult = await this.sendTaskFunctionOperationToWorkers({
       taskFunctionOperation: 'add',
       taskFunctionName: name,
-      taskFunction: taskFunction.toString()
+      taskFunction: fn.toString()
     })
+    this.taskFunctions.set(name, fn)
+    return opResult
   }
 
   /** @inheritDoc */
   public async removeTaskFunction (name: string): Promise<boolean> {
-    this.taskFunctions.delete(name)
-    return await this.sendTaskFunctionOperationToWorkers({
+    if (!this.taskFunctions.has(name)) {
+      throw new Error(
+        'Cannot remove a task function not handled on the pool side'
+      )
+    }
+    const opResult = await this.sendTaskFunctionOperationToWorkers({
       taskFunctionOperation: 'remove',
       taskFunctionName: name
     })
+    this.deleteTaskFunctionWorkerUsages(name)
+    this.taskFunctions.delete(name)
+    return opResult
   }
 
   /** @inheritDoc */
@@ -886,6 +912,12 @@ export abstract class AbstractPool<
     })
   }
 
+  private deleteTaskFunctionWorkerUsages (name: string): void {
+    for (const workerNode of this.workerNodes) {
+      workerNode.deleteTaskFunctionWorkerUsage(name)
+    }
+  }
+
   private shallExecuteTask (workerNodeKey: number): boolean {
     return (
       this.tasksQueueSize(workerNodeKey) === 0 &&