lib/workers.js

   1 'use strict'
   2 const {
   3   isMainThread, parentPort
   4 } = require('worker_threads')
   5 const maxInactiveTime = 1000 * 60
   6
   7 /**
   8  * An example worker that will be always alive, you just need to extend this class if you want a static pool.
   9  * @author Alessandro Pio Ardizio
  10  * @since 0.0.1
  11  */
  12 class ThreadWorker {
  13   constructor (fn) {
  14     if (!fn) throw new Error('Fn parameter is mandatory')
  15     // keep the worker active
  16     if (!isMainThread) {
  17       this.interval =
  18       setInterval(() => {
  19       }, 10000)
  20     }
  21     parentPort.on('message', (value) => {
  22       if (value.parent) {
  23         // save the port to communicate with the main thread
  24         this.parent = value.parent
  25       } else if (value && value._id) {
  26         // console.log('This is the main thread ' + isMainThread)
  27         this.parent.postMessage({ data: fn(value), _id: value._id })
  28       }
  29     })
  30   }
  31 }
  32
  33 /**
  34  * An example worker that will be always alive, you just need to extend this class if you want a static pool.<br>
  35  * When this worker is inactive for more than 1 minute, it will send this info to the main thread,<br>
  36  * if you are using DynamicThreadPool, the workers created after will be killed, the min num of thread will be guaranteed
  37  * @author Alessandro Pio Ardizio
  38  * @since 0.0.1
  39  */
  40 class DynamicWorker {
  41   constructor (fn) {
  42     this.lastTask = Date.now()
  43     if (!fn) throw new Error('Fn parameter is mandatory')
  44     // keep the worker active
  45     if (!isMainThread) {
  46       this.interval = setInterval(this._checkAlive.bind(this), maxInactiveTime)
  47       this._checkAlive.bind(this)()
  48     }
  49     parentPort.on('message', (value) => {
  50       if (value && value._id) {
  51         // here you will receive messages
  52         // console.log('This is the main thread ' + isMainThread)
  53         this.parent.postMessage({ data: fn(value), _id: value._id })
  54         this.lastTask = Date.now()
  55       } else if (value.parent) {
  56         // save the port to communicate with the main thread
  57         // this will be received once
  58         this.parent = value.parent
  59       } else if (value.kill) {
  60         // here is time to kill this thread, just clearing the interval
  61         clearInterval(this.interval)
  62       }
  63     })
  64   }
  65
  66   _checkAlive () {
  67     if ((Date.now() - this.lastTask) > maxInactiveTime) {
  68       this.parent.postMessage({ kill: 1 })
  69     }
  70   }
  71 }
  72
  73 module.exports.ThreadWorker = ThreadWorker
  74 module.exports.DynamicWorker = DynamicWorker