X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2Fabstract-worker.ts;h=e97c03473c2dc04da13d6507b40fc750ca72d8a8;hb=c8d883806c74b85db4f68d2df035bf0b49b75254;hp=6aa34e8e899fda868bf0c1935b656a21d8b67629;hpb=eb3b4c1592abc3f91acfe98c236a02ca994febff;p=poolifier.git

diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts
index 6aa34e8e..e97c0347 100644
--- a/src/worker/abstract-worker.ts
+++ b/src/worker/abstract-worker.ts
@@ -2,6 +2,7 @@ import { AsyncResource } from 'async_hooks'
 import type { Worker } from 'cluster'
 import type { MessagePort } from 'worker_threads'
 import type { MessageValue } from '../utility-types'
+import { EMPTY_FUNCTION } from '../utils'
 import type { KillBehavior, WorkerOptions } from './worker-options'
 import { KillBehaviors } from './worker-options'
 
@@ -12,34 +13,22 @@ const DEFAULT_KILL_BEHAVIOR: KillBehavior = KillBehaviors.SOFT
  * Base class containing some shared logic for all poolifier workers.
  *
  * @template MainWorker Type of main worker.
- * @template Data Type of data this worker receives from pool's execution.
- * @template Response Type of response the worker sends back to the main worker.
+ * @template Data Type of data this worker receives from pool's execution. This can only be serializable data.
+ * @template Response Type of response the worker sends back to the main worker. This can only be serializable data.
  */
 export abstract class AbstractWorker<
   MainWorker extends Worker | MessagePort,
   Data = unknown,
   Response = unknown
 > extends AsyncResource {
-  /**
-   * The maximum time to keep this worker alive while idle. The pool automatically checks and terminates this worker when the time expires.
-   */
-  protected readonly maxInactiveTime: number
-  /**
-   * The kill behavior set as option on the Worker constructor or a default value.
-   */
-  protected readonly killBehavior: KillBehavior
-  /**
-   * Whether the worker is working asynchronously or not.
-   */
-  protected readonly async: boolean
   /**
    * Timestamp of the last task processed by this worker.
    */
-  protected lastTask: number
+  protected lastTaskTimestamp: number
   /**
-   * Handler ID of the `interval` alive check.
+   * Handler Id of the `aliveInterval` worker alive check.
    */
-  protected readonly interval?: NodeJS.Timeout
+  protected readonly aliveInterval?: NodeJS.Timeout
 
   /**
    * Constructs a new poolifier worker.
@@ -54,24 +43,28 @@ export abstract class AbstractWorker<
     type: string,
     isMain: boolean,
     fn: (data: Data) => Response,
-    protected mainWorker?: MainWorker | null,
+    protected mainWorker: MainWorker | null,
     public readonly opts: WorkerOptions = {
+      /**
+       * The kill behavior option on this Worker or its default value.
+       */
       killBehavior: DEFAULT_KILL_BEHAVIOR,
+      /**
+       * The maximum time to keep this worker alive while idle.
+       * The pool automatically checks and terminates this worker when the time expires.
+       */
       maxInactiveTime: DEFAULT_MAX_INACTIVE_TIME
     }
   ) {
     super(type)
-    this.killBehavior = this.opts.killBehavior ?? DEFAULT_KILL_BEHAVIOR
-    this.maxInactiveTime =
-      this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME
-    this.async = !!this.opts.async
-    this.lastTask = Date.now()
-    if (!fn) throw new Error('fn parameter is mandatory')
+    this.checkFunctionInput(fn)
+    this.checkWorkerOptions(this.opts)
+    this.lastTaskTimestamp = Date.now()
     // Keep the worker active
     if (!isMain) {
-      this.interval = setInterval(
+      this.aliveInterval = setInterval(
         this.checkAlive.bind(this),
-        this.maxInactiveTime / 2
+        (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME) / 2
       )
       this.checkAlive.bind(this)()
     }
@@ -79,7 +72,7 @@ export abstract class AbstractWorker<
     this.mainWorker?.on('message', (value: MessageValue<Data, MainWorker>) => {
       if (value?.data && value.id) {
         // Here you will receive messages
-        if (this.async) {
+        if (this.opts.async) {
           this.runInAsyncScope(this.runAsync.bind(this), this, fn, value)
         } else {
           this.runInAsyncScope(this.run.bind(this), this, fn, value)
@@ -90,12 +83,31 @@ export abstract class AbstractWorker<
         this.mainWorker = value.parent
       } else if (value.kill) {
         // Here is time to kill this worker, just clearing the interval
-        if (this.interval) clearInterval(this.interval)
+        if (this.aliveInterval) clearInterval(this.aliveInterval)
         this.emitDestroy()
       }
     })
   }
 
+  private checkWorkerOptions (opts: WorkerOptions) {
+    this.opts.killBehavior = opts.killBehavior ?? DEFAULT_KILL_BEHAVIOR
+    this.opts.maxInactiveTime =
+      opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME
+    /**
+     * Whether the worker is working asynchronously or not.
+     */
+    this.opts.async = !!opts.async
+  }
+
+  /**
+   * Check if the `fn` parameter is passed to the constructor.
+   *
+   * @param fn The function that should be defined.
+   */
+  private checkFunctionInput (fn: (data: Data) => Response): void {
+    if (!fn) throw new Error('fn parameter is mandatory')
+  }
+
   /**
    * Returns the main worker.
    *
@@ -119,8 +131,11 @@ export abstract class AbstractWorker<
    * Check to see if the worker should be terminated, because its living too long.
    */
   protected checkAlive (): void {
-    if (Date.now() - this.lastTask > this.maxInactiveTime) {
-      this.sendToMainWorker({ kill: this.killBehavior })
+    if (
+      Date.now() - this.lastTaskTimestamp >
+      (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME)
+    ) {
+      this.sendToMainWorker({ kill: this.opts.killBehavior })
     }
   }
 
@@ -131,7 +146,7 @@ export abstract class AbstractWorker<
    * @returns Message of the error.
    */
   protected handleError (e: Error | string): string {
-    return (e as unknown) as string
+    return e.toString()
   }
 
   /**
@@ -147,11 +162,11 @@ export abstract class AbstractWorker<
     try {
       const res = fn(value.data)
       this.sendToMainWorker({ data: res, id: value.id })
-      this.lastTask = Date.now()
     } catch (e) {
-      const err = this.handleError(e)
+      const err = this.handleError(e as Error)
       this.sendToMainWorker({ error: err, id: value.id })
-      this.lastTask = Date.now()
+    } finally {
+      this.lastTaskTimestamp = Date.now()
     }
   }
 
@@ -168,13 +183,15 @@ export abstract class AbstractWorker<
     fn(value.data)
       .then(res => {
         this.sendToMainWorker({ data: res, id: value.id })
-        this.lastTask = Date.now()
         return null
       })
       .catch(e => {
         const err = this.handleError(e)
         this.sendToMainWorker({ error: err, id: value.id })
-        this.lastTask = Date.now()
       })
+      .finally(() => {
+        this.lastTaskTimestamp = Date.now()
+      })
+      .catch(EMPTY_FUNCTION)
   }
 }