refactor: cleanup message passing code
[poolifier.git] / src / pools / cluster / fixed.ts
1 import cluster, { type ClusterSettings, type Worker } from 'node:cluster'
2 import type { MessageValue } from '../../utility-types'
3 import { AbstractPool } from '../abstract-pool'
4 import { type PoolOptions, type PoolType, PoolTypes } from '../pool'
5 import { type WorkerType, WorkerTypes } from '../worker'
6
7 /**
8 * Options for a poolifier cluster pool.
9 */
10 export interface ClusterPoolOptions extends PoolOptions<Worker> {
11 /**
12 * Key/value pairs to add to worker process environment.
13 *
14 * @see https://nodejs.org/api/cluster.html#cluster_cluster_fork_env
15 */
16 env?: Record<string, unknown>
17 /**
18 * Cluster settings.
19 *
20 * @see https://nodejs.org/api/cluster.html#cluster_cluster_settings
21 */
22 settings?: ClusterSettings
23 }
24
25 /**
26 * A cluster pool with a fixed number of workers.
27 *
28 * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data.
29 * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
30 * @author [Christopher Quadflieg](https://github.com/Shinigami92)
31 * @since 2.0.0
32 */
33 export class FixedClusterPool<
34 Data = unknown,
35 Response = unknown
36 > extends AbstractPool<Worker, Data, Response> {
37 /**
38 * Constructs a new poolifier fixed cluster pool.
39 *
40 * @param numberOfWorkers - Number of workers for this pool.
41 * @param filePath - Path to an implementation of a `ClusterWorker` file, which can be relative or absolute.
42 * @param opts - Options for this fixed cluster pool.
43 */
44 public constructor (
45 numberOfWorkers: number,
46 filePath: string,
47 protected readonly opts: ClusterPoolOptions = {}
48 ) {
49 super(numberOfWorkers, filePath, opts)
50 }
51
52 /** @inheritDoc */
53 protected setupHook (): void {
54 cluster.setupPrimary({ ...this.opts.settings, exec: this.filePath })
55 }
56
57 /** @inheritDoc */
58 protected isMain (): boolean {
59 return cluster.isPrimary
60 }
61
62 /** @inheritDoc */
63 protected destroyWorker (worker: Worker): void {
64 this.sendToWorker(worker, { kill: true })
65 worker.on('disconnect', () => {
66 worker.kill()
67 })
68 worker.disconnect()
69 }
70
71 /** @inheritDoc */
72 protected sendToWorker (worker: Worker, message: MessageValue<Data>): void {
73 worker.send(message)
74 }
75
76 /** @inheritDoc */
77 protected createWorker (): Worker {
78 return cluster.fork(this.opts.env)
79 }
80
81 /** @inheritDoc */
82 protected get type (): PoolType {
83 return PoolTypes.fixed
84 }
85
86 /** @inheritDoc */
87 protected get worker (): WorkerType {
88 return WorkerTypes.cluster
89 }
90
91 /** @inheritDoc */
92 protected get minSize (): number {
93 return this.numberOfWorkers
94 }
95
96 /** @inheritDoc */
97 protected get maxSize (): number {
98 return this.numberOfWorkers
99 }
100
101 /** @inheritDoc */
102 protected get busy (): boolean {
103 return this.internalBusy()
104 }
105 }