src/worker/abstract-worker.ts

   1 import { AsyncResource } from 'async_hooks'
   2 import type { MessageValue } from '../utility-types'
   3 import type { WorkerOptions } from './worker-options'
   4
   5 /**
   6  * Base class containing some shared logic for all poolifier workers.
   7  *
   8  * @template MainWorker Type of main worker.
   9  * @template Data Type of data this worker receives from pool's execution.
  10  * @template Response Type of response the worker sends back to the main worker.
  11  */
  12 export abstract class AbstractWorker<
  13   MainWorker,
  14   Data = unknown,
  15   Response = unknown
  16 > extends AsyncResource {
  17   /**
  18    * The maximum time to keep this worker alive while idle. The pool automatically checks and terminates this worker when the time expires.
  19    */
  20   protected readonly maxInactiveTime: number
  21   /**
  22    * Whether the worker is working asynchronously or not.
  23    */
  24   protected readonly async: boolean
  25   /**
  26    * Timestamp of the last task processed by this worker.
  27    */
  28   protected lastTask: number
  29   /**
  30    * Handler ID of the `interval` alive check.
  31    */
  32   protected readonly interval?: NodeJS.Timeout
  33
  34   /**
  35    * Constructs a new poolifier worker.
  36    *
  37    * @param type The type of async event.
  38    * @param isMain Whether this is the main worker or not.
  39    * @param fn Function processed by the worker when the pool's `execution` function is invoked.
  40    * @param opts Options for the worker.
  41    */
  42   public constructor (
  43     type: string,
  44     isMain: boolean,
  45     fn: (data: Data) => Response,
  46     public readonly opts: WorkerOptions = {}
  47   ) {
  48     super(type)
  49
  50     this.maxInactiveTime = this.opts.maxInactiveTime ?? 1000 * 60
  51     this.async = !!this.opts.async
  52     this.lastTask = Date.now()
  53     if (!fn) throw new Error('fn parameter is mandatory')
  54     // keep the worker active
  55     if (!isMain) {
  56       this.interval = setInterval(
  57         this.checkAlive.bind(this),
  58         this.maxInactiveTime / 2
  59       )
  60       this.checkAlive.bind(this)()
  61     }
  62   }
  63
  64   /**
  65    * Returns the main worker.
  66    */
  67   protected abstract getMainWorker (): MainWorker
  68
  69   /**
  70    * Send a message to the main worker.
  71    *
  72    * @param message The response message.
  73    */
  74   protected abstract sendToMainWorker (message: MessageValue<Response>): void
  75
  76   /**
  77    * Check to see if the worker should be terminated, because its living too long.
  78    */
  79   protected checkAlive (): void {
  80     if (Date.now() - this.lastTask > this.maxInactiveTime) {
  81       this.sendToMainWorker({ kill: 1 })
  82     }
  83   }
  84
  85   /**
  86    * Handle an error and convert it to a string so it can be sent back to the main worker.
  87    *
  88    * @param e The error raised by the worker.
  89    * @returns Message of the error.
  90    */
  91   protected handleError (e: Error | string): string {
  92     return (e as unknown) as string
  93   }
  94
  95   /**
  96    * Run the given function synchronously.
  97    *
  98    * @param fn Function that will be executed.
  99    * @param value Input data for the given function.
 100    */
 101   protected run (
 102     fn: (data?: Data) => Response,
 103     value: MessageValue<Data>
 104   ): void {
 105     try {
 106       const res = fn(value.data)
 107       this.sendToMainWorker({ data: res, id: value.id })
 108       this.lastTask = Date.now()
 109     } catch (e) {
 110       const err = this.handleError(e)
 111       this.sendToMainWorker({ error: err, id: value.id })
 112       this.lastTask = Date.now()
 113     }
 114   }
 115
 116   /**
 117    * Run the given function asynchronously.
 118    *
 119    * @param fn Function that will be executed.
 120    * @param value Input data for the given function.
 121    */
 122   protected runAsync (
 123     fn: (data?: Data) => Promise<Response>,
 124     value: MessageValue<Data>
 125   ): void {
 126     fn(value.data)
 127       .then(res => {
 128         this.sendToMainWorker({ data: res, id: value.id })
 129         this.lastTask = Date.now()
 130         return null
 131       })
 132       .catch(e => {
 133         const err = this.handleError(e)
 134         this.sendToMainWorker({ error: err, id: value.id })
 135         this.lastTask = Date.now()
 136       })
 137   }
 138 }