Building patch instruction

Author: Josema

src/core/protocol/registerObjectToNode.js

@@ -17,10 +17,11 @@ dop.core.registerObjectToNode = function(node, object) {
         object_data.node[node.token] = {
             subscriber: 0, // 0 or 1 || false true 
             owner: 0, // object_id_owner || 0 === false
-            subscriber_version: 0, 
-            owner_version: 0,
-            pending_version: 0,
-            pending: {}
+            version: 0, // incremental integer for new patches
+            received_version: 0, // last patch version received correctly
+            received: {},
+            applied_version:0, // last patch version applied correctly
+            applied:{}
         };
     }
 

src/protocol/patch.js

@@ -1,7 +1,7 @@
 
 dop.protocol.patch = function(node, object_id, patch) {
     var object_node = dop.data.object[object_id].node[node.token],
-        version = ++object_node.subscriber_version,
+        version = ++object_node.version,
         request = dop.core.createRequest(
             node,
             dop.protocol.instructions.patch,
@@ -10,7 +10,7 @@ dop.protocol.patch = function(node, object_id, patch) {
             patch
         );
 
-    object_node.pending[version] = patch;
+    object_node.received[version] = {received:false, patch:dop.util.merge({}, patch)}; // Making a copy because this object is exposed to the api users and can be mutated
     dop.core.storeSendMessages(node, request, dop.encodeFunction);
     return request.promise;
 };
@@ -22,19 +22,29 @@ dop.protocol.onpatch = function(node, request_id, request) {
         patch = request[3],
         response = dop.core.createResponse(request_id),
         object_data = dop.data.object[object_id],
-        object_node;
-    
+        object_node,
+        collector;
+    console.log( version );
     if (isObject(object_data) && isObject(object_data.node[node.token]) && object_data.node[node.token].owner===object_id_owner) {
         object_node = object_data.node[node.token];
-        // We apply patch
-        if (object_node.owner_version+1 === version) {
-            object_node.owner_version += 1;
+        // Storing patch
+        if (object_node.applied_version < version && object_node.applied[version]===undefined) {
+            console.log(version, 'entra')
+            // Storing patch
+            object_node.applied[version] = patch;
+            // Applying
             collector = dop.collectFirst();
-            dop.core.setActionFunction(object_data.object, patch);
+            while (object_node.applied[object_node.applied_version+1]) {
+                object_node.applied_version += 1;
+                console.log(object_node.applied_version, object_node.applied[object_node.applied_version]  );
+                dop.core.setActionFunction(object_data.object, object_node.applied[object_node.applied_version]);
+                delete object_node.applied[object_node.applied_version];
+            }
             collector.emitAndDestroy();
-            response.push(0);
         }
-        //
+        console.log( '--' );
+        // Confirm received
+        response.push(0);
     }
     else
         response.push(dop.core.error.reject_remote.OBJECT_NOT_FOUND);
@@ -44,14 +54,21 @@ dop.protocol.onpatch = function(node, request_id, request) {
 
 dop.protocol._onpatch = function(node, request_id, request, response) {
     var rejection = response[0],
-        // object_id = request[2],
+        object_id = request[2],
+        object_node = dop.data.object[object_id].node[node.token],
+        version = request[3],
         promise = request.promise;
 
 
     if (rejection !== undefined) {
         if (rejection === 0) {
-            // var object_node = dop.data.object[object_id].node[node.token];
-            promise.resolve(response[1]);
+            // Correct version applyed
+            if (object_node.pending_version+1 === version) {
+            console.log( object_node.pending_version, version );
+                object_node.pending_version += 1;
+                delete object_node.pending[object_node.pending_version];
+                promise.resolve(response[1]);
+            }
         }
         else
             promise.reject(dop.core.getRejectError(rejection));

test/protocol/patchs.js

@@ -7,6 +7,7 @@ var transportListen = require('dop-transports').listen[transportName];
 var transportConnect = require('dop-transports').connect[transportName];
 
 dop.env = 'SERVER';
+dop.data.object_inc = 25;
 dopClient1.env = 'CLIENT1';
 dopClient2.env = 'CLIENT2';
 
@@ -36,15 +37,17 @@ dop.onsubscribe(function(){
 // test('BROADCASTING TO CLIENTS', function(t) {
 
 client1.subscribe().into({subobject:{broadcast:function(a,b){return a+b}}})
+// .then(function(obj) {
+//     return client2.subscribe().into({subobject:{broadcast:function(a,b){return a*b}}})
+// })
 .then(function(obj) {
-    return client2.subscribe().into({subobject:{broadcast:function(a,b){return a*b}}})
-})
-.then(function(obj) {
+            console.log(dopClient1.data.object[1].object)
 
-    // var collector = dop.collect();
     objServer.number = 25;
     objServer.arr = [1,2,3];
-    // collector.emitAndDestroy();
+    objServer.tres = 3;
+    objServer.cuatro = 4444;
+    objServer.cinco = 'elcinco';
 
 //     var promises = objServer.subobject.broadcast(2,5);
 //     t.equal(Array.isArray(promises), true, 'Promises is array');
@@ -63,4 +66,31 @@ client1.subscribe().into({subobject:{broadcast:function(a,b){return a+b}}})
 // })
 
 
-// // More test todo
+// // More test todo
+dopClient1.protocol.onpatchOri = dopClient1.protocol.onpatch; 
+dopClient1.protocol.onpatch = function(node, request_id, request) {
+    var version = request[2];
+    if (request[2] === 2) {
+        setTimeout(function() {
+            dopClient1.protocol.onpatchOri(node, request_id, request);
+        }, 100);
+        setTimeout(function() {
+            dopClient1.protocol.onpatchOri(node, request_id, request);
+            console.log(dopClient1.data.object[1].object)
+        }, 300);
+    }
+    else if (request[2] === 4) {
+        setTimeout(function() {
+            dopClient1.protocol.onpatchOri(node, request_id, request);
+        }, 150);
+        setTimeout(function() {
+            dopClient1.protocol.onpatchOri(node, request_id, request);
+        }, 200);
+    }
+    else if (request[2] === 3) {
+        dopClient1.protocol.onpatchOri(node, request_id, request);
+        dopClient1.protocol.onpatchOri(node, request_id, request);
+    }
+    else
+        dopClient1.protocol.onpatchOri(node, request_id, request);
+}