src/pools/cluster/fixed.ts

   1 import { fork, isMaster, setupMaster, Worker } from 'cluster'
   2 import type { MessageValue } from '../../utility-types'
   3 import type { PoolOptions } from '../abstract-pool'
   4 import { AbstractPool } from '../abstract-pool'
   5
   6 export interface ClusterPoolOptions extends PoolOptions<Worker> {
   7   /**
   8    * Key/value pairs to add to worker process environment.
   9    *
  10    * @see https://nodejs.org/api/cluster.html#cluster_cluster_fork_env
  11    */
  12   // eslint-disable-next-line @typescript-eslint/no-explicit-any
  13   env?: any
  14 }
  15
  16 /**
  17  * A cluster pool with a static number of workers, is possible to execute tasks in sync or async mode as you prefer.
  18  *
  19  * This pool will select the worker in a round robin fashion.
  20  *
  21  * @author [Christopher Quadflieg](https://github.com/Shinigami92)
  22  * @since 2.0.0
  23  */
  24 // eslint-disable-next-line @typescript-eslint/no-explicit-any
  25 export class FixedClusterPool<Data = any, Response = any> extends AbstractPool<
  26   Worker,
  27   Data,
  28   Response
  29 > {
  30   /**
  31    * @param numWorkers Number of workers for this pool.
  32    * @param filePath A file path with implementation of `ClusterWorker` class, relative path is fine.
  33    * @param opts An object with possible options for example `errorHandler`, `onlineHandler`. Default: `{ maxTasks: 1000 }`
  34    */
  35   public constructor (
  36     numWorkers: number,
  37     filePath: string,
  38     public readonly opts: ClusterPoolOptions = { maxTasks: 1000 }
  39   ) {
  40     super(numWorkers, filePath, opts)
  41   }
  42
  43   protected setupHook (): void {
  44     setupMaster({
  45       exec: this.filePath
  46     })
  47   }
  48
  49   protected isMain (): boolean {
  50     return isMaster
  51   }
  52
  53   protected destroyWorker (worker: Worker): void {
  54     worker.kill()
  55   }
  56
  57   protected sendToWorker (worker: Worker, message: MessageValue<Data>): void {
  58     worker.send(message)
  59   }
  60
  61   protected registerWorkerMessageListener (
  62     port: Worker,
  63     listener: (message: MessageValue<Response>) => void
  64   ): void {
  65     port.on('message', listener)
  66   }
  67
  68   protected unregisterWorkerMessageListener (
  69     port: Worker,
  70     listener: (message: MessageValue<Response>) => void
  71   ): void {
  72     port.removeListener('message', listener)
  73   }
  74
  75   protected newWorker (): Worker {
  76     return fork(this.opts.env)
  77   }
  78
  79   protected afterNewWorkerPushed (worker: Worker): void {
  80     // we will attach a listener for every task,
  81     // when task is completed the listener will be removed but to avoid warnings we are increasing the max listeners size
  82     worker.setMaxListeners(this.opts.maxTasks ?? 1000)
  83   }
  84 }