Add a pool option to register a message listener on pool workers (#487)
[poolifier.git] / src / pools / cluster / fixed.ts
CommitLineData
325f50bc 1import { fork, isMaster, setupMaster, Worker } from 'cluster'
deb85c12 2import type { MessageValue } from '../../utility-types'
c97c7edb
S
3import type { PoolOptions } from '../abstract-pool'
4import { AbstractPool } from '../abstract-pool'
7c0ba920 5import { PoolType } from '../pool-internal'
4ade5f1f 6
729c563d
S
7/**
8 * Options for a poolifier cluster pool.
9 */
c97c7edb 10export interface ClusterPoolOptions extends PoolOptions<Worker> {
325f50bc
S
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 // eslint-disable-next-line @typescript-eslint/no-explicit-any
17 env?: any
4ade5f1f
S
18}
19
20/**
729c563d
S
21 * A cluster pool with a fixed number of workers.
22 *
23 * It is possible to perform tasks in sync or asynchronous mode as you prefer.
4ade5f1f 24 *
729c563d
S
25 * This pool selects the workers in a round robin fashion.
26 *
91bfd799
JB
27 * @template DataType of data sent to the worker. This can only be serializable data.
28 * @template ResponseType of response of execution. This can only be serializable data.
325f50bc
S
29 * @author [Christopher Quadflieg](https://github.com/Shinigami92)
30 * @since 2.0.0
4ade5f1f 31 */
d3c8a1a8 32export class FixedClusterPool<
deb85c12
JB
33 Data = unknown,
34 Response = unknown
d3c8a1a8 35> extends AbstractPool<Worker, Data, Response> {
4ade5f1f 36 /**
729c563d
S
37 * Constructs a new poolifier fixed cluster pool.
38 *
5c5a1fb7 39 * @param numberOfWorkers Number of workers for this pool.
729c563d 40 * @param filePath Path to an implementation of a `ClusterWorker` file, which can be relative or absolute.
35cf1c03 41 * @param [opts={}] Options for this fixed cluster pool.
4ade5f1f
S
42 */
43 public constructor (
5c5a1fb7 44 numberOfWorkers: number,
c97c7edb 45 filePath: string,
1927ee67 46 public readonly opts: ClusterPoolOptions = {}
4ade5f1f 47 ) {
5c5a1fb7 48 super(numberOfWorkers, filePath, opts)
c97c7edb 49 }
4ade5f1f 50
6e9d10db 51 /** @inheritdoc */
c97c7edb 52 protected setupHook (): void {
325f50bc
S
53 setupMaster({
54 exec: this.filePath
55 })
c97c7edb 56 }
325f50bc 57
6e9d10db 58 /** @inheritdoc */
c97c7edb
S
59 protected isMain (): boolean {
60 return isMaster
4ade5f1f
S
61 }
62
a35560ba
S
63 /** @inheritdoc */
64 public destroyWorker (worker: Worker): void {
cefac5ba 65 this.sendToWorker(worker, { kill: 1 })
c97c7edb 66 worker.kill()
4ade5f1f
S
67 }
68
6e9d10db 69 /** @inheritdoc */
c97c7edb
S
70 protected sendToWorker (worker: Worker, message: MessageValue<Data>): void {
71 worker.send(message)
4ade5f1f
S
72 }
73
a35560ba
S
74 /** @inheritdoc */
75 public registerWorkerMessageListener<Message extends Data | Response> (
4f7fa42a
S
76 worker: Worker,
77 listener: (message: MessageValue<Message>) => void
c97c7edb 78 ): void {
4f7fa42a 79 worker.on('message', listener)
c97c7edb
S
80 }
81
6e9d10db 82 /** @inheritdoc */
280c2a77 83 protected createWorker (): Worker {
c97c7edb 84 return fork(this.opts.env)
4ade5f1f
S
85 }
86
6e9d10db 87 /** @inheritdoc */
280c2a77 88 protected afterWorkerSetup (worker: Worker): void {
1927ee67 89 // Listen worker messages.
be0676b3 90 this.registerWorkerMessageListener(worker, super.workerListener())
4ade5f1f 91 }
7c0ba920
JB
92
93 /** @inheritdoc */
94 public get type (): PoolType {
95 return PoolType.FIXED
96 }
97
98 /** @inheritdoc */
99 public get busy (): boolean {
100 return this.internalGetBusyStatus()
101 }
4ade5f1f 102}