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