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 import { PoolType } from '../pool-internal'
   6
   7 /**
   8  * Options for a poolifier cluster pool.
   9  */
  10 export interface ClusterPoolOptions extends PoolOptions<Worker> {
  11   /**
  12    * Key/value pairs to add to worker process environment.
  13    *
  14    * @see https://nodejs.org/api/cluster.html#cluster_cluster_fork_env
  15    */
  16   // eslint-disable-next-line @typescript-eslint/no-explicit-any
  17   env?: any
  18 }
  19
  20 /**
  21  * A cluster pool with a fixed number of workers.
  22  *
  23  * It is possible to perform tasks in sync or asynchronous mode as you prefer.
  24  *
  25  * This pool selects the workers in a round robin fashion.
  26  *
  27  * @template Data Type of data sent to the worker. This can only be serializable data.
  28  * @template Response Type of response of execution. This can only be serializable data.
  29  *
  30  * @author [Christopher Quadflieg](https://github.com/Shinigami92)
  31  * @since 2.0.0
  32  */
  33 export class FixedClusterPool<
  34   Data = unknown,
  35   Response = unknown
  36 > extends AbstractPool<Worker, Data, Response> {
  37   /**
  38    * Constructs a new poolifier fixed cluster pool.
  39    *
  40    * @param numberOfWorkers Number of workers for this pool.
  41    * @param filePath Path to an implementation of a `ClusterWorker` file, which can be relative or absolute.
  42    * @param opts Options for this fixed cluster pool. Default: `{}`
  43    */
  44   public constructor (
  45     numberOfWorkers: number,
  46     filePath: string,
  47     public readonly opts: ClusterPoolOptions = {}
  48   ) {
  49     super(numberOfWorkers, filePath, opts)
  50   }
  51
  52   /** @inheritdoc */
  53   protected setupHook (): void {
  54     setupMaster({
  55       exec: this.filePath
  56     })
  57   }
  58
  59   /** @inheritdoc */
  60   protected isMain (): boolean {
  61     return isMaster
  62   }
  63
  64   /** @inheritdoc */
  65   public destroyWorker (worker: Worker): void {
  66     this.sendToWorker(worker, { kill: 1 })
  67     worker.kill()
  68   }
  69
  70   /** @inheritdoc */
  71   protected sendToWorker (worker: Worker, message: MessageValue<Data>): void {
  72     worker.send(message)
  73   }
  74
  75   /** @inheritdoc */
  76   public registerWorkerMessageListener<Message extends Data | Response> (
  77     worker: Worker,
  78     listener: (message: MessageValue<Message>) => void
  79   ): void {
  80     worker.on('message', listener)
  81   }
  82
  83   /** @inheritdoc */
  84   protected createWorker (): Worker {
  85     return fork(this.opts.env)
  86   }
  87
  88   /** @inheritdoc */
  89   protected afterWorkerSetup (worker: Worker): void {
  90     // Listen worker messages.
  91     this.registerWorkerMessageListener(worker, super.workerListener())
  92   }
  93
  94   /** @inheritdoc */
  95   public get type (): PoolType {
  96     return PoolType.FIXED
  97   }
  98
  99   /** @inheritdoc */
 100   public get busy (): boolean {
 101     return this.internalGetBusyStatus()
 102   }
 103 }