Merge pull request #10 from flamebase/develop

Develop

Author: efraespada

README.md

@@ -1,4 +1,4 @@
-[ ![flamebase/flamebase-database-server-cluster](https://d25lcipzij17d.cloudfront.net/badge.svg?id=js&type=6&v=1.0.5&x2=0)](https://www.npmjs.com/package/flamebase-database-server-cluster)
+[ ![flamebase/flamebase-database-server-cluster](https://d25lcipzij17d.cloudfront.net/badge.svg?id=js&type=6&v=1.2.0&x2=0)](https://www.npmjs.com/package/flamebase-database-server-cluster)
 
 # :fire: flamebase-database-server-cluster
 Server cluster to hold a realtime JSON database.
@@ -22,7 +22,9 @@ Create a server cluster to hold all realtime changes.
 ```javascript
 var FlamebaseDatabaseCluster = require('flamebase-database-server-cluster');
 
-var FSC = new FlamebaseDatabaseCluster("draco", 1507);
+var serverKey = 'AIzafvwsdfmvsdfv-fdvwsdfgweFGfdAsfwefgw-z1JI';
+
+var FSC = new FlamebaseDatabaseCluster("myDatabase", 1507, serverKey);
 FSC.initCluster({
     start: function () {
         console.log("start!!")
@@ -52,7 +54,7 @@ Client options to connect with server cluster.
 
 - [Android](https://github.com/flamebase/flamebase-database-android)
 ```groovy
-compile 'com.flamebase:database:1.1.3'
+compile 'com.flamebase:database:1.2.0'
 ```
 
 - [Node](https://github.com/flamebase/flamebase-database-node)

index.js

@@ -4,7 +4,7 @@ var log4js  = require('log4js');
 const TAG   = "Flamebase Database";
 var logger  = log4js.getLogger(TAG);
 
-function FlamebaseDatabaseCluster(database, port) {
+function FlamebaseDatabaseCluster(database, port, APIKey) {
 
     this.initCluster = function (callback) {
         var child = forever.start('./server.js', {
@@ -19,7 +19,7 @@ function FlamebaseDatabaseCluster(database, port) {
 
             sourceDir: __dirname,
 
-            args:    ['DATABASE_NAME=' + database, 'DATABASE_PORT=' + port],
+            args:    ['DATABASE_NAME=' + database, 'DATABASE_PORT=' + port, 'API_KEY=' + APIKey],
 
             watch: false,
             watchIgnoreDotFiles: null,

model/path.js

@@ -4,7 +4,7 @@ var log4js =                require('log4js');
 var TAG =                   "PATH CLUSTER";
 var logger =                log4js.getLogger(TAG);
 
-function Path(databasePath, database, path, pid) {
+function Path(APIKey, databasePath, database, path, pid) {
 
     // object reference
     var object = this;
@@ -16,7 +16,7 @@ function Path(databasePath, database, path, pid) {
 
     this.FD.syncFromDatabase();
 
-    logger.debug("created reference listener for " + this.database + " - " + this.path + ": " + JSON.stringify(this.FD.ref));
+    // logger.debug("created reference listener for " + this.database + " - " + this.path + ": " + JSON.stringify(this.FD.ref));
 
 
     var config = {};
@@ -25,7 +25,7 @@ function Path(databasePath, database, path, pid) {
      * server API key for firebase cloud messaging
      */
     config.APIKey = function() {
-        return "AIzaSyDhzzf2kkKuknAdiB_ak-KJ0a3XgaBDO4A";
+        return APIKey;
     };
 
     /**
@@ -72,8 +72,8 @@ function Path(databasePath, database, path, pid) {
     this.FD.setSyncConfig(config);
     this.FD.debug(true);
 
-    this.start = function () {
-
+    this.sendUpdateFor = function (before, device) {
+        this.FD.sendDifferencesForClient(before, device);
     }
 
 }

package.json

@@ -1,6 +1,6 @@
 {
   "name": "flamebase-database-server-cluster",
-  "version": "1.0.5",
+  "version": "1.2.0",
   "description": "Server cluster to hold a realtime JSON database",
   "main": "index.js",
   "scripts": {
@@ -13,11 +13,12 @@
   "dependencies": {
     "body-parser": "latest",
     "express": "latest",
-    "flamebase-database-node": "^1.1.1",
+    "flamebase-database-node": "^1.2.0",
     "forever-monitor": "^1.7.1",
     "json-align": "^0.1.0",
     "log4js": "latest",
-    "rus-diff": "^1.1.0"
+    "rus-diff": "^1.1.0",
+    "sha1": "^1.1.1"
   },
   "keywords": [
     "flamebase",

server.js

@@ -8,6 +8,7 @@ var FlamebaseDatabase =     require("flamebase-database-node");
 var Path =                  require("./model/path.js");
 var apply =                 require('rus-diff').apply;
 var clone =                 require('rus-diff').clone;
+var sha1 =                  require('sha1');
 
 JSON.stringifyAligned = require('json-align');
 
@@ -17,16 +18,21 @@ String.prototype.replaceAll = function(search, replacement) {
 };
 
 var expectedDBNEnvVar = "DATABASE_NAME";
-var expectedPORTNEnvVar = "DATABASE_PORT";
+var expectedPORTEnvVar = "DATABASE_PORT";
+var expectedAPIKeyEnvVar = "API_KEY";
 var dbMaster = null;
 var port = null;
+var APIKey = null;
 
 process.argv.forEach(function (val, index, array) {
     if (val.indexOf(expectedDBNEnvVar) > -1) {
         dbMaster = val.replaceAll(expectedDBNEnvVar + "=", "");
     }
-    if (val.indexOf(expectedPORTNEnvVar) > -1) {
-        port = val.replaceAll(expectedPORTNEnvVar + "=", "");
+    if (val.indexOf(expectedPORTEnvVar) > -1) {
+        port = val.replaceAll(expectedPORTEnvVar + "=", "");
+    }
+    if (val.indexOf(expectedAPIKeyEnvVar) > -1) {
+        APIKey = val.replaceAll(expectedAPIKeyEnvVar + "=", "");
     }
 });
 
@@ -35,7 +41,6 @@ var TAG =                   "SERVER CLUSTER";
 var logger =                log4js.getLogger(TAG);
 
 var dbPaths = "paths";
-var paths = new FlamebaseDatabase(dbPaths, "/");
 
 if (cluster.isMaster) {
 
@@ -58,7 +63,8 @@ if (cluster.isMaster) {
 
     var action = {
         response: function (connection, data, error, pId) {
-            var result = {status: (data === null || error !== null ? "KO" : "OK"), data: data, error: error};
+            var result = {status: (data === null || error !== null ? "KO" : "OK"),
+                data: (data === null ? {} : data), error: error};
             logger.info("worker: " + pId);
             logger.info("response: " + JSON.stringify(result));
             connection.response.contentType('application/json');
@@ -68,7 +74,12 @@ if (cluster.isMaster) {
             this.addGreatListener(connection, pId);
         },
         addGreatListener: function (connection, pId) {
+            var paths = new FlamebaseDatabase(dbPaths, "/");
+            paths.syncFromDatabase();
+            logger.debug(JSON.stringifyAligned(paths.ref));
             paths.syncFromDatabase();
+            logger.debug("getting");
+            logger.debug(JSON.stringifyAligned(paths.ref));
 
             if (paths.ref === undefined) {
                 paths.ref = {}
@@ -92,27 +103,86 @@ if (cluster.isMaster) {
                     paths.ref[key].tokens[connection.token].os = connection.os;
                 }
 
+                var keys = Object.keys(paths.ref[key].tokens);
+                var lastChangeTime = 0;
+                var lastToken = null;
+
+                for (var i = keys.length - 1; i >= 0; i--) {
+                    var time = paths.ref[key].tokens[keys[i]].time;
+                    if (lastChangeTime < time) {
+                        lastChangeTime = time;
+                        lastToken = keys[i];
+                    }
+                }
+
                 paths.ref[key].tokens[connection.token].time = new Date().getTime();
 
                 paths.syncToDatabase();
 
+                var equals = this.verifyLenght(connection, pId);
+
                 var object = this.getReference(connection, pId);
                 object.FD.syncFromDatabase();
                 var len = 0;
+
                 if (typeof object !== "string") {
                     len = JSON.stringify(object.FD.ref).length;
+                } else {
+                    this.response(connection, null, object, pId);
+                    return;
                 }
+
                 var data = {};
                 data.len = len;
-                data.info = "listener_added";
+
+                if (lastToken === connection.token && equals) {
+                    data.info = "listener_up_to_date";
+                } else {
+                    data.info = "listener_ready_for_refresh_client";
+                }
 
                 this.response(connection, data, null, pId);
             } else {
                 this.response(connection, null, "path_contains_dots", pId);
             }
 
         },
+        verifyLenght: function (connection, pId) {
+            var paths = new FlamebaseDatabase(dbPaths, "/");
+            paths.syncFromDatabase();
+            var object = this.getReference(connection, pId);
+            // var len = JSON.stringify(object.FD.ref).length;
+            logger.debug(sha1(JSON.stringify(object.FD.ref)).toUpperCase());
+            logger.debug(connection.sha1);
+
+            var hash = sha1(JSON.stringify(object.FD.ref)).toUpperCase();
+            return hash === connection.sha1;
+        },
+        getUpdatesFrom: function (connection, pId) {
+            var paths = new FlamebaseDatabase(dbPaths, "/");
+            paths.syncFromDatabase();
+            logger.debug("getting");
+            logger.debug(JSON.stringifyAligned(paths.ref));
+            var object = this.getReference(connection, pId);
+            if (typeof object === "string") {
+                this.response(connection, null, object, pId);
+            } else {
+                var device = {
+                    token: connection.token,
+                    os: connection.os
+                }
+                object.sendUpdateFor(connection.content, device)
+                var data = {};
+                data.info = "updates_sent";
+                data.len = JSON.stringify(object.FD.ref).length;
+                this.response(connection, data, null, pId);
+            }
+        },
         updateData: function (connection, pId) {
+            var paths = new FlamebaseDatabase(dbPaths, "/");
+            paths.syncFromDatabase();
+            logger.debug("getting");
+            logger.debug(JSON.stringifyAligned(paths.ref));
             var object = this.getReference(connection, pId);
             if (typeof object === "string") {
                 this.response(connection, null, object, pId);
@@ -125,40 +195,43 @@ if (cluster.isMaster) {
                     logger.debug(JSON.stringify(differences));
                     apply(object.FD.ref, JSON.parse(differences));
 
-                    if (JSON.stringify(object.FD.ref).length < connection.len) {
-                        logger.error("##########_inconsistency_length");
-                        this.response(connection, null, "inconsistency_length", pId);
-                        return;
+                    if (JSON.stringify(object.FD.ref).length !== connection.len) {
+                        object.FD.syncToDatabase();
+                        this.response(connection, null, "data_updated_with_differences", pId);
+                    } else {
+                        object.FD.syncToDatabase();
+                        this.response(connection, "data_updated", null, pId);
                     }
-
-                    object.FD.syncToDatabase();
-
-                    this.response(connection, "data_updated", null, pId);
                 } else {
                     this.response(connection, "no_diff_updated", null, pId);
                 }
             }
         },
         getReference: function (connection, pId) {
+            var paths = new FlamebaseDatabase(dbPaths, "/");
+            paths.syncFromDatabase();
+            var error = null;
             if (connection.path !== undefined) {
                 if (connection.path.indexOf("\.") === -1) {
                     if (connection.path.indexOf("/") === 0) {
                         var key = connection.path.replaceAll("/", "\.");
                         key = key.substr(1, key.length - 1);
                         if (paths.ref[key] !== undefined) {
-                            return new Path(paths.ref[key], dbMaster, connection.path, pId);
+                            return new Path(APIKey,paths.ref[key], dbMaster, connection.path, pId);
                         } else {
-                            return "holder_not_found";
+                            error = "holder_not_found";
                         }
                     } else {
-                        return "path_not_starts_with_slash";
+                        error = "path_not_starts_with_slash";
                     }
                 } else {
-                    return "path_contains_dots";
+                    error = "path_contains_dots";
                 }
             } else {
-                return "json_path_not_found";
+                error = "json_path_not_found";
             }
+            logger.error(error);
+            return error;
         },
         parseRequest: function (req, res, worker) {
             var response = res;
@@ -193,6 +266,11 @@ if (cluster.isMaster) {
                             logger.debug("path: " + connection[key]);
                             break;
 
+                        case "sha1":
+                            connection[key] = message[key];
+                            logger.debug("sha1: " + connection[key]);
+                            break;
+
                         case "token":
                             connection[key] = message[key];
                             logger.debug("token: " + connection[key]);
@@ -203,6 +281,11 @@ if (cluster.isMaster) {
                             logger.debug("differences: " + connection[key]);
                             break;
 
+                        case "content":
+                            connection[key] = message[key];
+                            logger.debug("content: " + connection[key]);
+                            break;
+
                         case "len":
                             connection[key] = message[key];
                             logger.debug("len: " + connection[key]);
@@ -213,6 +296,11 @@ if (cluster.isMaster) {
                             logger.debug("os: " + connection[key]);
                             break;
 
+                        case "clean":
+                            connection[key] = message[key];
+                            logger.debug("clean: " + connection[key]);
+                            break;
+
                         default:
 
                             //
@@ -253,6 +341,15 @@ if (cluster.isMaster) {
                         }
                         break;
 
+                    case "get_updates":
+                        try {
+                            this.getUpdatesFrom(connection, worker);
+                        } catch (e) {
+                            logger.error("there was an error parsing request from getUpdatesFrom: " + e.toString());
+                            this.response(connection, null, "cluster_" + worker + "_error_getting_updates", worker);
+                        }
+                        break;
+
                     default:
                         //
                         break;
@@ -287,7 +384,6 @@ if (cluster.isMaster) {
         });
 
     app.listen(port, function () {
-        paths.syncFromDatabase();
         logger.info("server cluster started on port " + port + " on " + cluster.worker.id + " worker");
     });
 }