src/charging-station/Worker.ts

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