X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fcluster%2Ffixed.ts;h=c60266e3d72b623fb9225dcc20882c064d29438c;hb=efc4d37dd21d0c80e5fa2d58009e6298eefe5d48;hp=d7e3cc4bce0c4291bac5a71df3b9878933f96148;hpb=85aeb3f356d749b96361e74cf17d403a697e3dd7;p=poolifier.git diff --git a/src/pools/cluster/fixed.ts b/src/pools/cluster/fixed.ts index d7e3cc4b..c60266e3 100644 --- a/src/pools/cluster/fixed.ts +++ b/src/pools/cluster/fixed.ts @@ -1,26 +1,13 @@ -import cluster, { type ClusterSettings, type Worker } from 'node:cluster' -import type { MessageValue } from '../../utility-types' -import { AbstractPool } from '../abstract-pool' -import { type PoolOptions, type PoolType, PoolTypes } from '../pool' -import { type WorkerType, WorkerTypes } from '../worker' +import cluster, { type Worker } from 'node:cluster' +import type { MessageValue } from '../../utility-types.js' +import { AbstractPool } from '../abstract-pool.js' +import { type PoolOptions, type PoolType, PoolTypes } from '../pool.js' +import { type WorkerType, WorkerTypes } from '../worker.js' /** * Options for a poolifier cluster pool. */ -export interface ClusterPoolOptions extends PoolOptions { - /** - * Key/value pairs to add to worker process environment. - * - * @see https://nodejs.org/api/cluster.html#cluster_cluster_fork_env - */ - env?: Record - /** - * Cluster settings. - * - * @see https://nodejs.org/api/cluster.html#cluster_cluster_settings - */ - settings?: ClusterSettings -} +export type ClusterPoolOptions = PoolOptions /** * A cluster pool with a fixed number of workers. @@ -44,9 +31,10 @@ export class FixedClusterPool< public constructor ( numberOfWorkers: number, filePath: string, - protected readonly opts: ClusterPoolOptions = {} + opts: ClusterPoolOptions = {}, + maximumNumberOfWorkers?: number ) { - super(numberOfWorkers, filePath, opts) + super(numberOfWorkers, filePath, opts, maximumNumberOfWorkers) } /** @inheritDoc */ @@ -60,58 +48,65 @@ export class FixedClusterPool< } /** @inheritDoc */ - protected destroyWorker (worker: Worker): void { - this.sendToWorker(worker, { kill: true, workerId: worker.id }) - worker.on('disconnect', () => { - worker.kill() - }) - worker.disconnect() + protected sendToWorker ( + workerNodeKey: number, + message: MessageValue + ): void { + this.workerNodes[workerNodeKey].worker.send({ + ...message, + workerId: this.getWorkerInfo(workerNodeKey)?.id + } satisfies MessageValue) } /** @inheritDoc */ - protected sendToWorker (worker: Worker, message: MessageValue): void { - worker.send(message) + protected sendStartupMessageToWorker (workerNodeKey: number): void { + this.sendToWorker(workerNodeKey, { + ready: false + }) } /** @inheritDoc */ - protected sendStartupMessageToWorker (worker: Worker): void { - this.sendToWorker(worker, { - ready: false, - workerId: this.getWorkerInfoByWorker(worker).id as number - }) + protected registerWorkerMessageListener( + workerNodeKey: number, + listener: (message: MessageValue) => void + ): void { + this.workerNodes[workerNodeKey].worker.on('message', listener) } /** @inheritDoc */ - protected registerWorkerMessageListener( - worker: Worker, + protected registerOnceWorkerMessageListener( + workerNodeKey: number, listener: (message: MessageValue) => void ): void { - worker.on('message', listener) + this.workerNodes[workerNodeKey].worker.once('message', listener) } /** @inheritDoc */ - protected createWorker (): Worker { - return cluster.fork(this.opts.env) + protected deregisterWorkerMessageListener( + workerNodeKey: number, + listener: (message: MessageValue) => void + ): void { + this.workerNodes[workerNodeKey].worker.off('message', listener) } /** @inheritDoc */ - protected get type (): PoolType { - return PoolTypes.fixed + protected shallCreateDynamicWorker (): boolean { + return false } /** @inheritDoc */ - protected get worker (): WorkerType { - return WorkerTypes.cluster + protected checkAndEmitDynamicWorkerCreationEvents (): void { + /* noop */ } /** @inheritDoc */ - protected get minSize (): number { - return this.numberOfWorkers + protected get type (): PoolType { + return PoolTypes.fixed } /** @inheritDoc */ - protected get maxSize (): number { - return this.numberOfWorkers + protected get worker (): WorkerType { + return WorkerTypes.cluster } /** @inheritDoc */