dtp-core/lib/site-worker.js

// site-worker.js
// Copyright (C) 2022 DTP Technologies, LLC
// License: Apache-2.0

'use strict';

const path = require('path');

const SitePlatform = require(path.join(__dirname, 'site-platform'));

const { SiteAsync } = require(path.join(__dirname, 'site-async'));
const { SiteCommon } = require(path.join(__dirname, 'site-common'));

class SiteWorker extends SiteCommon {

  constructor (dtp, component) {
    super(dtp, component);
    this.processors = { };
  }

  async start ( ) {
    try {
      process.on('unhandledRejection', async (error, p) => {
        this.log.error('Unhandled rejection', {
          promise: p,
          message: error.message,
          stack: error.stack
        });
        process.exit(-2);
      });
      
      process.on('warning', (error) => {
        this.log.alert('warning', { error });
      });
  
      process.once('SIGINT', async ( ) => {
        this.log.info('SIGINT received');
        this.log.info('requesting shutdown...');
  
        await this.stop();

        const exitCode = await SitePlatform.shutdown();
        process.nextTick(( ) => {
          process.exit(exitCode);
        });
      });
  
      /*
       * Site Platform startup
       */
      await SitePlatform.startPlatform(this.dtp);
    } catch (error) {
      this.log.error('failed to start worker', {
        component: this.dtp.config.component,
        error,
      });
      process.exit(-1);
    }
  }

  /**
   * Load a script as a Worker processor. It must be derived from
   * SiteWorkerProcess and implement the expected interface.
   * @param {String} scriptFile the filename of the script to load as a Worker
   * processor.
   * @returns the new processor instance
   */
  async loadProcessor (scriptFile) {
    const ProcessorClass = require(scriptFile);
    const processor = new ProcessorClass(this);
    const { COMPONENT } = ProcessorClass;

    this.log.info('loading worker processor', { component: COMPONENT.name });
    this.processors[COMPONENT.name] = processor;

    return processor;
  }

  /**
   * Start all loaded processors. The assumption here is if you load any
   * additional processors *after* calling this method, you will start them
   * yourself in some way.
   */
  async startProcessors ( ) {
    const slugs = Object.keys(this.processors);
    await SiteAsync.each(slugs, async (slug) => {
      const processor = this.processors[slug];
      try {
        this.log.info('starting worker processor', {
          component: processor.component.name,
        });
        await processor.start();
      } catch (error) {
        this.log.error('failed to start processor', {
          component: processor.component.name,
          error,
        });
      }
    }, 1);
  }

  /**
   * Stops any running child processors and terminates the worker.
   */
   async stop ( ) {
    const slugs = Object.keys(this.processors);
    await SiteAsync.each(slugs, async (slug) => {
      const processor = this.processors[slug];
      try {
        this.log.info('stopping worker processor', {
          component: processor.component.name,
        });
        await processor.stop();
      } catch (error) {
        this.log.error('failed to stop processor', {
          component: processor.component.name,
          error,
        });
      }
    }, 1);
  }
}

module.exports.SiteWorker = SiteWorker;