arhs-atlasboard/lib/scheduler.js

var  generalConfigManager = require('./config-manager')(),
        loadGlobalAuth = require('./global-auth'),
        helpers = require('./helpers'),
        jobDependencyManager = require('./job-dependencies/loader.js');
/**
 * Scheduler
 * 
 * @param  {boolean} options.logSuccessfulJob logs output when job is executed succesfully (verbose)
 * @return {object}  scheduler
 */


var scheduler = {
    globalAuth: loadGlobalAuth(generalConfigManager.get('authentication-file-path')),
    domain: require('domain').create(),
    scheduleTasks:{},
    configurations:{},
    initialize: function() {
      this.domain.on("error", function(error) {
        console.error(error.stack);
      });
    },
    removeJob: function(configKey) {
      clearTimeout(this.scheduleTasks[configKey].timer);
    },
    updateConfig: function(configKey,config) {

      for(p in config) {
        this.configurations[configKey][p]  = config[p];
      }

      

      clearTimeout(this.scheduleTasks[configKey].timer);
      this.scheduleTasks[configKey].f();
    },
    scheduleNext: function(job_worker, widgets) {
      var thiz = this;
      var f = function(){
        thiz.schedule(job_worker, widgets);
      };

      var timer = setTimeout(f, job_worker.config.interval || 60 * 1000);


      if(this.scheduleTasks[job_worker.widget_item.config]) {
        clearTimeout(this.scheduleTasks[job_worker.widget_item.config].timer);
      }


      this.scheduleTasks[job_worker.widget_item.config] = {timer:timer, f:f};
    },

    schedule: function(job_worker, widgets) {

      function handleError(err) {
        job_worker.dependencies.logger.error('executed with errors: ' + err);

        // in case of error retry in one third of the original interval or 1 min, whatever is lower
        job_worker.config.interval = Math.min(job_worker.config.original_interval / 3, 60000);

        // -------------------------------------------------------------
        // Decide if we hold error notification according to widget config.
        // if the retryOnErrorTimes property found in config, the error notification
        // won´t be sent until we reach that number of consecutive errrors.
        // This will prevent showing too many error when connection to flaky, unreliable
        // servers.
        // -------------------------------------------------------------
        var sendError = true;        
        if (job_worker.firstRun === false) {
          if (job_worker.config.retryOnErrorTimes) {
            job_worker.retryOnErrorCounter = job_worker.retryOnErrorCounter || 0;
            if (job_worker.retryOnErrorCounter <= job_worker.config.retryOnErrorTimes) {
              job_worker.dependencies.logger.warn('widget with retryOnErrorTimes. attempts: '
                + job_worker.retryOnErrorCounter);
              sendError = false;
              job_worker.retryOnErrorCounter++;
            }
          }
        }
        else {
          // this is the first run for this job so if it fails, we want to inform immediately
          // since it may be a configuration or dev related problem.
          job_worker.firstRun = false;
        }

        if (sendError) {
          widgets.sendData({error: err, config: {interval: job_worker.config.interval}});
        }
      }

      var thiz = this;
      var task = this.domain.bind(job_worker.task);

      if (!job_worker.config.interval){
        job_worker.config.interval = 60 * 1000; // default to 60 secs if not provided
      }
      else if (job_worker.config.interval < 1000){
        job_worker.config.interval = 1000; // minium 1 sec
      }

      if (!job_worker.config.original_interval) // set original interval so we can get back to it
        job_worker.config.original_interval = job_worker.config.interval;


      // Save the job configuration in the singleton
      if(!this.configurations[job_worker.widget_item.config]) {
        this.configurations[job_worker.widget_item.config] = job_worker.config;
      } else {
        job_worker.config = this.configurations[job_worker.widget_item.config];
      }

      try {

        // TODO
        // - We are still passing job_worker.dependencies as a parameter for backwards compatibility
        // but makes more sense to be passed as a property of job_worker. 
        // - The same with config
        task.call(job_worker, job_worker.config, job_worker.dependencies, function(err, data){
          if (err) {
            handleError(err);
          }
          else {
            job_worker.retryOnErrorCounter = 0; //reset error counter on success
            if (!data) data = {};
            data.config = {interval: job_worker.config.interval}; // give access to interval to client
            widgets.sendData(data);
          }
          thiz.scheduleNext(job_worker, widgets);
        });
      }
      catch (e) {
        job_worker.dependencies.logger.error('Uncaught exception executing job: ' + e);
        handleError(e);
        thiz.scheduleNext(job_worker, widgets);
      }
    },

    scheduleFirst: function(jobWorker,withDelay) {

             var thiz = this;
                // unique id for this widget
            jobWorker.id = jobWorker.dashboard_name + "_" + jobWorker.widget_item.config + "_" +
                jobWorker.widget_item.widget + "_" + jobWorker.widget_item.job;

            var widgets = { sendData: function(data) { thiz.eventQueue.send(jobWorker.id, data); } };

            // add security info
            jobWorker.config.globalAuth = this.globalAuth;

            if (jobWorker.widget_item.enabled !== false){
              if (jobWorker.task){

                // introduce a random delay on job initialization to avoid a concurrency peak on start
                var rndDelay = helpers.getRndInt(0, 15000);

                jobDependencyManager.fillDependencies(jobWorker, this.io, generalConfigManager);

                setTimeout(function(){
                  //----------------------
                  // schedule job
                  //----------------------
                  thiz.schedule(jobWorker, widgets);

                }, withDelay ? rndDelay:0);

              }
              else{
                generalLogger.warn("no job task for " + eventId);
              }
            }
            else { // job is disabled
              widgets.sendData({error: 'disabled'});
            }
      }
  };

scheduler.initialize();

module.exports.scheduler = scheduler;