src/pools/cluster/fixed.ts

   1 import cluster, { type Worker } from 'node:cluster'
   2 import type { MessageValue } from '../../utility-types'
   3 import { AbstractPool } from '../abstract-pool'
   4 import { type PoolOptions, type PoolType, PoolTypes } from '../pool'
   5 import { type WorkerType, WorkerTypes } from '../worker'
   6
   7 /**
   8  * A cluster pool with a fixed number of workers.
   9  *
  10  * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
  11  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
  12  * @author [Christopher Quadflieg](https://github.com/Shinigami92)
  13  * @since 2.0.0
  14  */
  15 export class FixedClusterPool<
  16   Data = unknown,
  17   Response = unknown
  18 > extends AbstractPool<Worker, Data, Response> {
  19   /**
  20    * Constructs a new poolifier fixed cluster pool.
  21    *
  22    * @param numberOfWorkers - Number of workers for this pool.
  23    * @param filePath - Path to an implementation of a `ClusterWorker` file, which can be relative or absolute.
  24    * @param opts - Options for this fixed cluster pool.
  25    */
  26   public constructor (
  27     numberOfWorkers: number,
  28     filePath: string,
  29     opts: PoolOptions<Worker> = {},
  30     maximumNumberOfWorkers?: number
  31   ) {
  32     super(numberOfWorkers, filePath, opts, maximumNumberOfWorkers)
  33   }
  34
  35   /** @inheritDoc */
  36   protected setupHook (): void {
  37     cluster.setupPrimary({ ...this.opts.settings, exec: this.filePath })
  38   }
  39
  40   /** @inheritDoc */
  41   protected isMain (): boolean {
  42     return cluster.isPrimary
  43   }
  44
  45   /** @inheritDoc */
  46   protected sendToWorker (
  47     workerNodeKey: number,
  48     message: MessageValue<Data>
  49   ): void {
  50     this.workerNodes[workerNodeKey].worker.send({
  51       ...message,
  52       workerId: this.getWorkerInfo(workerNodeKey).id as number
  53     })
  54   }
  55
  56   /** @inheritDoc */
  57   protected sendStartupMessageToWorker (workerNodeKey: number): void {
  58     this.sendToWorker(workerNodeKey, {
  59       ready: false
  60     })
  61   }
  62
  63   /** @inheritDoc */
  64   protected registerWorkerMessageListener<Message extends Data | Response>(
  65     workerNodeKey: number,
  66     listener: (message: MessageValue<Message>) => void
  67   ): void {
  68     this.workerNodes[workerNodeKey].worker.on('message', listener)
  69   }
  70
  71   /** @inheritDoc */
  72   protected registerOnceWorkerMessageListener<Message extends Data | Response>(
  73     workerNodeKey: number,
  74     listener: (message: MessageValue<Message>) => void
  75   ): void {
  76     this.workerNodes[workerNodeKey].worker.once('message', listener)
  77   }
  78
  79   /** @inheritDoc */
  80   protected deregisterWorkerMessageListener<Message extends Data | Response>(
  81     workerNodeKey: number,
  82     listener: (message: MessageValue<Message>) => void
  83   ): void {
  84     this.workerNodes[workerNodeKey].worker.off('message', listener)
  85   }
  86
  87   /** @inheritDoc */
  88   protected get type (): PoolType {
  89     return PoolTypes.fixed
  90   }
  91
  92   /** @inheritDoc */
  93   protected get worker (): WorkerType {
  94     return WorkerTypes.cluster
  95   }
  96
  97   /** @inheritDoc */
  98   protected get busy (): boolean {
  99     return this.internalBusy()
 100   }
 101 }