[WIP] Refactor

Author: Alec Gibson

lib/agent.js

@@ -27,8 +27,12 @@ function Agent(backend, stream) {
   this.subscribedQueries = {};
 
   // Track which documents are subscribed to presence by the client. This is a
-  // map of collection -> id -> stream
+  // map of channel -> stream
   this.subscribedPresences = {};
+  // Highest seq received for a subscription request. Any seq lower than this
+  // value is stale, and should be ignored. Used for keeping the subscription
+  // state in sync with the client's desired state
+  this.presenceSubscriptionSeq = 0;
 
   // We need to track this manually to make sure we don't reply to messages
   // after the stream was closed.
@@ -78,12 +82,8 @@ Agent.prototype._cleanup = function() {
   }
   this.subscribedDocs = {};
 
-  for (var collection in this.subscribedPresences) {
-    var streams = this.subscribedPresences[collection];
-    for (var id in streams) {
-      var stream = streams[id];
-      stream.destroy();
-    }
+  for (var channel in this.subscribedPresences) {
+    this.subscribedPresences[channel].destroy();
   }
   this.subscribedPresences = {};
 
@@ -130,22 +130,19 @@ Agent.prototype._subscribeToStream = function(collection, id, stream) {
   });
 };
 
-Agent.prototype._subscribeToPresenceStream = function(collection, id, stream) {
+Agent.prototype._subscribeToPresenceStream = function(channel, stream) {
   if (this.closed) return stream.destroy();
 
   stream.on('data', function(data) {
     if (data.error) {
-      logger.error('Presence subscription stream error', collection, id, data.error);
+      logger.error('Presence subscription stream error', channel, data.error);
     }
     this._handlePresenceData(data);
   }.bind(this));
 
   stream.on('end', function() {
-    var streams = this.subscribedPresences[collection];
-    if (!streams || !streams[id] !== stream) return;
-    delete streams[id];
-    if (util.hasKeys(streams)) return;
-    delete agent.subscribedPresences[collection];
+    if (this.subscribedPresences[channel] !== stream) return;
+    delete this.subscribedPresences[channel];
   }.bind(this));
 };
 
@@ -394,10 +391,14 @@ Agent.prototype._handleMessage = function(request, callback) {
         return this._fetchSnapshotByTimestamp(request.c, request.d, request.ts, callback);
       case 'p':
         var presence = this._createPresence(request);
-        if (!util.supportsPresence(types.map[presence.t])) {
+        if (presence.t && !util.supportsPresence(types.map[presence.t])) {
           return callback({code: 9999, message: 'Type does not support presence: ' + presence.t});
         }
-        return this._broadcastPresence(request.c, request.d, presence, callback);
+        return this._broadcastPresence(presence, callback);
+      case 'ps':
+        return this._subscribePresence(request.ch, request.seq, callback);
+      case 'pu':
+        return this._unsubscribePresence(request.ch, request.seq, callback);
       default:
         callback({code: 4000, message: 'Invalid or unknown message'});
     }
@@ -666,77 +667,82 @@ Agent.prototype._fetchSnapshotByTimestamp = function(collection, id, timestamp,
   this.backend.fetchSnapshotByTimestamp(this, collection, id, timestamp, callback);
 };
 
-Agent.prototype._broadcastPresence = function(collection, id, presence, callback) {
-  var wantsSubscribe = presence.s;
-  this._handlePresenceSubscription(collection, id, wantsSubscribe, function(error) {
+Agent.prototype._broadcastPresence = function(presence, callback) {
+  this.backend.transformPresenceToLatestVersion(this, presence, function(error, presence) {
     if (error) return callback(error);
-    this.backend.transformPresenceToLatestVersion(this, presence, function(error, presence) {
+    var channel = this._getPresenceChannel(presence.ch);
+    this.backend.pubsub.publish([channel], presence, function(error) {
       if (error) return callback(error);
-      var channel = this.backend.getPresenceChannel(collection, id);
-      this.backend.pubsub.publish([channel], presence, function(error) {
-        if (error) return callback(error);
-        callback(null, presence);
-      });
-    }.bind(this));
-  }.bind(this));
-};
-
-Agent.prototype._handlePresenceSubscription = function(collection, id, wantsSubscribe, callback) {
-  var streams = this.subscribedPresences[collection] || (this.subscribedPresences[collection] = {});
-  var stream = streams[id];
-
-  if (stream) {
-    if (wantsSubscribe) {
-      return callback();
-    }
-    stream.destroy();
-    return callback();
-  }
-
-  if (!wantsSubscribe) return callback();
-
-  var channel = this.backend.getPresenceChannel(collection, id);
-  this.backend.pubsub.subscribe(channel, function(error, stream) {
-    if (error) return callback(error);
-    streams[id] = stream;
-    this._subscribeToPresenceStream(collection, id, stream);
-    callback();
+      callback(null, presence);
+    });
   }.bind(this));
 };
 
 Agent.prototype._createPresence = function(request) {
-  // src can be provided if it is not the same as the current agent,
-  // such as a resubmission after a reconnect, but it usually isn't needed
-  var src = request.src || this.clientId;
   return {
     a: 'p',
-    src: src,
+    ch: request.ch,
+    src: this.clientId,
     seq: request.seq,
+    id: request.id,
+    p: request.p,
     c: request.c,
     d: request.d,
-    id: request.id,
     v: request.v,
-    p: request.p,
-    t: request.t,
-    r: !!request.r,
-    s: !!request.s
+    t: request.t
   };
 };
 
+Agent.prototype._subscribePresence = function(channel, seq, callback) {
+  var presenceChannel = this._getPresenceChannel(channel);
+  this.backend.pubsub.subscribe(presenceChannel, function(error, stream) {
+    if (error) return callback(error);
+    if (seq < this.presenceSubscriptionSeq) return callback(null, {ch: channel, seq: seq});
+    this.presenceSubscriptionSeq = seq;
+    this.subscribedPresences[channel] = stream;
+    this._subscribeToPresenceStream(channel, stream);
+    this._requestPresence(channel, function(error) {
+      callback(error, {ch: channel, seq: seq});
+    });
+  }.bind(this));
+};
+
+Agent.prototype._unsubscribePresence = function(channel, seq, callback) {
+  if (seq < this.presenceSubscriptionSeq) return;
+  this.presenceSubscriptionSeq = seq;
+  var stream = this.subscribedPresences[channel];
+  if (stream) stream.destroy();
+  callback(null, {ch: channel, seq: seq});
+};
+
+Agent.prototype._getPresenceChannel = function(channel) {
+  // TODO: May need to namespace this further if we want to have automatic Doc channels that don't
+  // clash with arbitrary user input (eg if a user decides to name their channel the same as a doc collection)
+  // TODO: What if a user creates a collection called _presence?
+  return '_presence.' + channel;
+};
+
+Agent.prototype._requestPresence = function(channel, callback) {
+  var presenceChannel = this._getPresenceChannel(channel);
+  this.backend.pubsub.publish([presenceChannel], {ch: channel, r: true, src: this.clientId}, callback);
+};
+
 Agent.prototype._handlePresenceData = function(presence) {
-  if (presence.src !== this.clientId) {
-    var backend = this.backend;
-    var context = {
-      collection: presence.c,
-      presence: presence
-    };
-    backend.trigger(backend.MIDDLEWARE_ACTIONS.sendPresence, this, context, function(error) {
-      if (error) {
-        return this.send({a: 'p', c: presence.c, d: presence.d, id: presence.id, error: getReplyErrorObject(error)});
-      }
-      this.send(presence);
-    }.bind(this));
-  }
+  if (presence.src === this.clientId) return;
+
+  if (presence.r) return this.send({a: 'pr', ch: presence.ch});
+
+  var backend = this.backend;
+  var context = {
+    collection: presence.c,
+    presence: presence
+  };
+  backend.trigger(backend.MIDDLEWARE_ACTIONS.sendPresence, this, context, function(error) {
+    if (error) {
+      return this.send({a: 'p', ch: presence.ch, id: presence.id, error: getReplyErrorObject(error)});
+    }
+    this.send(presence);
+  }.bind(this));
 };
 
 function createClientOp(request, clientId) {

lib/backend.js

@@ -682,10 +682,6 @@ Backend.prototype.getDocChannel = function(collection, id) {
   return collection + '.' + id;
 };
 
-Backend.prototype.getPresenceChannel = function(collection, id) {
-  return this.getDocChannel(collection, id) + '.presence';
-};
-
 Backend.prototype.getChannels = function(collection, id) {
   return [
     this.getCollectionChannel(collection),
@@ -807,6 +803,7 @@ Backend.prototype._buildSnapshotFromOps = function(id, startingSnapshot, ops, ca
 };
 
 Backend.prototype.transformPresenceToLatestVersion = function(agent, presence, callback) {
+  if (!presence.c || !presence.d) return callback(null, presence);
   this.getOps(agent, presence.c, presence.d, presence.v, null, function(error, ops) {
     if (error) return callback(error);
     for (var i = 0; i < ops.length; i++) {