webapi-eca/js/engine.js

'use strict';

var path = require('path'),
    cp = require('child_process'),
    log = require('./logging'),
    qEvents = new (require('./queue')).Queue(), //TODO export queue into redis
    regex = /\$X\.[\w\.\[\]]*/g, // find properties of $X
    listRules = {},
    listActionModules = {},
    isRunning = true,
    actionsLoaded = false,
    eventsLoaded = false,
    ml, poller, db;

exports = module.exports = function(args) {
  args = args || {};
  log(args);
  ml = require('./module_loader')(args);
  poller = cp.fork(path.resolve(__dirname, 'eventpoller'), [log.getLogType()]);
  poller.on('message', function(evt) {
    if(evt.event === 'ep_finished_loading') {
      eventsLoaded = true;
      tryToLoadRules();
    } else exports.pushEvent(evt);
  });
  //start to poll the event queue
  pollQueue();
  return module.exports;
};

/*
 * Initialize the rules engine which initializes the module loader.
 * @param {Object} db_link the link to the db, see [db\_interface](db_interface.html)
 * @param {String} db_port the db port
 * @param {String} crypto_key the key to be used for encryption on the db, max legnth 256
 */
exports.addDBLink = function(db_link) { db = db_link; };

exports.loadActions = function(cb) {
  if(ml && db) db.getActionModules(function(err, obj) {
    if(err) log.error('EN', 'retrieving Action Modules from DB!');
    else {
      if(!obj) {
        log.print('EN', 'No Action Modules found in DB!');
        actionsLoaded = true;
        tryToLoadRules();
      } else {
        var m, semaphore = 0;
        for(var el in obj) {
          semaphore++;
          log.print('EN', 'Loading Action Module from DB: ' + el);
          m = ml.requireFromString(obj[el], el);
          db.getActionModuleAuth(el, function(mod) {
            return function(err, obj) {
              if(--semaphore == 0) {
                actionsLoaded = true;
                tryToLoadRules();
              }
              if(obj && mod.loadCredentials) mod.loadCredentials(JSON.parse(obj));
            };
          }(m));
          listActionModules[el] = m;
        }
      }
    }
  });
  else log.severe('EN', new Error('Module Loader not defined!'));
};

function tryToLoadRules() {
  if(db && eventsLoaded && actionsLoaded) {
    db.getRules(function(err, obj) {
      for(var el in obj) exports.loadRule(JSON.parse(obj[el]));
    });
  }
}

/**
 * Insert an action module into the list of available interfaces.
 * @param {Object} objModule the action module object
 */
exports.loadActionModule = function(name, objModule) {
  log.print('EN', 'Action module "' + name + '" loaded');
  listActionModules[name] = objModule;
};

/**
 * Insert a rule into the eca rules repository
 * @param {Object} objRule the rule object
 */
exports.loadRule = function(objRule) {
  //TODO validate rule
  log.print('EN', 'Loading Rule: ' + objRule.id);
  if(listRules[objRule.id]) log.print('EN', 'Replacing rule: ' + objRule.id);
  listRules[objRule.id] = objRule;

  // Notify poller about eventual candidate
  try {
    poller.send('event|'+objRule.event);
  } catch (err) {
    log.print('EN', 'Unable to inform poller about new active rule!');
  }
};

function pollQueue() {
  if(isRunning) {
    var evt = qEvents.dequeue();
    if(evt) {
      processEvent(evt);
    }
    setTimeout(pollQueue, 50); //TODO adapt to load
  }
}

/**
 * Stores correctly posted events in the queue
 * @param {Object} evt The event object
 */
exports.pushEvent = function(evt) {
  qEvents.enqueue(evt);
};

/**
 * Handles correctly posted events
 * @param {Object} evt The event object
 */
function processEvent(evt) {
  log.print('EN', 'processing event: ' + evt.event + '(' + evt.eventid + ')');
  var actions = checkEvent(evt);
  for(var i = 0; i < actions.length; i++) {
    invokeAction(evt, actions[i]);
  }
}

/**
 * Check an event against the rules repository and return the actions
 * if the conditons are met.
 * @param {Object} evt the event to check
 */
function checkEvent(evt) {
  var actions = [];
  for(var rn in listRules) {
    //TODO this needs to get depth safe, not only data but eventually also
    // on one level above (eventid and other meta)
    if(listRules[rn].event === evt.event && validConditions(evt.data, listRules[rn])) {
      log.print('EN', 'Rule "' + rn + '" fired');
      actions = actions.concat(listRules[rn].actions);
    }
  }
  return actions;
}

/**
 * Checks whether all conditions of the rule are met by the event.
 * @param {Object} evt the event to check
 * @param {Object} rule the rule with its conditions
 */
function validConditions(evt, rule) {
  for(var property in rule.condition){
    if(!evt[property] || evt[property] != rule.condition[property]) return false;
  }
  return true;
}

/**
 * Invoke an action according to its type.
 * @param {Object} evt The event that invoked the action
 * @param {Object} action The action to be invoked
 */
function invokeAction(evt, action) {
  var actionargs = {},
      arrModule = action.module.split('->');
  if(arrModule.length < 2) {
    log.error('EN', 'Invalid rule detected!');
    return;
  }
  var srvc = listActionModules[arrModule[0]];
  if(srvc && srvc[arrModule[1]]) {
    //FIXME preprocessing not only on data
    preprocessActionArguments(evt.data, action.arguments, actionargs);
    try {
      if(srvc[arrModule[1]]) srvc[arrModule[1]](actionargs);
    } catch(err) {
      log.error('EN', 'during action execution: ' + err);
    }
  }
  else log.print('EN', 'No api interface found for: ' + action.module);
}

/**
 * Action properties may contain event properties which need to be resolved beforehand.
 * @param {Object} evt The event whose property values can be used in the rules action
 * @param {Object} act The rules action arguments
 * @param {Object} res The object to be used to enter the new properties
 */
function preprocessActionArguments(evt, act, res) {
  for(var prop in act) {
    /*
     * If the property is an object itself we go into recursion
     */
    if(typeof act[prop] === 'object') {
      res[prop] = {};
      preprocessActionArguments(evt, act[prop], res[prop]);
    }
    else {
      var txt = act[prop];
      var arr = txt.match(regex);
      /*
       * If rules action property holds event properties we resolve them and
       * replace the original action property
       */
      // console.log(evt);
      if(arr) {
        for(var i = 0; i < arr.length; i++) {
          /*
           * The first three characters are '$X.', followed by the property
           */
          var actionProp = arr[i].substring(3).toLowerCase();
          // console.log(actionProp);
          for(var eprop in evt) {
            // our rules language doesn't care about upper or lower case
            if(eprop.toLowerCase() === actionProp) {
              txt = txt.replace(arr[i], evt[eprop]);
            }
          }
          txt = txt.replace(arr[i], '[property not available]');
        }
      }
      res[prop] = txt;
    }
  }
}

exports.loadEventModule = function(args, answHandler) {
  if(args && args.name) {
  	answHandler.answerSuccess('Loading event module ' + args.name + '...');
  	poller.send('cmd|loadevent|'+args.name);
  } else if(args) answHandler.answerError(args.name + ' not found');
};

exports.loadEventModules = function(args, answHandler) {
	answHandler.answerSuccess('Loading event moules...');
  poller.send('cmd|loadevents');
};

exports.shutDown = function() {
  log.print('EN', 'Shutting down Poller and DB Link');
  isRunning = false;
  if(poller) poller.send('cmd|shutdown');
  if(db) db.shutDown();
};