---
[poolifier.git] / src / worker / thread-worker.ts
1 import {
2 isMainThread,
3 type MessagePort,
4 parentPort,
5 threadId
6 } from 'node:worker_threads'
7
8 import type { MessageValue } from '../utility-types.js'
9 import { AbstractWorker } from './abstract-worker.js'
10 import type { TaskFunction, TaskFunctions } from './task-functions.js'
11 import type { WorkerOptions } from './worker-options.js'
12
13 /**
14 * A thread worker used by a poolifier `ThreadPool`.
15 *
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 *
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.
24 * @author [Alessandro Pio Ardizio](https://github.com/pioardi)
25 * @since 0.0.1
26 */
27 export class ThreadWorker<
28 Data = unknown,
29 Response = unknown
30 > extends AbstractWorker<MessagePort, Data, Response> {
31 /**
32 * Message port used to communicate with the main worker.
33 */
34 private port?: MessagePort
35
36 /**
37 * Constructs a new poolifier thread worker.
38 *
39 * @param taskFunctions - Task function(s) processed by the worker when the pool's `execution` function is invoked.
40 * @param opts - Options for the worker.
41 */
42 public constructor (
43 taskFunctions: TaskFunction<Data, Response> | TaskFunctions<Data, Response>,
44 opts: WorkerOptions = {}
45 ) {
46 super(isMainThread, parentPort, taskFunctions, opts)
47 }
48
49 /** @inheritDoc */
50 protected handleReadyMessage (message: MessageValue<Data>): void {
51 if (
52 message.workerId === this.id &&
53 message.ready === false &&
54 message.port != null
55 ) {
56 try {
57 this.port = message.port
58 this.port.on('message', this.messageListener.bind(this))
59 this.sendToMainWorker({
60 ready: true,
61 taskFunctionsProperties: this.listTaskFunctionsProperties()
62 })
63 } catch {
64 this.sendToMainWorker({
65 ready: false,
66 taskFunctionsProperties: this.listTaskFunctionsProperties()
67 })
68 }
69 }
70 }
71
72 /** @inheritDoc */
73 protected handleKillMessage (message: MessageValue<Data>): void {
74 super.handleKillMessage(message)
75 this.port?.unref()
76 this.port?.close()
77 }
78
79 /** @inheritDoc */
80 protected get id (): number {
81 return threadId
82 }
83
84 /** @inheritDoc */
85 protected readonly sendToMainWorker = (
86 message: MessageValue<Response>
87 ): void => {
88 this.port?.postMessage({
89 ...message,
90 workerId: this.id
91 } satisfies MessageValue<Response>)
92 }
93
94 /**
95 * @inheritDoc
96 * @override
97 */
98 protected handleError (error: Error | string): string {
99 return error as string
100 }
101 }