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