Alternative Presence Implementation from Propeller Factory

lib/agent.js

@@ -56,11 +56,11 @@ Agent.prototype.close = function(err) {
 
 Agent.prototype._cleanup = function() {
   this.closed = true;
-
   // Clean up doc subscription streams
   for (var collection in this.subscribedDocs) {
     var docs = this.subscribedDocs[collection];
     for (var id in docs) {
+      this.backend.notifyUnsubscribe(this.clientId, collection, id)
       var stream = docs[id];
       stream.destroy();
     }
@@ -98,6 +98,17 @@ Agent.prototype._subscribeToStream = function(collection, id, stream) {
       console.error('Doc subscription stream error', collection, id, data.error);
       return;
     }
+    if (data.a === 'pr') {
+      // Received a presense publish from another client
+      if (data.src !== agent.clientId) agent.send(data);
+      return
+    }
+    if (data.a === 'nus') {
+      // Another client has unsubscribed from this document
+      if (data.src !== agent.clientId) agent.send(data)
+      return
+    }
+
     if (agent._isOwnOp(collection, data)) return;
     agent._sendOp(collection, id, data);
   });
@@ -176,6 +187,7 @@ Agent.prototype._sendOp = function(collection, id, op) {
   if (op.op) message.op = op.op;
   if (op.create) message.create = op.create;
   if (op.del) message.del = true;
+  if (op.pr) message.pr = op.pr
 
   this.send(message);
 };
@@ -299,7 +311,10 @@ Agent.prototype._handleMessage = function(request, callback) {
       case 'op':
         var op = this._createOp(request);
         if (!op) return callback({code: 4000, message: 'Invalid op message'});
-        return this._submit(request.c, request.d, op, callback);
+        return this._submit(request.c, request.d, op, request.pr, callback);
+      case 'pr': {
+        return this._sendPresence(request, callback)
+      }
       case 'nf':
         return this._fetchSnapshot(request.c, request.d, request.v, callback);
       default:
@@ -496,6 +511,70 @@ Agent.prototype._subscribe = function(collection, id, version, callback) {
   });
 };
 
+
+Agent.prototype._sendPresence = function(request, callback) {
+  const collection = request.c
+  const id = request.d
+  const version = request.v
+  const source = request.src || this.clientId
+  const from = version
+  const backend = this.backend
+  const agent = this.agent
+  // Send a special projection so that getSnapshot knows to return all fields.
+  // With a null projection, it strips document metadata
+  var fields = {$submit: true};
+
+  backend.db.getSnapshot(collection, id, fields, null, function(err, snapshot) {
+    if (err) return callback(err);
+
+    if (version === snapshot.v) {
+        // Send presense to other clients
+        // We don't need to transform because there is no newer version of the document to transform to
+        var message = {
+          a: 'pr',
+          c: collection,
+          d: id,
+          v: version,
+          src: source,
+          pr: request.pr
+        };
+
+        backend.sendPresence(message, function(err) {
+          if (err) return callback(err);
+        });
+    } else {
+      // The version of the presence doesn't match the snapshot so fetch the ops
+      // and give middleware chance to transform the presence
+      backend.db.getOpsToSnapshot(collection, id, from, snapshot, null, function(err, ops) {
+        if (err) return callback(err);
+
+        if (ops.length !== snapshot.v - from) {
+          return callback(request.missingOpsError());
+        }
+
+        request.opsToSnapshot = ops
+        backend.trigger('transformPresence', agent, request, function(err) {
+          if (err) return callback(err);
+
+          var message = {
+            a: 'pr',
+            c: collection,
+            d: id,
+            v: version,
+            src: source,
+            pr: request.pr
+          };
+
+          // Send presense to other clients
+          backend.sendPresence(message, function(err) {
+            if (err) return callback(err);
+          });
+        })
+      })
+    }
+  })
+}
+
 Agent.prototype._subscribeBulk = function(collection, versions, callback) {
   var agent = this;
   this.backend.subscribeBulk(this, collection, versions, function(err, streams, snapshotMap) {
@@ -516,7 +595,10 @@ Agent.prototype._unsubscribe = function(collection, id, callback) {
   // stream or an inflight subscribing state
   var docs = this.subscribedDocs[collection];
   var stream = docs && docs[id];
-  if (stream) stream.destroy();
+  if (stream) {
+    stream.destroy();
+    this.backend.notifyUnsubscribe(this.clientId, collection, id)
+  }
   process.nextTick(callback);
 };
 
@@ -526,14 +608,17 @@ Agent.prototype._unsubscribeBulk = function(collection, ids, callback) {
   for (var i = 0; i < ids.length; i++) {
     var id = ids[i];
     var stream = docs[id];
-    if (stream) stream.destroy();
+    if (stream) {
+      this.backend.notifyUnsubscribe(this.clientId, collection, id)
+      stream.destroy();
+    }
   }
   process.nextTick(callback);
 };
 
-Agent.prototype._submit = function(collection, id, op, callback) {
+Agent.prototype._submit = function(collection, id, op, presence, callback) {
   var agent = this;
-  this.backend.submit(this, collection, id, op, null, function(err, ops) {
+  this.backend.submit(this, collection, id, op, presence, null, function(err, ops) {
     // Message to acknowledge the op was successfully submitted
     var ack = {src: op.src, seq: op.seq, v: op.v};
     if (err) {
@@ -588,3 +673,7 @@ Agent.prototype._createOp = function(request) {
 Agent.prototype._fetchSnapshot = function (collection, id, version, callback) {
   this.backend.fetchSnapshot(this, collection, id, version, callback);
 };
+
+Agent.prototype.missingOpsError = function() {
+  return {code: 5019, message: 'Presence send failed. DB missing ops needed to transform it up to the current snapshot version'};
+};

lib/backend.js

@@ -218,10 +218,10 @@ Backend.prototype.trigger = function(action, agent, request, callback) {
 // Submit an operation on the named collection/docname. op should contain a
 // {op:}, {create:} or {del:} field. It should probably contain a v: field (if
 // it doesn't, it defaults to the current version).
-Backend.prototype.submit = function(agent, index, id, op, options, callback) {
+Backend.prototype.submit = function(agent, index, id, op, presence, options, callback) {
   var err = ot.checkOp(op);
   if (err) return callback(err);
-  var request = new SubmitRequest(this, agent, index, id, op, options);
+  var request = new SubmitRequest(this, agent, index, id, op, presence, options);
   var backend = this;
   backend.trigger(backend.MIDDLEWARE_ACTIONS.submit, agent, request, function(err) {
     if (err) return callback(err);
@@ -239,6 +239,22 @@ Backend.prototype.submit = function(agent, index, id, op, options, callback) {
   });
 };
 
+Backend.prototype.sendPresence = function(presence, callback) {
+  var channels = [ this.getDocChannel(presence.c, presence.d) ];
+  this.pubsub.publish(channels, presence, callback);
+};
+
+Backend.prototype.notifyUnsubscribe = function(clientId, collectionId, docId, callback) {
+  const message = {
+    a: 'nus',
+    c: collectionId,
+    d: docId,
+    src: clientId
+  }
+
+  this.pubsub.publish([this.getDocChannel(collectionId, docId)], message, callback)
+}
+
 Backend.prototype._sanitizeOp = function(agent, projection, collection, id, op, callback) {
   if (projection) {
     try {

lib/client/connection.js

@@ -250,7 +250,14 @@ Connection.prototype.handleMessage = function(message) {
       var doc = this.getExisting(message.c, message.d);
       if (doc) doc._handleOp(err, message);
       return;
-
+    case 'pr':
+      var doc = this.getExisting(message.c, message.d);
+      if (doc) doc._handlePresence(err, message)
+      return;
+    case 'nus':
+      var doc = this.getExisting(message.c, message.d)
+      if (doc) doc._handleUnsubscribeNotification(err, message)
+      return;
     default:
       console.warn('Ignoring unrecognized message', message);
   }
@@ -418,9 +425,21 @@ Connection.prototype.sendOp = function(doc, op) {
   if (op.op) message.op = op.op;
   if (op.create) message.create = op.create;
   if (op.del) message.del = op.del;
+  if (op.pr) message.pr = op.pr
   this.send(message);
 };
 
+Connection.prototype.sendPresence = function(doc, presence) {
+  var message = {
+    a: 'pr',
+    c: doc.collection,
+    d: doc.id,
+    v: doc.version,
+    src: this.id,
+    pr: presence
+  };
+  this.send(message);
+}
 
 /**
  * Sends a message down the socket