Add Agent.src field

This change adds an `Agent.src` field, which will accept a
client-provided ID. This is kept separate from the Agent-generated
`clientId` so that the agent's ID does not change during the handshake
(which could confuse any consumers relying on a stable client ID in the
`connect` middleware).

Any internal references to `clientId` then give precedence to the
consumer-provided `src`. If it hasn't been set (ie we're dealing with a
legacy client), then we fall back to using `clientId`.

Author: Alec Gibson

lib/agent.js

@@ -20,6 +20,12 @@ function Agent(backend, stream) {
   this.stream = stream;
 
   this.clientId = hat();
+  // src is a client-configurable "id" which the client will set in its handshake,
+  // and attach to its ops. This should take precedence over clientId if set.
+  // Only legacy clients, or new clients connecting for the first time will use the
+  // Agent-provided clientId. Ideally we'll deprecate clientId in favour of src
+  // in the next breaking change.
+  this.src = null;
   this.connectTime = Date.now();
 
   // We need to track which documents are subscribed by the client. This is a
@@ -46,7 +52,7 @@ module.exports = Agent;
 // Close the agent with the client.
 Agent.prototype.close = function(err) {
   if (err) {
-    logger.warn('Agent closed due to error', this.clientId, err.stack || err);
+    logger.warn('Agent closed due to error', this._src(), err.stack || err);
   }
   if (this.closed) return;
   // This will end the writable stream and emit 'finish'
@@ -185,7 +191,7 @@ Agent.prototype._isOwnOp = function(collection, op) {
   // Detect ops from this client on the same projection. Since the client sent
   // these in, the submit reply will be sufficient and we can silently ignore
   // them in the streams for subscribed documents or queries
-  return (this.clientId === op.src) && (collection === (op.i || op.c));
+  return (this._src() === op.src) && (collection === (op.i || op.c));
 };
 
 Agent.prototype.send = function(message) {
@@ -328,7 +334,7 @@ Agent.prototype._handleMessage = function(request, callback) {
 
     switch (request.a) {
       case 'hs':
-        if (request.id) this.clientId = request.id;
+        if (request.id) this.src = request.id;
         return callback(null, this._initMessage('hs'));
       case 'qf':
         return this._queryFetch(request.id, request.c, request.q, getQueryOptions(request), callback);
@@ -350,7 +356,7 @@ Agent.prototype._handleMessage = function(request, callback) {
         return this._unsubscribe(request.c, request.d, callback);
       case 'op':
         // Normalize the properties submitted
-        var op = createClientOp(request, this.clientId);
+        var op = createClientOp(request, this._src());
         if (op.seq >= util.MAX_SAFE_INTEGER) {
           return callback(new ShareDBError(
             ERROR_CODE.ERR_CONNECTION_SEQ_INTEGER_OVERFLOW,
@@ -654,11 +660,15 @@ Agent.prototype._initMessage = function(action) {
     a: action,
     protocol: 1,
     protocolMinor: 1,
-    id: this.clientId,
+    id: this._src(),
     type: types.defaultType.uri
   };
 };
 
+Agent.prototype._src = function() {
+  return this.src || this.clientId;
+};
+
 
 function createClientOp(request, clientId) {
   // src can be provided if it is not the same as the current agent,

lib/client/connection.js

@@ -52,7 +52,7 @@ function Connection(socket) {
   // A unique message number for the given id
   this.seq = 1;
 
-  // Equals agent.clientId on the server
+  // Equals agent.src on the server
   this.id = null;
 
   // This direct reference from connection to agent is not used internal to