src/worker/thread-worker.ts

   1 import { isMainThread, parentPort } from 'worker_threads'
   2 import type { JSONValue, MessageValue } from '../utility-types'
   3 import { AbstractWorker } from './abstract-worker'
   4 import type { WorkerOptions } from './worker-options'
   5
   6 /**
   7  * A thread worker used by a poolifier `ThreadPool`.
   8  *
   9  * When this worker is inactive for more than the given `maxInactiveTime`,
  10  * it will send a termination request to its main thread.
  11  *
  12  * If you use a `DynamicThreadPool` the extra workers that were created will be terminated,
  13  * but the minimum number of workers will be guaranteed.
  14  *
  15  * @template Data Type of data this worker receives from pool's execution.
  16  * @template Response Type of response the worker sends back to the main thread.
  17  *
  18  * @author [Alessandro Pio Ardizio](https://github.com/pioardi)
  19  * @since 0.0.1
  20  */
  21 export class ThreadWorker<
  22   Data extends JSONValue = JSONValue,
  23   Response extends JSONValue = JSONValue
  24 > extends AbstractWorker<MessagePort, Data, Response> {
  25   /**
  26    * Reference to main thread.
  27    */
  28   protected parent?: MessagePort
  29
  30   /**
  31    * Constructs a new poolifier thread worker.
  32    *
  33    * @param fn Function processed by the worker when the pool's `execution` function is invoked.
  34    * @param opts Options for the worker.
  35    */
  36   public constructor (fn: (data: Data) => Response, opts: WorkerOptions = {}) {
  37     super('worker-thread-pool:pioardi', isMainThread, fn, opts)
  38
  39     parentPort?.on('message', (value: MessageValue<Data>) => {
  40       if (value?.data && value.id) {
  41         // here you will receive messages
  42         // console.log('This is the main worker ' + isMainThread)
  43         if (this.async) {
  44           this.runInAsyncScope(this.runAsync.bind(this), this, fn, value)
  45         } else {
  46           this.runInAsyncScope(this.run.bind(this), this, fn, value)
  47         }
  48       } else if (value.parent) {
  49         // save the port to communicate with the main thread
  50         // this will be received once
  51         this.parent = value.parent
  52       } else if (value.kill) {
  53         // here is time to kill this worker, just clearing the interval
  54         if (this.interval) clearInterval(this.interval)
  55         this.emitDestroy()
  56       }
  57     })
  58   }
  59
  60   protected getMainWorker (): MessagePort {
  61     if (!this.parent) {
  62       throw new Error('Parent was not set')
  63     }
  64     return this.parent
  65   }
  66
  67   protected sendToMainWorker (message: MessageValue<Response>): void {
  68     this.getMainWorker().postMessage(message)
  69   }
  70 }