src/pools/cluster/fixed.ts

   1 import type { Worker } from 'cluster'
   2 import cluster from 'cluster'
   3 import type { MessageValue } from '../../utility-types'
   4 import { AbstractPool } from '../abstract-pool'
   5 import type { PoolOptions } from '../pool'
   6 import { PoolType } from '../pool-internal'
   7
   8 /**
   9  * Options for a poolifier cluster pool.
  10  */
  11 export interface ClusterPoolOptions extends PoolOptions<Worker> {
  12   /**
  13    * Key/value pairs to add to worker process environment.
  14    *
  15    * @see https://nodejs.org/api/cluster.html#cluster_cluster_fork_env
  16    */
  17   // eslint-disable-next-line @typescript-eslint/no-explicit-any
  18   env?: any
  19 }
  20
  21 /**
  22  * A cluster pool with a fixed number of workers.
  23  *
  24  * It is possible to perform tasks in sync or asynchronous mode as you prefer.
  25  *
  26  * This pool selects the workers in a round robin fashion.
  27  *
  28  * @template DataType of data sent to the worker. This can only be serializable data.
  29  * @template ResponseType of response of execution. This can only be serializable data.
  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.
  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     cluster.setupPrimary({
  55       exec: this.filePath
  56     })
  57   }
  58
  59   /** @inheritDoc */
  60   protected isMain (): boolean {
  61     return cluster.isPrimary
  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 cluster.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 }