fix: fix worker function type definition and validation
[poolifier.git] / src / worker / abstract-worker.ts
CommitLineData
fc3e6586
JB
1import { AsyncResource } from 'node:async_hooks'
2import type { Worker } from 'node:cluster'
3import type { MessagePort } from 'node:worker_threads'
1a81f8af 4import type { MessageValue } from '../utility-types'
6e9d10db 5import { EMPTY_FUNCTION } from '../utils'
5919b303 6import type { KillBehavior, WorkerOptions } from './worker-options'
1a81f8af 7import { KillBehaviors } from './worker-options'
4c35177b 8
978aad6f 9const DEFAULT_MAX_INACTIVE_TIME = 60000
1a81f8af 10const DEFAULT_KILL_BEHAVIOR: KillBehavior = KillBehaviors.SOFT
c97c7edb 11
729c563d 12/**
ea7a90d3 13 * Base class that implements some shared logic for all poolifier workers.
729c563d 14 *
38e795c1
JB
15 * @typeParam MainWorker - Type of main worker.
16 * @typeParam Data - Type of data this worker receives from pool's execution. This can only be serializable data.
17 * @typeParam Response - Type of response the worker sends back to the main worker. This can only be serializable data.
729c563d 18 */
c97c7edb 19export abstract class AbstractWorker<
838898f1 20 MainWorker extends Worker | MessagePort,
d3c8a1a8
S
21 Data = unknown,
22 Response = unknown
c97c7edb 23> extends AsyncResource {
729c563d
S
24 /**
25 * Timestamp of the last task processed by this worker.
26 */
a9d9ea34 27 protected lastTaskTimestamp!: number
729c563d 28 /**
aee46736 29 * Handler id of the `aliveInterval` worker alive check.
729c563d 30 */
e088a00c 31 protected readonly aliveInterval?: NodeJS.Timeout
c97c7edb 32 /**
729c563d 33 * Constructs a new poolifier worker.
c97c7edb 34 *
38e795c1
JB
35 * @param type - The type of async event.
36 * @param isMain - Whether this is the main worker or not.
37 * @param fn - Function processed by the worker when the pool's `execution` function is invoked.
38 * @param mainWorker - Reference to main worker.
39 * @param opts - Options for the worker.
c97c7edb
S
40 */
41 public constructor (
42 type: string,
c2ade475 43 protected readonly isMain: boolean,
d4aeae5a 44 fn: (data: Data) => Response | Promise<Response>,
7e0d447f 45 protected mainWorker: MainWorker | undefined | null,
d99ba5a8 46 protected readonly opts: WorkerOptions = {
e088a00c 47 /**
aee46736 48 * The kill behavior option on this worker or its default value.
e088a00c 49 */
1a81f8af 50 killBehavior: DEFAULT_KILL_BEHAVIOR,
e088a00c
JB
51 /**
52 * The maximum time to keep this worker alive while idle.
53 * The pool automatically checks and terminates this worker when the time expires.
54 */
1a81f8af 55 maxInactiveTime: DEFAULT_MAX_INACTIVE_TIME
4c35177b 56 }
c97c7edb
S
57 ) {
58 super(type)
e088a00c 59 this.checkWorkerOptions(this.opts)
d4aeae5a 60 this.checkFunctionInput(fn)
7c24d88b 61 if (!this.isMain) {
3fafb1b2 62 this.lastTaskTimestamp = performance.now()
e088a00c 63 this.aliveInterval = setInterval(
c97c7edb 64 this.checkAlive.bind(this),
e088a00c 65 (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME) / 2
c97c7edb
S
66 )
67 this.checkAlive.bind(this)()
68 }
838898f1 69
aee46736
JB
70 this.mainWorker?.on(
71 'message',
72 (message: MessageValue<Data, MainWorker>) => {
73 this.messageListener(message, fn)
74 }
75 )
c97c7edb
S
76 }
77
aee46736
JB
78 /**
79 * Worker message listener.
80 *
81 * @param message - Message received.
82 * @param fn - Function processed by the worker when the pool's `execution` function is invoked.
83 */
cf597bc5 84 protected messageListener (
aee46736 85 message: MessageValue<Data, MainWorker>,
d4aeae5a 86 fn: (data: Data) => Response | Promise<Response>
cf597bc5 87 ): void {
a3f5f781 88 if (message.id != null && message.data != null) {
aee46736 89 // Task message received
6bd72cd0 90 if (this.opts.async === true) {
aee46736 91 this.runInAsyncScope(this.runAsync.bind(this), this, fn, message)
cf597bc5 92 } else {
aee46736 93 this.runInAsyncScope(this.run.bind(this), this, fn, message)
cf597bc5 94 }
aee46736
JB
95 } else if (message.parent != null) {
96 // Main worker reference message received
97 this.mainWorker = message.parent
98 } else if (message.kill != null) {
99 // Kill message received
73cff87e 100 this.aliveInterval != null && clearInterval(this.aliveInterval)
cf597bc5
JB
101 this.emitDestroy()
102 }
103 }
104
78cea37e 105 private checkWorkerOptions (opts: WorkerOptions): void {
e088a00c
JB
106 this.opts.killBehavior = opts.killBehavior ?? DEFAULT_KILL_BEHAVIOR
107 this.opts.maxInactiveTime =
108 opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME
78cea37e 109 this.opts.async = opts.async ?? false
e088a00c
JB
110 }
111
c510fea7 112 /**
8accb8d5 113 * Checks if the `fn` parameter is passed to the constructor.
c510fea7 114 *
38e795c1 115 * @param fn - The function that should be defined.
c510fea7 116 */
d4aeae5a
JB
117 private checkFunctionInput (
118 fn: (data: Data) => Response | Promise<Response>
119 ): void {
78cea37e 120 if (fn == null) throw new Error('fn parameter is mandatory')
af5204ed
JB
121 if (typeof fn !== 'function') {
122 throw new TypeError('fn parameter is not a function')
123 }
d4aeae5a
JB
124 if (fn.constructor.name === 'AsyncFunction' && this.opts.async === false) {
125 throw new Error(
126 'fn parameter is an async function, please set the async option to true'
127 )
128 }
c510fea7
APA
129 }
130
729c563d
S
131 /**
132 * Returns the main worker.
838898f1
S
133 *
134 * @returns Reference to the main worker.
729c563d 135 */
838898f1 136 protected getMainWorker (): MainWorker {
78cea37e 137 if (this.mainWorker == null) {
838898f1
S
138 throw new Error('Main worker was not set')
139 }
140 return this.mainWorker
141 }
c97c7edb 142
729c563d 143 /**
8accb8d5 144 * Sends a message to the main worker.
729c563d 145 *
38e795c1 146 * @param message - The response message.
729c563d 147 */
c97c7edb
S
148 protected abstract sendToMainWorker (message: MessageValue<Response>): void
149
729c563d 150 /**
a05c10de 151 * Checks if the worker should be terminated, because its living too long.
729c563d 152 */
c97c7edb 153 protected checkAlive (): void {
e088a00c 154 if (
3fafb1b2 155 performance.now() - this.lastTaskTimestamp >
e088a00c
JB
156 (this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME)
157 ) {
158 this.sendToMainWorker({ kill: this.opts.killBehavior })
c97c7edb
S
159 }
160 }
161
729c563d 162 /**
8accb8d5 163 * Handles an error and convert it to a string so it can be sent back to the main worker.
729c563d 164 *
38e795c1 165 * @param e - The error raised by the worker.
50eceb07 166 * @returns Message of the error.
729c563d 167 */
c97c7edb 168 protected handleError (e: Error | string): string {
4a34343d 169 return e as string
c97c7edb
S
170 }
171
729c563d 172 /**
8accb8d5 173 * Runs the given function synchronously.
729c563d 174 *
38e795c1 175 * @param fn - Function that will be executed.
aee46736 176 * @param message - Input data for the given function.
729c563d 177 */
c97c7edb
S
178 protected run (
179 fn: (data?: Data) => Response,
aee46736 180 message: MessageValue<Data>
c97c7edb
S
181 ): void {
182 try {
3fafb1b2 183 const startTimestamp = performance.now()
aee46736 184 const res = fn(message.data)
3fafb1b2
JB
185 const runTime = performance.now() - startTimestamp
186 this.sendToMainWorker({
187 data: res,
188 id: message.id,
189 runTime
190 })
c97c7edb 191 } catch (e) {
0a23f635 192 const err = this.handleError(e as Error)
aee46736 193 this.sendToMainWorker({ error: err, id: message.id })
6e9d10db 194 } finally {
3fafb1b2 195 !this.isMain && (this.lastTaskTimestamp = performance.now())
c97c7edb
S
196 }
197 }
198
729c563d 199 /**
8accb8d5 200 * Runs the given function asynchronously.
729c563d 201 *
38e795c1 202 * @param fn - Function that will be executed.
aee46736 203 * @param message - Input data for the given function.
729c563d 204 */
c97c7edb
S
205 protected runAsync (
206 fn: (data?: Data) => Promise<Response>,
aee46736 207 message: MessageValue<Data>
c97c7edb 208 ): void {
3fafb1b2 209 const startTimestamp = performance.now()
aee46736 210 fn(message.data)
c97c7edb 211 .then(res => {
3fafb1b2
JB
212 const runTime = performance.now() - startTimestamp
213 this.sendToMainWorker({
214 data: res,
215 id: message.id,
216 runTime
217 })
c97c7edb
S
218 return null
219 })
220 .catch(e => {
f3636726 221 const err = this.handleError(e as Error)
aee46736 222 this.sendToMainWorker({ error: err, id: message.id })
6e9d10db
JB
223 })
224 .finally(() => {
3fafb1b2 225 !this.isMain && (this.lastTaskTimestamp = performance.now())
c97c7edb 226 })
6e9d10db 227 .catch(EMPTY_FUNCTION)
c97c7edb
S
228 }
229}