webapi-eca/js/event-poller.js

// Generated by CoffeeScript 1.7.1

/*

Dynamic Modules
===============
> Compiles CoffeeScript modules and loads JS modules in a VM, together
> with only a few allowed node.js modules.
 */

(function() {
  var db, dynmod, encryption, fCallFunction, fCheckAndRun, fLoadModule, fPushEvent, isRunning, listUserModules, log, logconf, logger, pollLoop;

  logger = require('./logging');

  db = require('./persistence');

  dynmod = require('./dynamic-modules');

  encryption = require('./encryption');

  if (process.argv.length < 8) {
    console.error('Not all arguments have been passed!');
    process.exit();
  }

  logconf = {
    mode: process.argv[2],
    nolog: process.argv[6]
  };

  logconf['io-level'] = process.argv[3];

  logconf['file-level'] = process.argv[4];

  logconf['file-path'] = process.argv[5];

  log = logger.getLogger(logconf);

  log.info('EP | Event Poller starts up');

  db({
    logger: log
  });

  dynmod({
    logger: log
  });

  encryption({
    logger: log,
    keygen: process.argv[7]
  });

  listUserModules = {};

  isRunning = true;

  process.on('disconnect', function() {
    log.info('EP | Shutting down Event Poller');
    isRunning = false;
    return process.exit();
  });

  process.on('message', function(msg) {
    log.info("EP | Got info about new rule: " + msg.event);
    if (msg.event === 'new' || msg.event === 'init') {
      fLoadModule(msg);
    }
    if (msg.event === 'del') {
      delete listUserModules[msg.user][msg.ruleId];
      if (JSON.stringify(listUserModules[msg.user]) === "{}") {
        return delete listUserModules[msg.user];
      }
    }
  });

  fLoadModule = function(msg) {
    var arrName, fAnonymous;
    arrName = msg.rule.event.split(' -> ');
    fAnonymous = function() {
      return db.eventPollers.getModule(arrName[0], function(err, obj) {
        if (!obj) {
          return log.warn("EP | Strange... no module retrieved: " + arrName[0]);
        } else {
          return dynmod.compileString(obj.data, msg.user, msg.rule.id, arrName[0], obj.lang, db.eventPollers, function(result) {
            var oUser, ts;
            if (!result.answ === 200) {
              log.error("EP | Compilation of code failed! " + msg.user + ", " + msg.rule.id + ", " + arrName[0]);
            }
            if (!listUserModules[msg.user]) {
              listUserModules[msg.user] = {};
            }
            oUser = listUserModules[msg.user];
            ts = new Date();
            oUser[msg.rule.id] = {
              id: msg.rule.event,
              pollfunc: arrName[1],
              funcArgs: result.funcArgs,
              event_interval: msg.rule.event_interval * 60 * 1000,
              module: result.module,
              logger: result.logger,
              timestamp: ts
            };
            oUser[msg.rule.id].module.pushEvent = fPushEvent(msg.user, msg.rule.id, oUser[msg.rule.id]);
            log.info("EP | New event module '" + arrName[0] + "' loaded for user " + msg.user + ", in rule " + msg.rule.id + ", starting at UTC|" + (ts.toISOString()) + " and polling every " + msg.rule.event_interval + " minutes");
            return setTimeout(fCheckAndRun(msg.user, msg.rule.id, ts), 1000);
          });
        }
      });
    };
    if (msg.event === 'new' || !listUserModules[msg.user] || !listUserModules[msg.user][msg.rule.id]) {
      return fAnonymous();
    }
  };

  fPushEvent = function(userId, ruleId, oRule) {
    return function(obj) {
      return db.pushEvent({
        event: oRule.id,
        eventid: "polled " + oRule.id + " " + userId + "_UTC|" + ((new Date()).toISOString()),
        payload: obj
      });
    };
  };

  fCheckAndRun = function(userId, ruleId, timestamp) {
    return function() {
      var oRule;
      log.info("EP | Check and run user " + userId + ", rule " + ruleId);
      if (isRunning && listUserModules[userId] && listUserModules[userId][ruleId]) {
        if (listUserModules[userId][ruleId].timestamp === timestamp) {
          oRule = listUserModules[userId][ruleId];
          fCallFunction(userId, ruleId, oRule);
          return setTimeout(fCheckAndRun(userId, ruleId, timestamp), oRule.event_interval);
        } else {
          return log.info("EP | We found a newer polling interval and discontinue this one which was created at UTC|" + (timestamp.toISOString()));
        }
      }
    };
  };

  fCallFunction = function(userId, ruleId, oRule) {
    var arrArgs, err, oArg, _i, _len, _ref;
    try {
      arrArgs = [];
      if (oRule.funcArgs) {
        _ref = oRule.funcArgs[oRule.pollfunc];
        for (_i = 0, _len = _ref.length; _i < _len; _i++) {
          oArg = _ref[_i];
          arrArgs.push(oArg.value);
        }
      }
      return oRule.module[oRule.pollfunc].apply(null, arrArgs);
    } catch (_error) {
      err = _error;
      log.info("EP | ERROR in module when polled: " + oRule.id + " " + userId + ": " + err.message);
      return oRule.logger(err.message);
    }
  };


  /*
  This function will loop infinitely every 10 seconds until isRunning is set to false
  
  @private pollLoop()
   */

  pollLoop = function() {
    if (isRunning) {
      return setTimeout(pollLoop, 10000);
    }
  };

  pollLoop();

}).call(this);