WIP Presence

.gitignore

@@ -34,4 +34,5 @@ coverage
 # Dependency directories
 node_modules
 package-lock.json
+yarn.lock
 jspm_packages

README.md

@@ -19,6 +19,7 @@ tracker](https://github.com/share/sharedb/issues).
 
 - Realtime synchronization of any JSON document
 - Concurrent multi-user collaboration
+- Realtime synchronization of any ephemeral "presence" data
 - Synchronous editing API with asynchronous eventual consistency
 - Realtime query subscriptions
 - Simple integration with any database - [MongoDB](https://github.com/share/sharedb-mongo), [PostgresQL](https://github.com/share/sharedb-postgres) (experimental)
@@ -73,6 +74,12 @@ initial data. Then you can submit editing operations on the document (using
 OT). Finally you can delete the document with a delete operation. By
 default, ShareDB stores all operations forever - nothing is truly deleted.
 
+## User presence synchronization
+
+ShareDB supports synchronization of user presence data. This feature is opt-in, not enabled by default. To enable this feature, pass the `enablePresence: true` option to the ShareDB constructor (e.g. `var share = new ShareDB({ enablePresence: true })`).
+
+Presence data represents a user and is automatically synchronized between all clients subscribed to the same document. Its format is defined by the document's [OT Type](https://github.com/ottypes/docs), for example it may contain a user ID and a cursor position in a text document. All clients can modify their own presence data and receive a read-only version of other client's data. Presence data is automatically cleared when a client unsubscribes from the document or disconnects. It is also automatically transformed against applied operations, so that it still makes sense in the context of a modified document, for example a cursor position may be automatically advanced when a user types at the beginning of a text document.
+
 ## Server API
 
 ### Initialization
@@ -91,6 +98,8 @@ __Options__
 * `options.pubsub` _(instance of `ShareDB.PubSub`)_
   Notify other ShareDB processes when data changes
   through this pub/sub adapter. Defaults to `ShareDB.MemoryPubSub()`.
+* `options.Presence` _(implementation of `ShareDB.Presence`)_
+  Enable user presence synchronization. If not specified, presence features are not enabled.
 
 #### Database Adapters
 * `ShareDB.MemoryDB`, backed by a non-persistent database with no queries
@@ -308,6 +317,9 @@ Unique document ID
 `doc.data` _(Object)_
 Document contents. Available after document is fetched or subscribed to.
 
+`doc.presence.current` _(Object)_
+Each property under `doc.presence.current` contains presence data shared by a client subscribed to this document. The property name is an empty string for this client's data and connection IDs for other clients' data.
+
 `doc.fetch(function(err) {...})`
 Populate the fields on `doc` with a snapshot of the document from the server.
 
@@ -337,6 +349,9 @@ An operation was applied to the data. `source` will be `false` for ops received
 `doc.on('del', function(data, source) {...})`
 The document was deleted. Document contents before deletion are passed in as an argument. `source` will be `false` for ops received from the server and defaults to `true` for ops generated locally.
 
+`doc.on('presence', function(srcList, submitted) {...})`
+Presence data has changed. `srcList` is an Array of `doc.presence` property names for which values have changed. `submitted` is `true`, if the event is the result of new presence data being submitted by the local or remote user, otherwise it is `false` - eg if the presence data was transformed against an operation or was cleared on unsubscribe, disconnect or roll-back.
+
 `doc.on('error', function(err) {...})`
 There was an error fetching the document or applying an operation.
 
@@ -370,6 +385,11 @@ Invokes the given callback function after
 
 Note that `whenNothingPending` does NOT wait for pending `model.query()` calls.
 
+`doc.submitPresence(presenceData[, function(err) {...}])`
+Set local presence data and publish it for other clients.
+`presenceData` structure depends on the document type.
+Presence is synchronized only when subscribed to the document.
+
 ### Class: `ShareDB.Query`
 
 `query.ready` _(Boolean)_
@@ -467,6 +487,9 @@ Additional fields may be added to the error object for debugging context dependi
 * 4022 - Database adapter does not support queries
 * 4023 - Cannot project snapshots of this type
 * 4024 - Invalid version
+* 4025 - Not subscribed to document
+* 4026 - Presence data superseded
+* 4027 - OT Type does not support presence
 
 ### 5000 - Internal error
 

lib/agent.js

@@ -2,6 +2,7 @@ var hat = require('hat');
 var util = require('./util');
 var types = require('./types');
 var logger = require('./logger');
+var ShareDBError = require('./error');
 
 /**
  * Agent deserializes the wire protocol messages received from the stream and
@@ -26,6 +27,9 @@ function Agent(backend, stream) {
   // Map from queryId -> emitter
   this.subscribedQueries = {};
 
+  // The max presence sequence number received from the client.
+  this.maxPresenceSeq = 0;
+
   // We need to track this manually to make sure we don't reply to messages
   // after the stream was closed.
   this.closed = false;
@@ -106,10 +110,17 @@ Agent.prototype._subscribeToStream = function(collection, id, stream) {
       logger.error('Doc subscription stream error', collection, id, data.error);
       return;
     }
+    if (data.a === 'p') {
+      // Send other clients' presence data
+      if (data.src !== agent.clientId) agent.send(data);
+      return;
+    }
     if (agent._isOwnOp(collection, data)) return;
     agent._sendOp(collection, id, data);
   });
   stream.on('end', function() {
+    var presence = agent._createPresence(collection, id);
+    agent.backend.sendPresence(presence);
     // The op stream is done sending, so release its reference
     var streams = agent.subscribedDocs[collection];
     if (!streams || streams[id] !== stream) return;
@@ -288,6 +299,13 @@ Agent.prototype._checkRequest = function(request) {
     // Bulk request
     if (request.c != null && typeof request.c !== 'string') return 'Invalid collection';
     if (typeof request.b !== 'object') return 'Invalid bulk subscribe data';
+  } else if (request.a === 'p') {
+    // Presence
+    if (typeof request.c !== 'string') return 'Invalid collection';
+    if (typeof request.d !== 'string') return 'Invalid id';
+    if (typeof request.v !== 'number' || request.v < 0) return 'Invalid version';
+    if (typeof request.seq !== 'number' || request.seq <= 0) return 'Invalid seq';
+    if (typeof request.r !== 'undefined' && typeof request.r !== 'boolean') return 'Invalid "request reply" value';
   }
 };
 
@@ -324,6 +342,9 @@ Agent.prototype._handleMessage = function(request, callback) {
         return this._fetchSnapshot(request.c, request.d, request.v, callback);
       case 'nt':
         return this._fetchSnapshotByTimestamp(request.c, request.d, request.ts, callback);
+      case 'p':
+        var presence = this._createPresence(request.c, request.d, request.p, request.v, request.r, request.seq);
+        return this._presence(presence, callback);
       default:
         callback({code: 4000, message: 'Invalid or unknown message'});
     }
@@ -614,3 +635,34 @@ Agent.prototype._fetchSnapshot = function (collection, id, version, callback) {
 Agent.prototype._fetchSnapshotByTimestamp = function (collection, id, timestamp, callback) {
   this.backend.fetchSnapshotByTimestamp(this, collection, id, timestamp, callback);
 };
+
+Agent.prototype._presence = function(presence, callback) {
+  if (presence.seq <= this.maxPresenceSeq) {
+    return process.nextTick(function() {
+      callback(new ShareDBError(4026, 'Presence data superseded'));
+    });
+  }
+  this.maxPresenceSeq = presence.seq;
+  if (!this.subscribedDocs[presence.c] || !this.subscribedDocs[presence.c][presence.d]) {
+    return process.nextTick(function() {
+      callback(new ShareDBError(4025, 'Cannot send presence. Not subscribed to document: ' + presence.c + ' ' + presence.d));
+    });
+  }
+  this.backend.sendPresence(presence, function(err) {
+    if (err) return callback(err);
+    callback(null, { seq: presence.seq });
+  });
+};
+
+Agent.prototype._createPresence = function(collection, id, data, version, requestReply, seq) {
+  return {
+    a: 'p',
+    src: this.clientId,
+    seq: seq != null ? seq : this.maxPresenceSeq,
+    c: collection,
+    d: id,
+    p: data,
+    v: version,
+    r: requestReply
+  };
+};

lib/backend.js

@@ -48,6 +48,8 @@ function Backend(options) {
   if (!options.disableSpaceDelimitedActions) {
     this._shimAfterSubmit();
   }
+
+  this.Presence = options.Presence;
 }
 module.exports = Backend;
 emitter.mixin(Backend);
@@ -155,6 +157,11 @@ Backend.prototype.connect = function(connection, req) {
   // not used internal to ShareDB, but it is handy for server-side only user
   // code that may cache state on the agent and read it in middleware
   connection.agent = agent;
+
+  // Pass through information on whether or not presence is enabled,
+  // so that Doc instances can use it.
+  connection.Presence = this.Presence;
+
   return connection;
 };
 
@@ -720,6 +727,11 @@ Backend.prototype._buildSnapshotFromOps = function (id, startingSnapshot, ops, c
   callback(error, snapshot);
 };
 
+Backend.prototype.sendPresence = function(presence, callback) {
+  var channels = [ this.getDocChannel(presence.c, presence.d) ];
+  this.pubsub.publish(channels, presence, callback);
+};
+
 function pluckIds(snapshots) {
   var ids = [];
   for (var i = 0; i < snapshots.length; i++) {

lib/client/connection.js

@@ -254,6 +254,11 @@ Connection.prototype.handleMessage = function(message) {
       if (doc) doc._handleOp(err, message);
       return;
 
+    case 'p':
+      var doc = this.getExisting(message.c, message.d);
+      if (doc) doc._handlePresence(err, message);
+      return;
+
     default:
       logger.warn('Ignoring unrecognized message', message);
   }
@@ -424,6 +429,23 @@ Connection.prototype.sendOp = function(doc, op) {
   this.send(message);
 };
 
+Connection.prototype.sendPresence = function(doc, data, requestReply) {
+  // Ensure the doc is registered so that it receives the reply message
+  this._addDoc(doc);
+  var message = {
+    a: 'p',
+    c: doc.collection,
+    d: doc.id,
+    p: data,
+    v: doc.version || 0,
+    seq: this.seq++
+  };
+  if (requestReply) {
+    message.r = true;
+  }
+  this.send(message);
+};
+
 
 /**
  * Sends a message down the socket