src/charging-station/Worker.js

   1 const Configuration = require('../utils/Configuration');
   2 const EventEmitter = require('events');
   3 const {Worker} = require('worker_threads');
   4 const Pool = require('worker-threads-pool');
   5
   6 class Wrk {
   7   /**
   8    * Create a new `Wrk`.
   9    *
  10    * @param {String} workerScript
  11    * @param {Object} workerData
  12    * @param {Number} numConcurrentWorkers
  13    */
  14   constructor(workerScript, workerData, numConcurrentWorkers) {
  15     this._workerData = workerData;
  16     this._workerScript = workerScript;
  17     this._numConcurrentWorkers = numConcurrentWorkers;
  18     if (Configuration.useWorkerPool()) {
  19       this._pool = new Pool({max: Configuration.getWorkerPoolSize()});
  20     }
  21   }
  22
  23   /**
  24    * @param {Number} numConcurrentWorkers
  25    * @private
  26    */
  27   // eslint-disable-next-line class-methods-use-this
  28   set _numConcurrentWorkers(numConcurrentWorkers) {
  29     if (numConcurrentWorkers > 10) {
  30       EventEmitter.defaultMaxListeners = numConcurrentWorkers + 1;
  31     }
  32     this._concurrentWorkers = numConcurrentWorkers;
  33   }
  34
  35   // eslint-disable-next-line class-methods-use-this
  36   get _numConcurrentWorkers() {
  37     return this._concurrentWorkers;
  38   }
  39
  40   /**
  41    *
  42    * @return {Promise}
  43    * @private
  44    */
  45   _startWorkerWithPool() {
  46     return new Promise((resolve, reject) => {
  47       this._pool.acquire(this._workerScript, {workerData: this._workerData}, (err, worker) => {
  48         if (err) {
  49           return reject(err);
  50         }
  51         worker.once('message', resolve);
  52         worker.once('error', reject);
  53       });
  54     });
  55   }
  56
  57   /**
  58    *
  59    * @return {Promise}
  60    * @private
  61    */
  62   _startWorker() {
  63     return new Promise((resolve, reject) => {
  64       const worker = new Worker(this._workerScript, {workerData: this._workerData});
  65       worker.on('message', resolve);
  66       worker.on('error', reject);
  67       worker.on('exit', (code) => {
  68         if (code !== 0) {
  69           reject(new Error(`Worker stopped with exit code ${code}`));
  70         }
  71       });
  72     });
  73   }
  74
  75   /**
  76    *
  77    * @return {Promise}
  78    * @public
  79    */
  80   start() {
  81     if (Configuration.useWorkerPool()) {
  82       return this._startWorkerWithPool();
  83     }
  84     return this._startWorker();
  85   }
  86 }
  87
  88 module.exports = Wrk;