X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fworker%2Fabstract-worker.ts;h=fbb8f8390eff2207253195eb3817c68dc879e2b7;hb=600543a2d1fa2acd379ca7abee65ace63775ef50;hp=0a6530e3889c61bb925ce4d787736ba72c104001;hpb=4dc274761ce688172117009d1b306077d321f461;p=poolifier.git

diff --git a/src/worker/abstract-worker.ts b/src/worker/abstract-worker.ts
index 0a6530e3..fbb8f839 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,39 +13,26 @@ 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
   /**
-   * This value is immediately set to true when the kill from the main worker is received.
+   * Options for the worker.
    */
-  private isKilled: boolean = false
+  public readonly opts: WorkerOptions
 
   /**
    * Constructs a new poolifier worker.
@@ -59,24 +47,29 @@ export abstract class AbstractWorker<
     type: string,
     isMain: boolean,
     fn: (data: Data) => Response,
-    protected mainWorker?: MainWorker | null,
-    public readonly opts: WorkerOptions = {
+    protected mainWorker: MainWorker | undefined | null,
+    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()
+    this.opts = opts
     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)()
     }
@@ -84,7 +77,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)
@@ -95,19 +88,28 @@ export abstract class AbstractWorker<
         this.mainWorker = value.parent
       } else if (value.kill) {
         // Here is time to kill this worker, just clearing the interval
-        this.isKilled = true
-        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) {
+  private checkFunctionInput (fn: (data: Data) => Response): void {
     if (!fn) throw new Error('fn parameter is mandatory')
   }
 
@@ -134,8 +136,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.isKilled) {
-      this.sendToMainWorker({ kill: this.killBehavior })
+    if (
+      Date.now() - this.lastTaskTimestamp >
+      (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME)
+    ) {
+      this.sendToMainWorker({ kill: this.opts.killBehavior })
     }
   }
 
@@ -146,7 +151,7 @@ export abstract class AbstractWorker<
    * @returns Message of the error.
    */
   protected handleError (e: Error | string): string {
-    return (e as unknown) as string
+    return e as string
   }
 
   /**
@@ -162,11 +167,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()
     }
   }
 
@@ -183,13 +188,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)
   }
 }