refactor: cleanup worker id handling 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, workerId: worker.id })
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 sendStartupMessageToWorker (worker: Worker): void {
78 this.sendToWorker(worker, {
79 ready: false,
80 workerId: worker.id
81 })
82 }
83
84 /** @inheritDoc */
85 protected registerWorkerMessageListener<Message extends Data | Response>(
86 worker: Worker,
87 listener: (message: MessageValue<Message>) => void
88 ): void {
89 worker.on('message', listener)
90 }
91
92 /** @inheritDoc */
93 protected createWorker (): Worker {
94 return cluster.fork(this.opts.env)
95 }
96
97 /** @inheritDoc */
98 protected get type (): PoolType {
99 return PoolTypes.fixed
100 }
101
102 /** @inheritDoc */
103 protected get worker (): WorkerType {
104 return WorkerTypes.cluster
105 }
106
107 /** @inheritDoc */
108 protected get minSize (): number {
109 return this.numberOfWorkers
110 }
111
112 /** @inheritDoc */
113 protected get maxSize (): number {
114 return this.numberOfWorkers
115 }
116
117 /** @inheritDoc */
118 protected get busy (): boolean {
119 return this.internalBusy()
120 }
121 }