Major refactor for frontend. Switching to crossfilter and dc js

bin/databases-config.js

@@ -4,7 +4,8 @@ module.exports = {
         "mongo-port": 27017,
         "mongo-database": "events",
         "mongo-username": null,
-        "mongo-password": null
+        "mongo-password": null,
+        "collectionSize": 1024 * 1024 * 1024
     },
     aggregations: {
         "mongo-host": "192.168.56.101",

bin/generator.js

@@ -38,11 +38,10 @@ var options = require("./generator-config.js"),
             }
             return a;
         },
-        sendPacket: function () {
+        sendPacket: function (type) {
             var country = options.data.country[randomTools.getRandomInt(0, options.data.country.length - 1)],
                 gender = options.data.gender[randomTools.getRandomInt(0, options.data.gender.length - 1)],
                 device = options.data.device[randomTools.getRandomInt(0, options.data.device.length - 1)],
-                type = options.data.type[randomTools.getRandomInt(0, options.data.type.length - 1)],
                 packet = JSON.stringify({
                     type: type,
                     data: {
@@ -54,19 +53,20 @@ var options = require("./generator-config.js"),
             //            console.log(packet);
             ws.send(packet);
         },
-        sendPackets: function (n) {
+        sendPackets: function (n, type) {
             var i;
             console.log((new Date()).getTime() + ": Sending packets: " + n);
             for (i = 0; i < n; i++) {
-                this.sendPacket();
+                this.sendPacket(type);
             }
         },
 
         sendForPacketsForInterval: function (arrayOfPacketsInInterval) {
             var i,
                 doSetTimeout = function (i) {
                     setTimeout(function () {
-                            packetGenerator.sendPackets(arrayOfPacketsInInterval[i]);
+                            var type = options.data.type[randomTools.getRandomInt(0, options.data.type.length - 1)];
+                            packetGenerator.sendPackets(arrayOfPacketsInInterval[i], type);
                         },
                         i * options.sampleRateInMS);
                 };

bower.json

@@ -14,7 +14,7 @@
         "jquery": "2.1.0",
         "bootstrap": "3.1.1",
         "d3": "3.3.*",
-        "crossfilter": "1.2.0",
+        "crossfilter": "1.3.*",
         "nvd3": "1.1.*",
         "dcjs": "1.6.0"
     },

examples/event-stream/event-get.html

@@ -30,8 +30,7 @@ <h1>Streaming Events</h1>
             console.log("connected!");
             socket.send(JSON.stringify({
                 type: type,
-                stop: new Date(),
-                start: new Date().getTime()-5*60*1000
+                start: new Date().getTime()-60*1000
             }));
         };
 

lib/analytics/disperser.js

@@ -12,10 +12,12 @@ exports.register = function (dbs, endpoints, options) {
     var eventsDB = dbs.events,
         aggregationsDB = dbs.aggregations,
         eventGetter = require("./event.js").getterInit(eventsDB),
+        eventCounter = require("./event.js").counterInit(eventsDB),
         aggregationGetter = require("./aggregation.js").getterInit(aggregationsDB);
 
     endpoints.ws.push(
         endpoint("/1.0/event/get", eventGetter),
+        endpoint("/1.0/event/getCount", eventCounter),
         endpoint("/1.0/aggregation/get", aggregationGetter)
     );
 
@@ -30,7 +32,7 @@ exports.register = function (dbs, endpoints, options) {
             request.limit = LIMIT_MAX;
         }
 
-        function documentHandler(dataElem) {
+        function messageSenderHttp(dataElem) {
             if (dataElem === null) {
                 response.end(JSON.stringify(data.reverse()));
             }
@@ -39,7 +41,7 @@ exports.register = function (dbs, endpoints, options) {
             }
         }
 
-        if (eventGetter(request, documentHandler) < 0) {
+        if (eventGetter(request, messageSenderHttp) < 0) {
             response.writeHead(400, headers);
             response.end(JSON.stringify(data[0]));
         } else {
@@ -59,7 +61,7 @@ exports.register = function (dbs, endpoints, options) {
             request.limit = LIMIT_MAX;
         }
 
-        function documentHandler(dataElem) {
+        function messageSenderHttp(dataElem) {
             if (dataElem === null) {
                 response.end(JSON.stringify(data.reverse()));
             }
@@ -68,7 +70,36 @@ exports.register = function (dbs, endpoints, options) {
             }
         }
 
-        if (aggregationGetter(request, documentHandler) < 0) {
+        if (aggregationGetter(request, messageSenderHttp) < 0) {
+            response.writeHead(400, headers);
+            response.end(JSON.stringify(data[0]));
+        } else {
+            response.writeHead(200, headers);
+        }
+
+    }
+
+    function eventCount(request, response) {
+        request = url.parse(request.url, true).query;
+        var data = [];
+
+        if (!request.hasOwnProperty('start')) {
+            request.start = 0;
+        }
+        if ((request.hasOwnProperty('limit') && (request.limit >= LIMIT_MAX))) {
+            request.limit = LIMIT_MAX;
+        }
+
+        function messageSenderHttp(dataElem) {
+            if (dataElem === null) {
+                response.end(JSON.stringify(data.reverse()));
+            }
+            else {
+                data.push(dataElem);
+            }
+        }
+
+        if (eventCounter(request, messageSenderHttp) < 0) {
             response.writeHead(400, headers);
             response.end(JSON.stringify(data[0]));
         } else {
@@ -79,6 +110,7 @@ exports.register = function (dbs, endpoints, options) {
 
     endpoints.http.push(
         endpoint("GET", "/1.0/event/get", eventGet),
+        endpoint("GET", "/1.0/event/getCount", eventCount),
         endpoint("GET", "/1.0/aggregation/get", aggregationGet)
     );
 

lib/analytics/event.js

@@ -2,18 +2,20 @@
 var mongodb = require("mongodb"),
     type_re = /^[a-z][a-zA-Z0-9_]+$/,
     genericGetter = require("./genericGetter.js"),
+    genericCounter = require("./genericCounter.js"),
     myutils = require("./myutils.js"),
-    util = require("util");
+    util = require("util"),
+    mongoConfigs = require("../../bin/databases-config");
 
 exports.putterInit = function (db, options) {
     var eventsCollectionCreated = 0,
         eventsToSave = [],
         events = [],
-        collectionSize = options.collectionSize,
+        collectionSize = mongoConfigs.events.collectionSize,
         bufferSize = 10000,
         bufferTimestamp = (new Date()).getTime();
 
-    if (myutils.isInt(collectionSize)) {
+    if (!myutils.isInt(collectionSize)) {
         throw "Invalid collection size: " + collectionSize;
     }
     function handle(error) {
@@ -61,7 +63,8 @@ exports.putterInit = function (db, options) {
             request = [request];
         }
         for (i = 0; i < request.length; i++) {
-            events[i] = {t: time, d: request[i].data, type: request[i].type};
+            //diminishing time granularity, as it is not needed and saving just the second allows us to aggregate on time
+            events[i] = {t: time - time % 1000, d: request[i].data, type: request[i].type};
         }
         // If eventsCollectionCreated, save immediately.
         if (eventsCollectionCreated === 1) {
@@ -93,7 +96,7 @@ exports.putterInit = function (db, options) {
                 return save();
             }
 
-            // Events are indexed by time which is _id, which is natural order.
+            // Events are indexed by _id, which is natural order. (which is time)
             db.createCollection("events", {capped: true, autoIndexId: true, size: collectionSize}, function (error, result) {
                 handle(error);
                 eventsCollectionCreated = 1;
@@ -108,3 +111,6 @@ exports.putterInit = function (db, options) {
 exports.getterInit = function (eventsDB) {
     return genericGetter(eventsDB);
 };
+exports.counterInit = function (eventsDB) {
+    return genericCounter(eventsDB);
+};