src/pools/cluster/fixed.ts

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