src/pools/cluster/fixed.ts

   1 import { fork, isMaster, setupMaster, Worker } from 'cluster'
   2 import type { JSONValue, 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 export class FixedClusterPool<
  25   Data extends JSONValue = JSONValue,
  26   Response extends JSONValue = JSONValue
  27 > extends AbstractPool<Worker, Data, Response> {
  28   /**
  29    * @param numWorkers Number of workers for this pool.
  30    * @param filePath A file path with implementation of `ClusterWorker` class, relative path is fine.
  31    * @param opts An object with possible options for example `errorHandler`, `onlineHandler`. Default: `{ maxTasks: 1000 }`
  32    */
  33   public constructor (
  34     numWorkers: number,
  35     filePath: string,
  36     public readonly opts: ClusterPoolOptions = { maxTasks: 1000 }
  37   ) {
  38     super(numWorkers, filePath, opts)
  39   }
  40
  41   protected setupHook (): void {
  42     setupMaster({
  43       exec: this.filePath
  44     })
  45   }
  46
  47   protected isMain (): boolean {
  48     return isMaster
  49   }
  50
  51   protected destroyWorker (worker: Worker): void {
  52     worker.kill()
  53   }
  54
  55   protected sendToWorker (worker: Worker, message: MessageValue<Data>): void {
  56     worker.send(message)
  57   }
  58
  59   protected registerWorkerMessageListener (
  60     port: Worker,
  61     listener: (message: MessageValue<Response>) => void
  62   ): void {
  63     port.on('message', listener)
  64   }
  65
  66   protected unregisterWorkerMessageListener (
  67     port: Worker,
  68     listener: (message: MessageValue<Response>) => void
  69   ): void {
  70     port.removeListener('message', listener)
  71   }
  72
  73   protected newWorker (): Worker {
  74     return fork(this.opts.env)
  75   }
  76
  77   protected afterNewWorkerPushed (worker: Worker): void {
  78     // we will attach a listener for every task,
  79     // when task is completed the listener will be removed but to avoid warnings we are increasing the max listeners size
  80     worker.setMaxListeners(this.opts.maxTasks ?? 1000)
  81   }
  82 }