src/worker/WorkerStaticPool.ts

   1 import { FixedThreadPool, type PoolEmitter, type PoolInfo } from 'poolifier';
   2
   3 import { WorkerAbstract } from './WorkerAbstract';
   4 import type { WorkerData, WorkerOptions } from './WorkerTypes';
   5 import { randomizeDelay, sleep } from './WorkerUtils';
   6
   7 export class WorkerStaticPool extends WorkerAbstract<WorkerData> {
   8   private readonly pool: FixedThreadPool<WorkerData>;
   9
  10   /**
  11    * Creates a new `WorkerStaticPool`.
  12    *
  13    * @param workerScript -
  14    * @param workerOptions -
  15    */
  16   constructor(workerScript: string, workerOptions: WorkerOptions) {
  17     super(workerScript, workerOptions);
  18     this.pool = new FixedThreadPool(
  19       this.workerOptions.poolMaxSize,
  20       this.workerScript,
  21       this.workerOptions.poolOptions,
  22     );
  23   }
  24
  25   get info(): PoolInfo {
  26     return this.pool.info;
  27   }
  28
  29   get size(): number {
  30     return this.pool.info.workerNodes;
  31   }
  32
  33   get maxElementsPerWorker(): number | undefined {
  34     return undefined;
  35   }
  36
  37   get emitter(): PoolEmitter | undefined {
  38     return this.pool?.emitter;
  39   }
  40
  41   /** @inheritDoc */
  42   public async start(): Promise<void> {
  43     // This is intentional
  44   }
  45
  46   /** @inheritDoc */
  47   public async stop(): Promise<void> {
  48     return this.pool.destroy();
  49   }
  50
  51   /** @inheritDoc */
  52   public async addElement(elementData: WorkerData): Promise<void> {
  53     await this.pool.execute(elementData);
  54     // Start element sequentially to optimize memory at startup
  55     this.workerOptions.elementStartDelay! > 0 &&
  56       (await sleep(randomizeDelay(this.workerOptions.elementStartDelay!)));
  57   }
  58 }