src/pools/pool.ts

   1 import EventEmitterAsyncResource from 'node:events'
   2 import type {
   3   ErrorHandler,
   4   ExitHandler,
   5   IWorker,
   6   MessageHandler,
   7   OnlineHandler,
   8   WorkerNode
   9 } from './worker'
  10 import type {
  11   WorkerChoiceStrategy,
  12   WorkerChoiceStrategyOptions
  13 } from './selection-strategies/selection-strategies-types'
  14
  15 /**
  16  * Enumeration of pool types.
  17  */
  18 export const PoolTypes = Object.freeze({
  19   /**
  20    * Fixed pool type.
  21    */
  22   fixed: 'fixed',
  23   /**
  24    * Dynamic pool type.
  25    */
  26   dynamic: 'dynamic'
  27 } as const)
  28
  29 /**
  30  * Pool type.
  31  */
  32 export type PoolType = keyof typeof PoolTypes
  33
  34 /**
  35  * Pool events emitter.
  36  */
  37 export class PoolEmitter extends EventEmitterAsyncResource {}
  38
  39 /**
  40  * Enumeration of pool events.
  41  */
  42 export const PoolEvents = Object.freeze({
  43   full: 'full',
  44   busy: 'busy',
  45   error: 'error',
  46   taskError: 'taskError'
  47 } as const)
  48
  49 /**
  50  * Pool event.
  51  */
  52 export type PoolEvent = keyof typeof PoolEvents
  53
  54 /**
  55  * Pool information.
  56  */
  57 export interface PoolInfo {
  58   type: PoolType
  59   minSize: number
  60   maxSize: number
  61   workerNodes: number
  62   idleWorkerNodes: number
  63   busyWorkerNodes: number
  64   runningTasks: number
  65   queuedTasks: number
  66   maxQueuedTasks: number
  67 }
  68
  69 /**
  70  * Worker tasks queue options.
  71  */
  72 export interface TasksQueueOptions {
  73   /**
  74    * Maximum number of tasks that can be executed concurrently on a worker.
  75    *
  76    * @defaultValue 1
  77    */
  78   concurrency?: number
  79 }
  80
  81 /**
  82  * Options for a poolifier pool.
  83  *
  84  * @typeParam Worker - Type of worker.
  85  */
  86 export interface PoolOptions<Worker extends IWorker> {
  87   /**
  88    * A function that will listen for message event on each worker.
  89    */
  90   messageHandler?: MessageHandler<Worker>
  91   /**
  92    * A function that will listen for error event on each worker.
  93    */
  94   errorHandler?: ErrorHandler<Worker>
  95   /**
  96    * A function that will listen for online event on each worker.
  97    */
  98   onlineHandler?: OnlineHandler<Worker>
  99   /**
 100    * A function that will listen for exit event on each worker.
 101    */
 102   exitHandler?: ExitHandler<Worker>
 103   /**
 104    * The worker choice strategy to use in this pool.
 105    *
 106    * @defaultValue WorkerChoiceStrategies.ROUND_ROBIN
 107    */
 108   workerChoiceStrategy?: WorkerChoiceStrategy
 109   /**
 110    * The worker choice strategy options.
 111    */
 112   workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 113   /**
 114    * Restart worker on error.
 115    */
 116   restartWorkerOnError?: boolean
 117   /**
 118    * Pool events emission.
 119    *
 120    * @defaultValue true
 121    */
 122   enableEvents?: boolean
 123   /**
 124    * Pool worker tasks queue.
 125    *
 126    * @defaultValue false
 127    */
 128   enableTasksQueue?: boolean
 129   /**
 130    * Pool worker tasks queue options.
 131    */
 132   tasksQueueOptions?: TasksQueueOptions
 133 }
 134
 135 /**
 136  * Contract definition for a poolifier pool.
 137  *
 138  * @typeParam Worker - Type of worker which manages this pool.
 139  * @typeParam Data - Type of data sent to the worker. This can only be serializable data.
 140  * @typeParam Response - Type of execution response. This can only be serializable data.
 141  */
 142 export interface IPool<
 143   Worker extends IWorker,
 144   Data = unknown,
 145   Response = unknown
 146 > {
 147   /**
 148    * Pool type.
 149    *
 150    * If it is `'dynamic'`, it provides the `max` property.
 151    */
 152   readonly type: PoolType
 153   /**
 154    * Pool information.
 155    */
 156   readonly info: PoolInfo
 157   /**
 158    * Pool worker nodes.
 159    */
 160   readonly workerNodes: Array<WorkerNode<Worker, Data>>
 161   /**
 162    * Emitter on which events can be listened to.
 163    *
 164    * Events that can currently be listened to:
 165    *
 166    * - `'full'`: Emitted when the pool is dynamic and full.
 167    * - `'busy'`: Emitted when the pool is busy.
 168    * - `'error'`: Emitted when an uncaught error occurs.
 169    * - `'taskError'`: Emitted when an error occurs while executing a task.
 170    */
 171   readonly emitter?: PoolEmitter
 172   /**
 173    * Executes the specified function in the worker constructor with the task data input parameter.
 174    *
 175    * @param data - The task input data for the specified worker function. This can only be serializable data.
 176    * @param name - The name of the worker function to execute. If not specified, the default worker function will be executed.
 177    * @returns Promise that will be fulfilled when the task is completed.
 178    */
 179   execute: (data?: Data, name?: string) => Promise<Response>
 180   /**
 181    * Shutdowns every current worker in this pool.
 182    */
 183   destroy: () => Promise<void>
 184   /**
 185    * Sets the worker choice strategy in this pool.
 186    *
 187    * @param workerChoiceStrategy - The worker choice strategy.
 188    * @param workerChoiceStrategyOptions - The worker choice strategy options.
 189    */
 190   setWorkerChoiceStrategy: (
 191     workerChoiceStrategy: WorkerChoiceStrategy,
 192     workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions
 193   ) => void
 194   /**
 195    * Sets the worker choice strategy options in this pool.
 196    *
 197    * @param workerChoiceStrategyOptions - The worker choice strategy options.
 198    */
 199   setWorkerChoiceStrategyOptions: (
 200     workerChoiceStrategyOptions: WorkerChoiceStrategyOptions
 201   ) => void
 202   /**
 203    * Enables/disables the worker tasks queue in this pool.
 204    *
 205    * @param enable - Whether to enable or disable the worker tasks queue.
 206    * @param tasksQueueOptions - The worker tasks queue options.
 207    */
 208   enableTasksQueue: (
 209     enable: boolean,
 210     tasksQueueOptions?: TasksQueueOptions
 211   ) => void
 212   /**
 213    * Sets the worker tasks queue options in this pool.
 214    *
 215    * @param tasksQueueOptions - The worker tasks queue options.
 216    */
 217   setTasksQueueOptions: (tasksQueueOptions: TasksQueueOptions) => void
 218 }