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