Use the current node.js LTS version for publication. (#228)
[poolifier.git] / src / worker / abstract-worker.ts
CommitLineData
c97c7edb 1import { AsyncResource } from 'async_hooks'
838898f1
S
2import type { Worker } from 'cluster'
3import type { MessagePort } from 'worker_threads'
1a81f8af
S
4import type { MessageValue } from '../utility-types'
5import type { KillBehavior, WorkerOptions } from './worker-options'
6import { KillBehaviors } from './worker-options'
4c35177b 7
1a81f8af
S
8const DEFAULT_MAX_INACTIVE_TIME = 1000 * 60
9const DEFAULT_KILL_BEHAVIOR: KillBehavior = KillBehaviors.SOFT
c97c7edb 10
729c563d
S
11/**
12 * Base class containing some shared logic for all poolifier workers.
13 *
14 * @template MainWorker Type of main worker.
deb85c12
JB
15 * @template Data Type of data this worker receives from pool's execution. This can only be serializable data.
16 * @template Response Type of response the worker sends back to the main worker. This can only be serializable data.
729c563d 17 */
c97c7edb 18export abstract class AbstractWorker<
838898f1 19 MainWorker extends Worker | MessagePort,
d3c8a1a8
S
20 Data = unknown,
21 Response = unknown
c97c7edb 22> extends AsyncResource {
729c563d
S
23 /**
24 * The maximum time to keep this worker alive while idle. The pool automatically checks and terminates this worker when the time expires.
25 */
c97c7edb 26 protected readonly maxInactiveTime: number
4c35177b 27 /**
28 * The kill behavior set as option on the Worker constructor or a default value.
29 */
fb41d7f7 30 protected readonly killBehavior: KillBehavior
729c563d
S
31 /**
32 * Whether the worker is working asynchronously or not.
33 */
c97c7edb 34 protected readonly async: boolean
729c563d
S
35 /**
36 * Timestamp of the last task processed by this worker.
37 */
c97c7edb 38 protected lastTask: number
729c563d
S
39 /**
40 * Handler ID of the `interval` alive check.
41 */
c97c7edb
S
42 protected readonly interval?: NodeJS.Timeout
43
44 /**
729c563d 45 * Constructs a new poolifier worker.
c97c7edb
S
46 *
47 * @param type The type of async event.
729c563d
S
48 * @param isMain Whether this is the main worker or not.
49 * @param fn Function processed by the worker when the pool's `execution` function is invoked.
838898f1 50 * @param mainWorker Reference to main worker.
729c563d 51 * @param opts Options for the worker.
c97c7edb
S
52 */
53 public constructor (
54 type: string,
55 isMain: boolean,
56 fn: (data: Data) => Response,
838898f1 57 protected mainWorker?: MainWorker | null,
4c35177b 58 public readonly opts: WorkerOptions = {
1a81f8af
S
59 killBehavior: DEFAULT_KILL_BEHAVIOR,
60 maxInactiveTime: DEFAULT_MAX_INACTIVE_TIME
4c35177b 61 }
c97c7edb
S
62 ) {
63 super(type)
1a81f8af
S
64 this.killBehavior = this.opts.killBehavior ?? DEFAULT_KILL_BEHAVIOR
65 this.maxInactiveTime =
66 this.opts.maxInactiveTime ?? DEFAULT_MAX_INACTIVE_TIME
c97c7edb
S
67 this.async = !!this.opts.async
68 this.lastTask = Date.now()
c510fea7 69 this.checkFunctionInput(fn)
838898f1 70 // Keep the worker active
c97c7edb
S
71 if (!isMain) {
72 this.interval = setInterval(
73 this.checkAlive.bind(this),
74 this.maxInactiveTime / 2
75 )
76 this.checkAlive.bind(this)()
77 }
838898f1
S
78
79 this.mainWorker?.on('message', (value: MessageValue<Data, MainWorker>) => {
80 if (value?.data && value.id) {
81 // Here you will receive messages
82 if (this.async) {
83 this.runInAsyncScope(this.runAsync.bind(this), this, fn, value)
84 } else {
85 this.runInAsyncScope(this.run.bind(this), this, fn, value)
86 }
87 } else if (value.parent) {
88 // Save a reference of the main worker to communicate with it
89 // This will be received once
90 this.mainWorker = value.parent
91 } else if (value.kill) {
92 // Here is time to kill this worker, just clearing the interval
93 if (this.interval) clearInterval(this.interval)
94 this.emitDestroy()
95 }
96 })
c97c7edb
S
97 }
98
c510fea7
APA
99 /**
100 * Check if the `fn` parameter is passed to the constructor.
101 *
102 * @param fn The function that should be defined.
103 */
a35560ba 104 private checkFunctionInput (fn: (data: Data) => Response): void {
c510fea7
APA
105 if (!fn) throw new Error('fn parameter is mandatory')
106 }
107
729c563d
S
108 /**
109 * Returns the main worker.
838898f1
S
110 *
111 * @returns Reference to the main worker.
729c563d 112 */
838898f1
S
113 protected getMainWorker (): MainWorker {
114 if (!this.mainWorker) {
115 throw new Error('Main worker was not set')
116 }
117 return this.mainWorker
118 }
c97c7edb 119
729c563d
S
120 /**
121 * Send a message to the main worker.
122 *
123 * @param message The response message.
124 */
c97c7edb
S
125 protected abstract sendToMainWorker (message: MessageValue<Response>): void
126
729c563d
S
127 /**
128 * Check to see if the worker should be terminated, because its living too long.
129 */
c97c7edb 130 protected checkAlive (): void {
8bc77620 131 if (Date.now() - this.lastTask > this.maxInactiveTime) {
4c35177b 132 this.sendToMainWorker({ kill: this.killBehavior })
c97c7edb
S
133 }
134 }
135
729c563d
S
136 /**
137 * Handle an error and convert it to a string so it can be sent back to the main worker.
138 *
139 * @param e The error raised by the worker.
50eceb07 140 * @returns Message of the error.
729c563d 141 */
c97c7edb
S
142 protected handleError (e: Error | string): string {
143 return (e as unknown) as string
144 }
145
729c563d
S
146 /**
147 * Run the given function synchronously.
148 *
149 * @param fn Function that will be executed.
150 * @param value Input data for the given function.
151 */
c97c7edb
S
152 protected run (
153 fn: (data?: Data) => Response,
154 value: MessageValue<Data>
155 ): void {
156 try {
157 const res = fn(value.data)
158 this.sendToMainWorker({ data: res, id: value.id })
159 this.lastTask = Date.now()
160 } catch (e) {
161 const err = this.handleError(e)
162 this.sendToMainWorker({ error: err, id: value.id })
163 this.lastTask = Date.now()
164 }
165 }
166
729c563d
S
167 /**
168 * Run the given function asynchronously.
169 *
170 * @param fn Function that will be executed.
171 * @param value Input data for the given function.
172 */
c97c7edb
S
173 protected runAsync (
174 fn: (data?: Data) => Promise<Response>,
175 value: MessageValue<Data>
176 ): void {
177 fn(value.data)
178 .then(res => {
179 this.sendToMainWorker({ data: res, id: value.id })
180 this.lastTask = Date.now()
181 return null
182 })
183 .catch(e => {
184 const err = this.handleError(e)
185 this.sendToMainWorker({ error: err, id: value.id })
186 this.lastTask = Date.now()
187 })
188 }
189}