Merge pull request #2174 from poolifier/task-functions-properties
[poolifier.git] / src / worker / thread-worker.ts
CommitLineData
f59e1027 1import {
f59e1027 2 isMainThread,
ded253e2 3 type MessagePort,
f59e1027
JB
4 parentPort,
5 threadId
6} from 'node:worker_threads'
ded253e2 7
d35e5717
JB
8import type { MessageValue } from '../utility-types.js'
9import { AbstractWorker } from './abstract-worker.js'
d35e5717 10import type { TaskFunction, TaskFunctions } from './task-functions.js'
ded253e2 11import type { WorkerOptions } from './worker-options.js'
a32e02ba 12
a32e02ba 13/**
729c563d 14 * A thread worker used by a poolifier `ThreadPool`.
4ade5f1f 15 *
729c563d
S
16 * When this worker is inactive for more than the given `maxInactiveTime`,
17 * it will send a termination request to its main thread.
18 *
19 * If you use a `DynamicThreadPool` the extra workers that were created will be terminated,
20 * but the minimum number of workers will be guaranteed.
21 *
e102732c
JB
22 * @typeParam Data - Type of data this worker receives from pool's execution. This can only be structured-cloneable data.
23 * @typeParam Response - Type of response the worker sends back to the main thread. This can only be structured-cloneable data.
4ade5f1f 24 * @author [Alessandro Pio Ardizio](https://github.com/pioardi)
a32e02ba 25 * @since 0.0.1
26 */
d3c8a1a8 27export class ThreadWorker<
deb85c12
JB
28 Data = unknown,
29 Response = unknown
d3c8a1a8 30> extends AbstractWorker<MessagePort, Data, Response> {
85aeb3f3 31 /**
2761efb4 32 * Message port used to communicate with the main worker.
85aeb3f3 33 */
9b5c72ff 34 private port?: MessagePort
49890030 35
729c563d
S
36 /**
37 * Constructs a new poolifier thread worker.
38 *
82888165 39 * @param taskFunctions - Task function(s) processed by the worker when the pool's `execution` function is invoked.
38e795c1 40 * @param opts - Options for the worker.
729c563d 41 */
d4aeae5a 42 public constructor (
82ea6492 43 taskFunctions: TaskFunction<Data, Response> | TaskFunctions<Data, Response>,
d4aeae5a
JB
44 opts: WorkerOptions = {}
45 ) {
c63a35a0 46 super(isMainThread, parentPort, taskFunctions, opts)
f59e1027
JB
47 }
48
85aeb3f3
JB
49 /** @inheritDoc */
50 protected handleReadyMessage (message: MessageValue<Data>): void {
51 if (
52 message.workerId === this.id &&
f05ed93c 53 message.ready === false &&
85aeb3f3
JB
54 message.port != null
55 ) {
f05ed93c
JB
56 try {
57 this.port = message.port
58 this.port.on('message', this.messageListener.bind(this))
a5d15204
JB
59 this.sendToMainWorker({
60 ready: true,
31847469 61 taskFunctionsProperties: this.listTaskFunctionsProperties()
a5d15204 62 })
f05ed93c 63 } catch {
a5d15204
JB
64 this.sendToMainWorker({
65 ready: false,
31847469 66 taskFunctionsProperties: this.listTaskFunctionsProperties()
a5d15204 67 })
f05ed93c 68 }
85aeb3f3
JB
69 }
70 }
71
984dc9c8 72 /** @inheritDoc */
d655027b 73 protected handleKillMessage (message: MessageValue<Data>): void {
984dc9c8
JB
74 super.handleKillMessage(message)
75 this.port?.unref()
76 this.port?.close()
77 }
78
85aeb3f3 79 /** @inheritDoc */
f59e1027
JB
80 protected get id (): number {
81 return threadId
106744f7 82 }
7784f548 83
afc003b2 84 /** @inheritDoc */
803f948f
JB
85 protected readonly sendToMainWorker = (
86 message: MessageValue<Response>
87 ): void => {
fea198e8
JB
88 this.port?.postMessage({
89 ...message,
90 workerId: this.id
91 } satisfies MessageValue<Response>)
7784f548 92 }
985d0e79 93
34d8846f
JB
94 /**
95 * @inheritDoc
96 * @override
97 */
646bced2
JB
98 protected handleError (error: Error | string): string {
99 return error as string
985d0e79 100 }
a32e02ba 101}