src/charging-station/Worker.ts

   1 import Configuration from '../utils/Configuration';
   2 import Constants from '../utils/Constants';
   3 import { Worker } from 'worker_threads';
   4 import WorkerData from '../types/WorkerData';
   5 import WorkerPool from './WorkerPool';
   6
   7 export default class Wrk {
   8   private workerScript: string;
   9   private workerData: WorkerData;
  10   private worker: Worker;
  11   private maxWorkerElements: number;
  12   private numWorkerElements: number;
  13
  14   /**
  15    * Create a new `Wrk`.
  16    *
  17    * @param {string} workerScript
  18    * @param {WorkerData} workerData
  19    * @param {number} maxWorkerElements
  20    */
  21   constructor(workerScript: string, workerData: WorkerData, maxWorkerElements = 1) {
  22     this.workerData = workerData;
  23     this.workerScript = workerScript;
  24     if (Configuration.useWorkerPool()) {
  25       WorkerPool.maxConcurrentWorkers = Configuration.getWorkerPoolMaxSize();
  26     } else {
  27       this.maxWorkerElements = maxWorkerElements;
  28       this.numWorkerElements = 0;
  29     }
  30   }
  31
  32   /**
  33    *
  34    * @return {Promise}
  35    * @public
  36    */
  37   async start(): Promise<Worker> {
  38     if (Configuration.useWorkerPool()) {
  39       await this.startWorkerPool();
  40     } else {
  41       await this.startWorker();
  42     }
  43     return this.worker;
  44   }
  45
  46   /**
  47    *
  48    * @return {void}
  49    * @public
  50    */
  51   addWorkerElement(workerData: WorkerData): void {
  52     if (Configuration.useWorkerPool()) {
  53       throw Error('Cannot add Wrk element if the worker pool is enabled');
  54     }
  55     if (this.numWorkerElements >= this.maxWorkerElements) {
  56       throw Error('Cannot add Wrk element: max number of elements per worker reached');
  57     }
  58     this.workerData = workerData;
  59     this.worker.postMessage({ id: Constants.START_WORKER_ELEMENT, workerData: workerData });
  60     this.numWorkerElements++;
  61   }
  62
  63   /**
  64    *
  65    * @return {number}
  66    * @public
  67    */
  68   public getWorkerPoolSize(): number {
  69     if (Configuration.useWorkerPool()) {
  70       return WorkerPool.getPoolSize();
  71     }
  72   }
  73
  74   /**
  75    *
  76    * @return {Promise}
  77    * @private
  78    */
  79   private async startWorkerPool() {
  80     return new Promise((resolve, reject) => {
  81       WorkerPool.acquire(this.workerScript, { workerData: this.workerData }, (err, worker) => {
  82         if (err) {
  83           return reject(err);
  84         }
  85         worker.once('message', resolve);
  86         worker.once('error', reject);
  87         this.worker = worker;
  88       });
  89     });
  90   }
  91
  92   /**
  93    *
  94    * @return {Promise}
  95    * @private
  96    */
  97   private async startWorker() {
  98     return new Promise((resolve, reject) => {
  99       const worker = new Worker(this.workerScript, { workerData: this.workerData });
 100       worker.on('message', resolve);
 101       worker.on('error', reject);
 102       worker.on('exit', (code) => {
 103         if (code !== 0) {
 104           reject(new Error(`Worker stopped with exit code ${code}`));
 105         }
 106       });
 107       this.numWorkerElements++;
 108       this.worker = worker;
 109     });
 110   }
 111 }