fixed y-js/y-webrtc#2

Bugfix 21

bower.json

@@ -1,6 +1,6 @@
 {
   "name": "yjs",
-  "version": "0.5.0",
+  "version": "0.5.2",
   "homepage": "https://github.com/DadaMonad/yjs",
   "authors": [
     "Kevin Jahns <kevin.jahns@rwth-aachen.de>"

build/node/ConnectorAdapter.js

@@ -8,16 +8,6 @@ adaptConnector = function(connector, engine, HB, execution_listener) {
     f = ConnectorClass[name];
     connector[name] = f;
   }
-  connector.setIsBoundToY();
-  send_ = function(o) {
-    if ((o.uid.creator === HB.getUserId()) && (typeof o.uid.op_number !== "string") && (HB.getUserId() !== "_temp")) {
-      return connector.broadcast(o);
-    }
-  };
-  if (connector.invokeSync != null) {
-    HB.setInvokeSyncHandler(connector.invokeSync);
-  }
-  execution_listener.push(send_);
   encode_state_vector = function(v) {
     var results, value;
     results = [];
@@ -61,11 +51,21 @@ adaptConnector = function(connector, engine, HB, execution_listener) {
   if (connector.receive_handlers == null) {
     connector.receive_handlers = [];
   }
-  return connector.receive_handlers.push(function(sender, op) {
+  connector.receive_handlers.push(function(sender, op) {
     if (op.uid.creator !== HB.getUserId()) {
       return engine.applyOp(op);
     }
   });
+  connector.setIsBoundToY();
+  send_ = function(o) {
+    if ((o.uid.creator === HB.getUserId()) && (typeof o.uid.op_number !== "string") && (HB.getUserId() !== "_temp")) {
+      return connector.broadcast(o);
+    }
+  };
+  if (connector.invokeSync != null) {
+    HB.setInvokeSyncHandler(connector.invokeSync);
+  }
+  return execution_listener.push(send_);
 };
 
 module.exports = adaptConnector;

build/node/ConnectorClass.js

@@ -118,7 +118,7 @@ module.exports = {
     }
   },
   whenSynced: function(args) {
-    if (args.constructore === Function) {
+    if (args.constructor === Function) {
       args = [args];
     }
     if (this.is_synced) {
@@ -154,7 +154,7 @@ module.exports = {
       this.send(user, {
         sync_step: "getHB",
         send_again: "true",
-        data: []
+        data: this.getStateVector()
       });
       if (!this.sent_hb_to_all_users) {
         this.sent_hb_to_all_users = true;
@@ -184,7 +184,7 @@ module.exports = {
     this.send(user, {
       sync_step: "getHB",
       send_again: "true",
-      data: []
+      data: this.getStateVector()
     });
     hb = this.getHB([]).hb;
     _hb = [];
@@ -205,22 +205,30 @@ module.exports = {
     });
   },
   setStateSynced: function() {
-    var f, i, len, ref;
+    var args, el, f, i, len, ref;
     if (!this.is_synced) {
       this.is_synced = true;
       if (this.compute_when_synced != null) {
         ref = this.compute_when_synced;
         for (i = 0, len = ref.length; i < len; i++) {
-          f = ref[i];
-          f();
+          el = ref[i];
+          f = el[0];
+          args = el.slice(1);
+          f.apply(args);
         }
         delete this.compute_when_synced;
       }
       return null;
     }
   },
+  whenReceivedStateVector: function(f) {
+    if (this.when_received_state_vector_listeners == null) {
+      this.when_received_state_vector_listeners = [];
+    }
+    return this.when_received_state_vector_listeners.push(f);
+  },
   receiveMessage: function(sender, res) {
-    var _hb, data, f, hb, i, j, len, len1, o, ref, results, sendApplyHB, send_again;
+    var _hb, data, f, hb, i, j, k, len, len1, len2, o, ref, ref1, results, sendApplyHB, send_again;
     if (res.sync_step == null) {
       ref = this.receive_handlers;
       results = [];
@@ -234,6 +242,14 @@ module.exports = {
         return;
       }
       if (res.sync_step === "getHB") {
+        if (this.when_received_state_vector_listeners != null) {
+          ref1 = this.when_received_state_vector_listeners;
+          for (j = 0, len1 = ref1.length; j < len1; j++) {
+            f = ref1[j];
+            f.call(this, res.data);
+          }
+        }
+        delete this.when_received_state_vector_listeners;
         data = this.getHB(res.data);
         hb = data.hb;
         _hb = [];
@@ -250,8 +266,8 @@ module.exports = {
             };
           })(this);
         }
-        for (j = 0, len1 = hb.length; j < len1; j++) {
-          o = hb[j];
+        for (k = 0, len2 = hb.length; k < len2; k++) {
+          o = hb[k];
           _hb.push(o);
           if (_hb.length > 10) {
             sendApplyHB({
@@ -269,10 +285,10 @@ module.exports = {
           send_again = (function(_this) {
             return function(sv) {
               return function() {
-                var k, len2;
+                var l, len3;
                 hb = _this.getHB(sv).hb;
-                for (k = 0, len2 = hb.length; k < len2; k++) {
-                  o = hb[k];
+                for (l = 0, len3 = hb.length; l < len3; l++) {
+                  o = hb[l];
                   _hb.push(o);
                   if (_hb.length > 10) {
                     _this.send(sender, {

build/node/HistoryBuffer.js

@@ -16,30 +16,26 @@ HistoryBuffer = (function() {
     setTimeout(this.emptyGarbage, this.garbageCollectTimeout);
   }
 
-  HistoryBuffer.prototype.resetUserId = function(id) {
-    var o, o_name, own;
-    own = this.buffer[this.user_id];
-    if (own != null) {
-      for (o_name in own) {
-        o = own[o_name];
-        if (o.uid.creator != null) {
-          o.uid.creator = id;
-        }
-        if (o.uid.alt != null) {
-          o.uid.alt.creator = id;
-        }
-      }
-      if (this.buffer[id] != null) {
-        throw new Error("You are re-assigning an old user id - this is not (yet) possible!");
-      }
-      this.buffer[id] = own;
-      delete this.buffer[this.user_id];
+  HistoryBuffer.prototype.setUserId = function(user_id1, state_vector) {
+    var base, buff, counter_diff, name, o, o_name, ref;
+    this.user_id = user_id1;
+    if ((base = this.buffer)[name = this.user_id] == null) {
+      base[name] = [];
     }
-    if (this.operation_counter[this.user_id] != null) {
-      this.operation_counter[id] = this.operation_counter[this.user_id];
-      delete this.operation_counter[this.user_id];
+    buff = this.buffer[this.user_id];
+    counter_diff = state_vector[this.user_id] || 0;
+    if (this.buffer._temp != null) {
+      ref = this.buffer._temp;
+      for (o_name in ref) {
+        o = ref[o_name];
+        o.uid.creator = this.user_id;
+        o.uid.op_number += counter_diff;
+        buff[o.uid.op_number] = o;
+      }
     }
-    return this.user_id = id;
+    this.operation_counter[this.user_id] = (this.operation_counter._temp || 0) + counter_diff;
+    delete this.operation_counter._temp;
+    return delete this.buffer._temp;
   };
 
   HistoryBuffer.prototype.emptyGarbage = function() {
@@ -243,15 +239,13 @@ HistoryBuffer = (function() {
     if ((base = this.operation_counter)[name = o.uid.creator] == null) {
       base[name] = 0;
     }
-    if (o.uid.creator !== this.getUserId()) {
-      if (o.uid.op_number === this.operation_counter[o.uid.creator]) {
-        this.operation_counter[o.uid.creator]++;
-      }
-      while (this.buffer[o.uid.creator][this.operation_counter[o.uid.creator]] != null) {
-        this.operation_counter[o.uid.creator]++;
-      }
-      return void 0;
+    if (o.uid.op_number === this.operation_counter[o.uid.creator]) {
+      this.operation_counter[o.uid.creator]++;
     }
+    while (this.buffer[o.uid.creator][this.operation_counter[o.uid.creator]] != null) {
+      this.operation_counter[o.uid.creator]++;
+    }
+    return void 0;
   };
 
   return HistoryBuffer;

build/node/Operations/Basic.js

@@ -442,7 +442,7 @@ module.exports = function() {
       if (callLater) {
         this.parent.callOperationSpecificDeleteEvents(this, o);
       }
-      if ((this.prev_cl != null) && this.prev_cl.isDeleted()) {
+      if ((this.prev_cl != null) && this.prev_cl.isDeleted() && this.prev_cl.garbage_collected !== true) {
         return this.prev_cl.applyDelete();
       }
     };

build/node/y.js

@@ -10,15 +10,15 @@ adaptConnector = require("./ConnectorAdapter");
 
 createY = function(connector) {
   var HB, ct, engine, model, ops, ops_manager, user_id;
-  user_id = null;
   if (connector.user_id != null) {
     user_id = connector.user_id;
   } else {
     user_id = "_temp";
-    connector.on_user_id_set = function(id) {
-      user_id = id;
-      return HB.resetUserId(id);
-    };
+    connector.when_received_state_vector_listeners = [
+      function(state_vector) {
+        return HB.setUserId(this.user_id, state_vector);
+      }
+    ];
   }
   HB = new HistoryBuffer(user_id);
   ops_manager = structured_ops_uninitialized(HB, this.constructor);

lib/ConnectorAdapter.coffee

@@ -10,18 +10,6 @@ adaptConnector = (connector, engine, HB, execution_listener)->
   for name, f of ConnectorClass
     connector[name] = f
 
-  connector.setIsBoundToY()
-
-  send_ = (o)->
-    if (o.uid.creator is HB.getUserId()) and
-        (typeof o.uid.op_number isnt "string") and # TODO: i don't think that we need this anymore..
-        (HB.getUserId() isnt "_temp")
-      connector.broadcast o
-
-  if connector.invokeSync?
-    HB.setInvokeSyncHandler connector.invokeSync
-
-  execution_listener.push send_
   # For the XMPPConnector: lets send it as an array
   # therefore, we have to restructure it later
   encode_state_vector = (v)->
@@ -57,5 +45,17 @@ adaptConnector = (connector, engine, HB, execution_listener)->
     if op.uid.creator isnt HB.getUserId()
       engine.applyOp op
 
+  connector.setIsBoundToY()
+
+  send_ = (o)->
+    if (o.uid.creator is HB.getUserId()) and
+        (typeof o.uid.op_number isnt "string") and # TODO: i don't think that we need this anymore..
+        (HB.getUserId() isnt "_temp")
+      connector.broadcast o
+
+  if connector.invokeSync?
+    HB.setInvokeSyncHandler connector.invokeSync
+
+  execution_listener.push send_
 
 module.exports = adaptConnector

lib/ConnectorClass.coffee

@@ -102,10 +102,10 @@ module.exports =
 
   #
   # Execute a function _when_ we are connected. If not connected, wait until connected.
-  # @param f {Function} Will be executed on the PeerJs-Connector context.
+  # @param f {Function} Will be executed on the Connector context.
   #
   whenSynced: (args)->
-    if args.constructore is Function
+    if args.constructor is Function
       args = [args]
     if @is_synced
       args[0].apply this, args[1..]
@@ -143,7 +143,7 @@ module.exports =
       @send user,
         sync_step: "getHB"
         send_again: "true"
-        data: [] # @getStateVector()
+        data: @getStateVector()
       if not @sent_hb_to_all_users
         @sent_hb_to_all_users = true
 
@@ -171,7 +171,7 @@ module.exports =
     @send user,
       sync_step: "getHB"
       send_again: "true"
-      data: []
+      data: @getStateVector()
     hb = @getHB([]).hb
     _hb = []
     for o in hb
@@ -192,11 +192,19 @@ module.exports =
     if not @is_synced
       @is_synced = true
       if @compute_when_synced?
-        for f in @compute_when_synced
-          f()
+        for el in @compute_when_synced
+          f = el[0]
+          args = el[1..]
+          f.apply(args)
         delete @compute_when_synced
       null
 
+  # executed when the a state_vector is received. listener will be called only once!
+  whenReceivedStateVector: (f)->
+    @when_received_state_vector_listeners ?= []
+    @when_received_state_vector_listeners.push f
+
+
   #
   # You received a raw message, and you know that it is intended for to Yjs. Then call this function.
   #
@@ -208,6 +216,12 @@ module.exports =
       if sender is @user_id
         return
       if res.sync_step is "getHB"
+        # call listeners
+        if @when_received_state_vector_listeners?
+          for f in @when_received_state_vector_listeners
+            f.call this, res.data
+        delete @when_received_state_vector_listeners
+
         data = @getHB(res.data)
         hb = data.hb
         _hb = []

lib/HistoryBuffer.coffee

@@ -22,22 +22,32 @@ class HistoryBuffer
     @reserved_identifier_counter = 0
     setTimeout @emptyGarbage, @garbageCollectTimeout
 
-  resetUserId: (id)->
-    own = @buffer[@user_id]
-    if own?
-      for o_name,o of own
-        if o.uid.creator?
-          o.uid.creator = id
-        if o.uid.alt?
-          o.uid.alt.creator = id
-      if @buffer[id]?
-        throw new Error "You are re-assigning an old user id - this is not (yet) possible!"
-      @buffer[id] = own
-      delete @buffer[@user_id]
-    if @operation_counter[@user_id]?
-      @operation_counter[id] = @operation_counter[@user_id]
-      delete @operation_counter[@user_id]
-    @user_id = id
+  # At the beginning (when the user id was not assigned yet),
+  # the operations are added to buffer._temp. When you finally get your user id,
+  # the operations are copies from buffer._temp to buffer[id]. Furthermore, when buffer[id] does already contain operations
+  # (because of a previous session), the uid.op_numbers of the operations have to be reassigned.
+  # This is what this function does. It adds them to buffer[id],
+  # and assigns them the correct uid.op_number and uid.creator
+  setUserId: (@user_id, state_vector)->
+    @buffer[@user_id] ?= []
+    buff = @buffer[@user_id]
+
+    # we assumed that we started with counter = 0.
+    # when we receive tha state_vector, and actually have
+    # counter = 10. Then we have to add 10 to every op_counter
+    counter_diff = state_vector[@user_id] or 0
+
+    if @buffer._temp?
+      for o_name,o of @buffer._temp
+        o.uid.creator = @user_id
+        o.uid.op_number += counter_diff
+        buff[o.uid.op_number] = o
+
+    @operation_counter[@user_id] = (@operation_counter._temp or 0) + counter_diff
+
+    delete @operation_counter._temp
+    delete @buffer._temp
+
 
   emptyGarbage: ()=>
     for o in @garbage
@@ -207,17 +217,11 @@ class HistoryBuffer
   #
   addToCounter: (o)->
     @operation_counter[o.uid.creator] ?= 0
-    if o.uid.creator isnt @getUserId()
-      # TODO: check if operations are send in order
-      if o.uid.op_number is @operation_counter[o.uid.creator]
-        @operation_counter[o.uid.creator]++
-      while @buffer[o.uid.creator][@operation_counter[o.uid.creator]]?
-        @operation_counter[o.uid.creator]++
-      undefined
-
-    #if @operation_counter[o.uid.creator] isnt (o.uid.op_number + 1)
-      #console.log (@operation_counter[o.uid.creator] - (o.uid.op_number + 1))
-      #console.log o
-      #throw new Error "You don't receive operations in the proper order. Try counting like this 0,1,2,3,4,.. ;)"
+    # TODO: check if operations are send in order
+    if o.uid.op_number is @operation_counter[o.uid.creator]
+      @operation_counter[o.uid.creator]++
+    while @buffer[o.uid.creator][@operation_counter[o.uid.creator]]?
+      @operation_counter[o.uid.creator]++
+    undefined
 
 module.exports = HistoryBuffer

lib/Operations/Basic.coffee

@@ -430,7 +430,7 @@ module.exports = ()->
       super garbagecollect
       if callLater
         @parent.callOperationSpecificDeleteEvents(this, o)
-      if @prev_cl? and @prev_cl.isDeleted()
+      if @prev_cl? and @prev_cl.isDeleted() and @prev_cl.garbage_collected isnt true
         # garbage collect prev_cl
         @prev_cl.applyDelete()
 

lib/y.coffee

@@ -6,14 +6,13 @@ Engine = require "./Engine"
 adaptConnector = require "./ConnectorAdapter"
 
 createY = (connector)->
-  user_id = null
   if connector.user_id?
     user_id = connector.user_id # TODO: change to getUniqueId()
   else
     user_id = "_temp"
-    connector.on_user_id_set = (id)->
-      user_id = id
-      HB.resetUserId id
+    connector.when_received_state_vector_listeners = [(state_vector)->
+        HB.setUserId this.user_id, state_vector
+      ]
   HB = new HistoryBuffer user_id
   ops_manager = structured_ops_uninitialized HB, this.constructor
   ops = ops_manager.operations

package.json

@@ -1,6 +1,6 @@
 {
   "name": "yjs",
-  "version": "0.5.1",
+  "version": "0.5.2",
   "description": "A Framework that enables Real-Time Collaboration on arbitrary data structures.",
   "main": "./build/node/y.js",
   "scripts": {