v13.0.0-12 -- distribution files

package.json

@@ -1,6 +1,6 @@
 {
   "name": "yjs",
-  "version": "13.0.0-9",
+  "version": "13.0.0-12",
   "description": "A framework for real-time p2p shared editing on any data",
   "main": "./y.node.js",
   "browser": "./y.js",

src/Connector.js

@@ -43,10 +43,12 @@ export default function extendConnector (Y/* :any */) {
         this.setUserId(Y.utils.generateUserId())
       }
     }
+
     reconnect () {
       this.log('reconnecting..')
       return this.y.db.startGarbageCollector()
     }
+
     disconnect () {
       this.log('discronnecting..')
       this.connections = new Map()
@@ -56,13 +58,16 @@ export default function extendConnector (Y/* :any */) {
       this.y.db.stopGarbageCollector()
       return this.y.db.whenTransactionsFinished()
     }
+
     repair () {
       this.log('Repairing the state of Yjs. This can happen if messages get lost, and Yjs detects that something is wrong. If this happens often, please report an issue here: https://github.com/y-js/yjs/issues')
-      this.connections.forEach(user => { user.isSynced = false })
       this.isSynced = false
-      this.currentSyncTarget = null
-      this.findNextSyncTarget()
+      this.connections.forEach((user, userId) => {
+        user.isSynced = false
+        this._syncWithUser(userId)
+      })
     }
+
     setUserId (userId) {
       if (this.userId == null) {
         if (!Number.isInteger(userId)) {
@@ -77,20 +82,21 @@ export default function extendConnector (Y/* :any */) {
         return null
       }
     }
+
     onUserEvent (f) {
       this.userEventListeners.push(f)
     }
+
     removeUserEventListener (f) {
       this.userEventListeners = this.userEventListeners.filter(g => f !== g)
     }
+
     userLeft (user) {
       if (this.connections.has(user)) {
         this.log('%s: User left %s', this.userId, user)
         this.connections.delete(user)
-        if (user === this.currentSyncTarget) {
-          this.currentSyncTarget = null
-          this.findNextSyncTarget()
-        }
+        // check if isSynced event can be sent now
+        this._setSyncedWith(null)
         for (var f of this.userEventListeners) {
           f({
             action: 'userLeft',
@@ -99,7 +105,7 @@ export default function extendConnector (Y/* :any */) {
         }
       }
     }
-    userJoined (user, role) {
+    userJoined (user, role, auth) {
       if (role == null) {
         throw new Error('You must specify the role of the joined user!')
       }
@@ -112,7 +118,7 @@ export default function extendConnector (Y/* :any */) {
         isSynced: false,
         role: role,
         processAfterAuth: [],
-        auth: null,
+        auth: auth || null,
         receivedSyncStep2: false
       })
       let defer = {}
@@ -125,9 +131,7 @@ export default function extendConnector (Y/* :any */) {
           role: role
         })
       }
-      if (this.currentSyncTarget == null) {
-        this.findNextSyncTarget()
-      }
+      this._syncWithUser(user)
     }
     // Execute a function _when_ we are connected.
     // If not connected, wait until connected
@@ -138,39 +142,25 @@ export default function extendConnector (Y/* :any */) {
         this.whenSyncedListeners.push(f)
       }
     }
-    findNextSyncTarget () {
-      if (this.currentSyncTarget != null || this.role === 'slave') {
+    _syncWithUser (userid) {
+      if (this.role === 'slave') {
         return // "The current sync has not finished or this is controlled by a master!"
       }
-
-      var syncUser = null
-      for (var [uid, user] of this.connections) {
-        if (!user.isSynced) {
-          syncUser = uid
-          break
+      sendSyncStep1(this, userid)
+    }
+    _fireIsSyncedListeners () {
+      this.y.db.whenTransactionsFinished().then(() => {
+        if (!this.isSynced) {
+          this.isSynced = true
+          // It is safer to remove this!
+          // TODO: remove: yield * this.garbageCollectAfterSync()
+          // call whensynced listeners
+          for (var f of this.whenSyncedListeners) {
+            f()
+          }
+          this.whenSyncedListeners = []
         }
-      }
-      var conn = this
-      if (syncUser != null) {
-        this.currentSyncTarget = syncUser
-        sendSyncStep1(this, syncUser)
-      } else {
-        if (!conn.isSynced) {
-          this.y.db.requestTransaction(function * () {
-            if (!conn.isSynced) {
-              // it is crucial that isSynced is set at the time garbageCollectAfterSync is called
-              conn.isSynced = true
-              // It is safer to remove this!
-              // TODO: remove: yield * this.garbageCollectAfterSync()
-              // call whensynced listeners
-              for (var f of conn.whenSyncedListeners) {
-                f()
-              }
-              conn.whenSyncedListeners = []
-            }
-          })
-        }
-      }
+      })
     }
     send (uid, buffer) {
       if (!(buffer instanceof ArrayBuffer || buffer instanceof Uint8Array)) {
@@ -222,10 +212,10 @@ export default function extendConnector (Y/* :any */) {
     */
     receiveMessage (sender, buffer) {
       if (!(buffer instanceof ArrayBuffer || buffer instanceof Uint8Array)) {
-        throw new Error('Expected Message to be an ArrayBuffer or Uint8Array!')
+        return Promise.reject(new Error('Expected Message to be an ArrayBuffer or Uint8Array!'))
       }
       if (sender === this.userId) {
-        return
+        return Promise.resolve()
       }
       let decoder = new BinaryDecoder(buffer)
       let encoder = new BinaryEncoder()
@@ -244,29 +234,33 @@ export default function extendConnector (Y/* :any */) {
       if (messageType === 'sync step 1' || messageType === 'sync step 2') {
         let auth = decoder.readVarUint()
         if (senderConn.auth == null) {
-          senderConn.processAfterAuth.push([sender, buffer])
-
+          senderConn.processAfterAuth.push([messageType, senderConn, decoder, encoder, sender])
           // check auth
           return this.checkAuth(auth, this.y, sender).then(authPermissions => {
-            senderConn.auth = authPermissions
-            this.y.emit('userAuthenticated', {
-              user: senderConn.uid,
-              auth: authPermissions
-            })
-            return senderConn.syncStep2.promise
-          }).then(() => {
-            if (senderConn.processAfterAuth == null) {
-              return Promise.resolve()
+            if (senderConn.auth == null) {
+              senderConn.auth = authPermissions
+              this.y.emit('userAuthenticated', {
+                user: senderConn.uid,
+                auth: authPermissions
+              })
             }
             let messages = senderConn.processAfterAuth
-            senderConn.processAfterAuth = null
-            return Promise.all(messages.map(m =>
-              this.receiveMessage(m[0], m[1])
-            ))
+            senderConn.processAfterAuth = []
+
+            return messages.reduce((p, m) =>
+              p.then(() => this.computeMessage(m[0], m[1], m[2], m[3], m[4]))
+            , Promise.resolve())
           })
         }
       }
+      if (senderConn.auth != null) {
+        return this.computeMessage(messageType, senderConn, decoder, encoder, sender)
+      } else {
+        senderConn.processAfterAuth.push([messageType, senderConn, decoder, encoder, sender])
+      }
+    }
 
+    computeMessage (messageType, senderConn, decoder, encoder, sender) {
       if (messageType === 'sync step 1' && (senderConn.auth === 'write' || senderConn.auth === 'read')) {
         // cannot wait for sync step 1 to finish, because we may wait for sync step 2 in sync step 1 (->lock)
         computeMessageSyncStep1(decoder, encoder, this, senderConn, sender)
@@ -276,19 +270,20 @@ export default function extendConnector (Y/* :any */) {
       } else if (messageType === 'update' && senderConn.auth === 'write') {
         return computeMessageUpdate(decoder, encoder, this, senderConn, sender)
       } else {
-        Promise.reject(new Error('Unable to receive message'))
+        return Promise.reject(new Error('Unable to receive message'))
       }
     }
+
     _setSyncedWith (user) {
-      var conn = this.connections.get(user)
-      if (conn != null) {
-        conn.isSynced = true
+      if (user != null) {
+        this.connections.get(user).isSynced = true
       }
-      if (user === this.currentSyncTarget) {
-        this.currentSyncTarget = null
-        this.findNextSyncTarget()
+      let conns = Array.from(this.connections.values())
+      if (conns.length > 0 && conns.every(u => u.isSynced)) {
+        this._fireIsSyncedListeners()
       }
     }
+
     /*
       Currently, the HB encodes operations as JSON. For the moment I want to keep it
       that way. Maybe we support encoding in the HB as XML in the future, but for now I don't want

src/MessageHandler.js

@@ -92,27 +92,30 @@ export function computeMessageSyncStep1 (decoder, encoder, conn, senderConn, sen
     conn.y.destroy()
   }
 
-  // send sync step 2
-  conn.y.db.requestTransaction(function * () {
-    encoder.writeVarString('sync step 2')
-    encoder.writeVarString(conn.authInfo || '')
+  return conn.y.db.whenTransactionsFinished().then(() => {
+    // send sync step 2
+    conn.y.db.requestTransaction(function * () {
+      encoder.writeVarString('sync step 2')
+      encoder.writeVarString(conn.authInfo || '')
 
-    if (preferUntransformed) {
-      encoder.writeUint8(1)
-      yield * this.writeOperationsUntransformed(encoder)
-    } else {
-      encoder.writeUint8(0)
-      yield * this.writeOperations(encoder, decoder)
-    }
+      if (preferUntransformed) {
+        encoder.writeUint8(1)
+        yield * this.writeOperationsUntransformed(encoder)
+      } else {
+        encoder.writeUint8(0)
+        yield * this.writeOperations(encoder, decoder)
+      }
 
-    yield * this.writeDeleteSet(encoder)
-    conn.send(senderConn.uid, encoder.createBuffer())
-    senderConn.receivedSyncStep2 = true
+      yield * this.writeDeleteSet(encoder)
+      conn.send(senderConn.uid, encoder.createBuffer())
+      senderConn.receivedSyncStep2 = true
+    })
+    return conn.y.db.whenTransactionsFinished().then(() => {
+      if (conn.role === 'slave') {
+        sendSyncStep1(conn, sender)
+      }
+    })
   })
-  if (conn.role === 'slave') {
-    sendSyncStep1(conn, sender)
-  }
-  return conn.y.db.whenTransactionsFinished()
 }
 
 export function logSS (decoder, strBuilder) {

tests-lib/helper.js

@@ -141,7 +141,7 @@ export async function initArrays (t, opts) {
     let connOpts
     if (i === 0) {
       // Only one instance can gc!
-      dbOpts = Object.assign({ gc: true }, opts.db)
+      dbOpts = Object.assign({ gc: false }, opts.db)
       connOpts = Object.assign({ role: 'master' }, connector)
     } else {
       dbOpts = Object.assign({ gc: false }, opts.db)

y.node.js

@@ -1,7 +1,7 @@
 
 /**
  * yjs - A framework for real-time p2p shared editing on any data
- * @version v13.0.0-9
+ * @version v13.0.0-12
  * @license MIT
  */
 
@@ -506,27 +506,30 @@ function computeMessageSyncStep1 (decoder, encoder, conn, senderConn, sender) {
     conn.y.destroy();
   }
 
-  // send sync step 2
-  conn.y.db.requestTransaction(function * () {
-    encoder.writeVarString('sync step 2');
-    encoder.writeVarString(conn.authInfo || '');
+  return conn.y.db.whenTransactionsFinished().then(() => {
+    // send sync step 2
+    conn.y.db.requestTransaction(function * () {
+      encoder.writeVarString('sync step 2');
+      encoder.writeVarString(conn.authInfo || '');
 
-    if (preferUntransformed) {
-      encoder.writeUint8(1);
-      yield * this.writeOperationsUntransformed(encoder);
-    } else {
-      encoder.writeUint8(0);
-      yield * this.writeOperations(encoder, decoder);
-    }
+      if (preferUntransformed) {
+        encoder.writeUint8(1);
+        yield * this.writeOperationsUntransformed(encoder);
+      } else {
+        encoder.writeUint8(0);
+        yield * this.writeOperations(encoder, decoder);
+      }
 
-    yield * this.writeDeleteSet(encoder);
-    conn.send(senderConn.uid, encoder.createBuffer());
-    senderConn.receivedSyncStep2 = true;
-  });
-  if (conn.role === 'slave') {
-    sendSyncStep1(conn, sender);
-  }
-  return conn.y.db.whenTransactionsFinished()
+      yield * this.writeDeleteSet(encoder);
+      conn.send(senderConn.uid, encoder.createBuffer());
+      senderConn.receivedSyncStep2 = true;
+    });
+    return conn.y.db.whenTransactionsFinished().then(() => {
+      if (conn.role === 'slave') {
+        sendSyncStep1(conn, sender);
+      }
+    })
+  })
 }
 
 function logSS (decoder, strBuilder) {
@@ -640,10 +643,12 @@ function extendConnector (Y/* :any */) {
         this.setUserId(Y.utils.generateUserId());
       }
     }
+
     reconnect () {
       this.log('reconnecting..');
       return this.y.db.startGarbageCollector()
     }
+
     disconnect () {
       this.log('discronnecting..');
       this.connections = new Map();
@@ -653,13 +658,16 @@ function extendConnector (Y/* :any */) {
       this.y.db.stopGarbageCollector();
       return this.y.db.whenTransactionsFinished()
     }
+
     repair () {
       this.log('Repairing the state of Yjs. This can happen if messages get lost, and Yjs detects that something is wrong. If this happens often, please report an issue here: https://github.com/y-js/yjs/issues');
-      this.connections.forEach(user => { user.isSynced = false; });
       this.isSynced = false;
-      this.currentSyncTarget = null;
-      this.findNextSyncTarget();
+      this.connections.forEach((user, userId) => {
+        user.isSynced = false;
+        this._syncWithUser(userId);
+      });
     }
+
     setUserId (userId) {
       if (this.userId == null) {
         if (!Number.isInteger(userId)) {
@@ -674,20 +682,21 @@ function extendConnector (Y/* :any */) {
         return null
       }
     }
+
     onUserEvent (f) {
       this.userEventListeners.push(f);
     }
+
     removeUserEventListener (f) {
       this.userEventListeners = this.userEventListeners.filter(g => f !== g);
     }
+
     userLeft (user) {
       if (this.connections.has(user)) {
         this.log('%s: User left %s', this.userId, user);
         this.connections.delete(user);
-        if (user === this.currentSyncTarget) {
-          this.currentSyncTarget = null;
-          this.findNextSyncTarget();
-        }
+        // check if isSynced event can be sent now
+        this._setSyncedWith(null);
         for (var f of this.userEventListeners) {
           f({
             action: 'userLeft',
@@ -696,7 +705,7 @@ function extendConnector (Y/* :any */) {
         }
       }
     }
-    userJoined (user, role) {
+    userJoined (user, role, auth) {
       if (role == null) {
         throw new Error('You must specify the role of the joined user!')
       }
@@ -709,7 +718,7 @@ function extendConnector (Y/* :any */) {
         isSynced: false,
         role: role,
         processAfterAuth: [],
-        auth: null,
+        auth: auth || null,
         receivedSyncStep2: false
       });
       let defer = {};
@@ -722,9 +731,7 @@ function extendConnector (Y/* :any */) {
           role: role
         });
       }
-      if (this.currentSyncTarget == null) {
-        this.findNextSyncTarget();
-      }
+      this._syncWithUser(user);
     }
     // Execute a function _when_ we are connected.
     // If not connected, wait until connected
@@ -735,39 +742,25 @@ function extendConnector (Y/* :any */) {
         this.whenSyncedListeners.push(f);
       }
     }
-    findNextSyncTarget () {
-      if (this.currentSyncTarget != null || this.role === 'slave') {
+    _syncWithUser (userid) {
+      if (this.role === 'slave') {
         return // "The current sync has not finished or this is controlled by a master!"
       }
-
-      var syncUser = null;
-      for (var [uid, user] of this.connections) {
-        if (!user.isSynced) {
-          syncUser = uid;
-          break
+      sendSyncStep1(this, userid);
+    }
+    _fireIsSyncedListeners () {
+      this.y.db.whenTransactionsFinished().then(() => {
+        if (!this.isSynced) {
+          this.isSynced = true;
+          // It is safer to remove this!
+          // TODO: remove: yield * this.garbageCollectAfterSync()
+          // call whensynced listeners
+          for (var f of this.whenSyncedListeners) {
+            f();
+          }
+          this.whenSyncedListeners = [];
         }
-      }
-      var conn = this;
-      if (syncUser != null) {
-        this.currentSyncTarget = syncUser;
-        sendSyncStep1(this, syncUser);
-      } else {
-        if (!conn.isSynced) {
-          this.y.db.requestTransaction(function * () {
-            if (!conn.isSynced) {
-              // it is crucial that isSynced is set at the time garbageCollectAfterSync is called
-              conn.isSynced = true;
-              // It is safer to remove this!
-              // TODO: remove: yield * this.garbageCollectAfterSync()
-              // call whensynced listeners
-              for (var f of conn.whenSyncedListeners) {
-                f();
-              }
-              conn.whenSyncedListeners = [];
-            }
-          });
-        }
-      }
+      });
     }
     send (uid, buffer) {
       if (!(buffer instanceof ArrayBuffer || buffer instanceof Uint8Array)) {
@@ -819,10 +812,10 @@ function extendConnector (Y/* :any */) {
     */
     receiveMessage (sender, buffer) {
       if (!(buffer instanceof ArrayBuffer || buffer instanceof Uint8Array)) {
-        throw new Error('Expected Message to be an ArrayBuffer or Uint8Array!')
+        return Promise.reject(new Error('Expected Message to be an ArrayBuffer or Uint8Array!'))
       }
       if (sender === this.userId) {
-        return
+        return Promise.resolve()
       }
       let decoder = new BinaryDecoder(buffer);
       let encoder = new BinaryEncoder();
@@ -841,29 +834,33 @@ function extendConnector (Y/* :any */) {
       if (messageType === 'sync step 1' || messageType === 'sync step 2') {
         let auth = decoder.readVarUint();
         if (senderConn.auth == null) {
-          senderConn.processAfterAuth.push([sender, buffer]);
-
+          senderConn.processAfterAuth.push([messageType, senderConn, decoder, encoder, sender]);
           // check auth
           return this.checkAuth(auth, this.y, sender).then(authPermissions => {
-            senderConn.auth = authPermissions;
-            this.y.emit('userAuthenticated', {
-              user: senderConn.uid,
-              auth: authPermissions
-            });
-            return senderConn.syncStep2.promise
-          }).then(() => {
-            if (senderConn.processAfterAuth == null) {
-              return Promise.resolve()
+            if (senderConn.auth == null) {
+              senderConn.auth = authPermissions;
+              this.y.emit('userAuthenticated', {
+                user: senderConn.uid,
+                auth: authPermissions
+              });
             }
             let messages = senderConn.processAfterAuth;
-            senderConn.processAfterAuth = null;
-            return Promise.all(messages.map(m =>
-              this.receiveMessage(m[0], m[1])
-            ))
+            senderConn.processAfterAuth = [];
+
+            return messages.reduce((p, m) =>
+              p.then(() => this.computeMessage(m[0], m[1], m[2], m[3], m[4]))
+            , Promise.resolve())
           })
         }
       }
+      if (senderConn.auth != null) {
+        return this.computeMessage(messageType, senderConn, decoder, encoder, sender)
+      } else {
+        senderConn.processAfterAuth.push([messageType, senderConn, decoder, encoder, sender]);
+      }
+    }
 
+    computeMessage (messageType, senderConn, decoder, encoder, sender) {
       if (messageType === 'sync step 1' && (senderConn.auth === 'write' || senderConn.auth === 'read')) {
         // cannot wait for sync step 1 to finish, because we may wait for sync step 2 in sync step 1 (->lock)
         computeMessageSyncStep1(decoder, encoder, this, senderConn, sender);
@@ -873,19 +870,20 @@ function extendConnector (Y/* :any */) {
       } else if (messageType === 'update' && senderConn.auth === 'write') {
         return computeMessageUpdate(decoder, encoder, this, senderConn, sender)
       } else {
-        Promise.reject(new Error('Unable to receive message'));
+        return Promise.reject(new Error('Unable to receive message'))
       }
     }
+
     _setSyncedWith (user) {
-      var conn = this.connections.get(user);
-      if (conn != null) {
-        conn.isSynced = true;
+      if (user != null) {
+        this.connections.get(user).isSynced = true;
       }
-      if (user === this.currentSyncTarget) {
-        this.currentSyncTarget = null;
-        this.findNextSyncTarget();
+      let conns = Array.from(this.connections.values());
+      if (conns.length > 0 && conns.every(u => u.isSynced)) {
+        this._fireIsSyncedListeners();
       }
     }
+
     /*
       Currently, the HB encodes operations as JSON. For the moment I want to keep it
       that way. Maybe we support encoding in the HB as XML in the future, but for now I don't want