functional test

simulator.js

// Elphi System Simulator in Node.js

// TODO: check failure case
// 0) request times
// 0) server hangs
// 0) HTTP request timeouts where we never get a response
// 1) no api
// 2) no proxy on
// 3) plugs unresponsive

var async = require("async");
var net = require("net");
var http = require('http');
var numeric_arrays = require('numeric_arrays');

// CL Option Parser
var argv = require('optimist')
  .default('n',1)
  .default('m',1)
  .default('l',1)
  .default('host','localhost')
  .default('proxy_tcp_port',5280)
  .default('proxy_http_port',9001)
  .default('api_http_port',9292)
  .argv;

var tcp_timeout = 12000; // time to wait for keep alives on http connection
var tcp_conns = {};
var sid = 'TimerDaemon';
var rids = []; // all request ids
var req_start = {}; // rid->time
var req_end = {}; // rid->time
var req_delay = {}; // rid->time


function generateRID(string_length) {
  var chars = "0123456789abcdef";
  var randomstring = '';
  for (var i=0; i<string_length; i++) {
    var rnum = Math.floor(Math.random() * chars.length);
    randomstring += chars.substring(rnum,rnum+1);
  }
  return randomstring
}

var openConnections = function(callback){
  // We open up N connections and use callbacks to check if they are active.
  // Algorithm:
  //    0) Create N PlugIds
  //    1) Open up N TCP connections
  //    2) For each TCP connection, add it to the tcp_conn hash once it
  //    has received a keep alive, and call verifyKeepAlives()
  //    3) verifyKeepAlives only calls the original callback which
  //    successfully finishes openConnections() once all plugs get keep
  //    alive messages
  //    4) A timer calls an errback if we don't get a successful finish
  //    within a certain time limit
  
  console.log('Opening n='+argv.n+' connections ')
  
  var plug_ids = [];
  for(var i = 0; i < argv.n; i++) { plug_ids.push("sim_"+i); }
  
  function openTCPConnection(i) {
    if( i < plug_ids.length ) {
      //console.log("plug: "+plug_ids[i]);
      var conn = net.createConnection(argv.proxy_tcp_port, argv.host);
      conn.plug_id = plug_ids[i];
      conn.responsive = true;
      conn.state = '0';
      conn.on('connect', function() {
        var add_msg = "{\"m\":\"add\",\"state\":\""+this.state+"\",\"id\":\""+ this.plug_id+"\",\"sid\":\""+sid+"\",\"name\":\""+this.plug_id+"\",\"loc\":\"sim_loc\",\"timezone\":\"-7\",\"ver\":\"1.0\"}"
        conn.write(add_msg);
      })
      .on('end', function() {
        this.connected = false;
      })
      .on('data', function(data){
        console.log("plug="+this.plug_id+" received: "+data)
        if(this.responsive){
          try {
            // HANDLE DIFFERENT PLUG MESSAGE TYPES
            result = JSON.parse(data);
          } catch (err) {
              console.log("Unable to parse JSON("+data+"):", err)
          }

          if('m' in result && result['m']=='keep_alive'){
            conn.write("{\"m\":\"keep_alive\",\"success\":\"true\"}");
            if(this.connected){
              tcp_conns[this.plug_id] = conn; // this is what the verification step checks
            }
          }
          else if('m' in result && result['m']=='set'){
            // TODO: use rid to log requests
            if('state' in result && 'rid' in result) {
              // We introduce delay into the response!
              
              var rid = result['rid'];
              setTimeout( function() {
                var old_state = this.state
                this.state = result['state'];
                var rid = result['rid'];
                //console.log("plug="+this.plug_id+" changed state: "+old_state+"->"+this.state);
                conn.write("{\"m\":\"set\",\"state\":\""+this.state+"\",\"success\":\"true\",\"rid\":\""+rid+"\"}");
              }, req_delay[rid] ); // a random wait between 0 and 0.5
            }
          }
          else if('success' in result && result['success']=='true'){ // on conn start
            this.connected = true;
            console.log("plug="+this.plug_id+" connected to server")
          }
        }
      });
      openTCPConnection(i+1);
    }
  }
  openTCPConnection(0);
  
  setTimeout( function() {
    console.log("Verifying Keep Alives")
    console.log("plug_ids["+plug_ids.length+"]: "+Object.keys(plug_ids));
    console.log("tcp_conns["+Object.keys(tcp_conns).length+"]: "+Object.keys(tcp_conns));
    
    if(plug_ids.length == Object.keys(tcp_conns).length) {
      console.log("All keep alives received");
      callback(null,'1');
    } else {
      console.log("Error, unable to establish all connections");
      process.exit(1);
    }
  }, tcp_timeout);
}


var sendHTTPRequests = function(callback){
  console.log('Send HTTP requests m='+argv.m);
  
  // Statistics we collect
  var success = 0;
    var failure = 0;
  
  var num_loops = argv.m;
  var count = 0;
  async.whilst(
    function () { return count < num_loops; },
    function (callback) {
      count++;

      for (var plugId in tcp_conns) {
        rid = generateRID(12);
        rids.push(rid);
        state = '255';
        sid = '0xac99d8929a89e6c01b56378336dec3a8L';
        
        var options = {
          host: argv.host,
          port: argv.proxy_http_port,
          path: "/?rid="+rid+"&id="+plugId+"&json={%22m%22:%22set%22,%22state%22:%22"+state+"%22,%22sid%22:%22"+sid+"%22,%22rid%22:%22"+rid+"%22,%22id%22:%22"+plugId+"%22}"
        };

        // get req start time
        req_start[rid] = Date.now();
        req_delay[rid] = Math.floor(Math.random()*1000); // delay of 0-1000 ms
        
        var httpCallback = function(response) {
          var str = '';
          
          response.on('data', function (chunk) { str += chunk; });

          response.on('close', function (err) { // http connection was closed before 'end' received
            console.log("CLOSE EVENT rid="+rid+" err="+err);
            req_end[rid] = Date.now();
            failure++;
          });

          response.on('end', function () {
            console.log("END EVENT rid="+rid);
            req_end[rid] = Date.now();
            
            // Verify response
            result = JSON.parse(str);
            if('response' in result){
              result = result['response'];
              if('success' in result && result['success']=='true'){
                success++;
              }
              else if('success' in result && result['success']=='false'){
                failure++;
              }
            }
            else {
              failure++;
            }
        
          });
        }

        var req = http.request(options, httpCallback);

        req.on('error', function(e) {
          console.log('ERROR EVENT: ' + e.message+", rid="+rid);
        });

        req.end();
      }

      setTimeout(callback, argv.l*1000);
    },
    function (err) {
      console.log("End of HTTP Requests");
      console.log("\n\n");
      
      // Report...
      
      // Do a synchronous foreach and calculate request times
      var rtime = [];
      console.log("Request Statistics")
      console.log("---------------------");
      async.forEach(rids, function (rid, callback){
        var start = req_start[rid];
        var delay = req_delay[rid];
        var end = req_end[rid];
        var request_time = end-start;

        var request_time_no_delay = request_time - delay;
        rtime.push(request_time_no_delay);
        
        console.log("rid="+rid+", start="+start+", end="+end+", delay="+delay+", rtime="+request_time_no_delay);
        
          callback(); // tell async that the iterator has completed
      }, function(err) {
          //process.stdout.write("\n");
      });
      
      numeric_arrays.extend(rtime);
      var mean_rtime = rtime.mean();
      var mean_stddev = rtime.stddev();
      
      console.log("---------------------");
      console.log("rtime avg:    "+mean_rtime);
      console.log("rtime stddev: "+mean_stddev);
      console.log("success:      "+success);
      console.log("failure:      "+failure);
      console.log("---------------------");
      process.exit();
    }
  );
}

// TODO: experiment with waterfall
async.series([
    openConnections
  ,sendHTTPRequests
  //,generateReport
]);