Sync ephemeral presence data

.editorconfig

@@ -0,0 +1,9 @@
+root = true
+
+[*]
+indent_style = space
+indent_size = 2
+end_of_line = LF
+charset = utf-8
+trim_trailing_whitespace = true
+insert_final_newline = true

.travis.yml

@@ -1,9 +1,9 @@
 language: node_js
 node_js:
-  - 6
-  - 5
-  - 4
-  - 0.10
+  - "9"
+  - "8"
+  - "6"
+  - "4"
 script: "npm run jshint && npm run test-cover"
 # Send coverage data to Coveralls
 after_script: "cat ./coverage/lcov.info | ./node_modules/coveralls/bin/coveralls.js"

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)
@@ -57,6 +58,10 @@ 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
+
+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
@@ -221,6 +226,9 @@ Unique document ID
 `doc.data` _(Object)_
 Document contents. Available after document is fetched or subscribed to.
 
+`doc.presence` _(Object)_
+Each property under `doc.presence` 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.
 
@@ -250,6 +258,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) {...})`
+Presence data has changed. `srcList` is an Array of `doc.presence` property names for which values have changed.
+
 `doc.on('error', function(err) {...})`
 There was an error fetching the document or applying an operation.
 
@@ -283,6 +294,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)_
@@ -358,6 +374,9 @@ Additional fields may be added to the error object for debugging context dependi
 * 4021 - Invalid client id
 * 4022 - Database adapter does not support queries
 * 4023 - Cannot project snapshots of this type
+* 4024 - OT Type does not support presence
+* 4025 - Not subscribed to document
+* 4026 - Presence data superseded
 
 ### 5000 - Internal error
 

lib/agent.js

@@ -1,6 +1,7 @@
 var hat = require('hat');
 var util = require('./util');
 var types = require('./types');
+var ShareDBError = require('./error');
 
 /**
  * Agent deserializes the wire protocol messages received from the stream and
@@ -25,6 +26,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;
@@ -98,10 +102,17 @@ Agent.prototype._subscribeToStream = function(collection, id, stream) {
       console.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) return;
@@ -268,6 +279,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';
   }
 };
 
@@ -300,6 +318,9 @@ Agent.prototype._handleMessage = function(request, callback) {
         var op = this._createOp(request);
         if (!op) return callback({code: 4000, message: 'Invalid op message'});
         return this._submit(request.c, request.d, op, 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'});
     }
@@ -582,3 +603,30 @@ Agent.prototype._createOp = function(request) {
     return new DeleteOp(src, request.seq, request.v, request.del);
   }
 };
+
+Agent.prototype._presence = function(presence, callback) {
+  if (presence.seq <= this.maxPresenceSeq) {
+    return callback(new ShareDBError(4026, 'Presence data superseded'));
+  }
+  this.maxPresenceSeq = presence.seq;
+  if (!this.subscribedDocs[presence.c] || !this.subscribedDocs[presence.c][presence.d]) {
+    return 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

@@ -515,6 +515,11 @@ Backend.prototype.getChannels = function(collection, id) {
   ];
 };
 
+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

@@ -243,6 +243,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:
       console.warn('Ignoring unrecognized message', message);
   }
@@ -408,6 +413,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