1 import type { EventEmitterAsyncResource
} from
'node:events'
3 import { DynamicThreadPool
, type PoolInfo
} from
'poolifier'
5 import type { WorkerData
, WorkerOptions
} from
'./WorkerTypes.js'
7 import { WorkerAbstract
} from
'./WorkerAbstract.js'
8 import { randomizeDelay
, sleep
} from
'./WorkerUtils.js'
10 export class WorkerDynamicPool
<D
extends WorkerData
, R
extends WorkerData
> extends WorkerAbstract
<
14 private readonly pool
: DynamicThreadPool
<D
, R
>
17 * Creates a new `WorkerDynamicPool`.
18 * @param workerScript -
19 * @param workerOptions -
21 constructor (workerScript
: string, workerOptions
: WorkerOptions
) {
22 super(workerScript
, workerOptions
)
23 this.pool
= new DynamicThreadPool
<D
, R
>(
24 this.workerOptions
.poolMinSize
,
25 this.workerOptions
.poolMaxSize
,
27 this.workerOptions
.poolOptions
32 public async addElement (elementData
: D
): Promise
<R
> {
33 const response
= await this.pool
.execute(elementData
)
34 // Start element sequentially to optimize memory at startup
35 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
36 this.workerOptions
.elementAddDelay
! > 0 &&
37 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
38 (await sleep(randomizeDelay(this.workerOptions
.elementAddDelay
!)))
43 public start (): void {
48 public async stop (): Promise
<void> {
49 await this.pool
.destroy()
52 get
emitter (): EventEmitterAsyncResource
| undefined {
53 return this.pool
.emitter
56 get
info (): PoolInfo
{
60 get
maxElementsPerWorker (): number | undefined {
65 return this.pool
.info
.workerNodes