Merge a2f871a into 1038607

lib/client/connection.js

@@ -705,9 +705,11 @@ Connection.prototype._handleSnapshotFetch = function(error, message) {
 };
 
 Connection.prototype._handleLegacyInit = function(message) {
-  // If the minor protocol version has been set, we can ignore this legacy
-  // init message, and wait for a response to our handshake message.
-  if (message.protocolMinor) return;
+  // If the minor protocol version has been set, we want to use the
+  // new handshake protocol. Let's send a handshake initialize, because
+  // we now know the server is ready. If we've already sent it, we'll
+  // just ignore the response anyway.
+  if (message.protocolMinor) return this._initializeHandshake();
   this._initialize(message);
 };
 
@@ -721,6 +723,8 @@ Connection.prototype._handleHandshake = function(error, message) {
 };
 
 Connection.prototype._initialize = function(message) {
+  if (this.state !== 'connecting') return;
+
   if (message.protocol !== 1) {
     return this.emit('error', new ShareDBError(
       ERROR_CODE.ERR_PROTOCOL_VERSION_NOT_SUPPORTED,

test/client/connection.js

@@ -36,11 +36,15 @@ describe('client connection', function() {
     connection.close();
   });
 
-  it('ends the agent steam on call to agent.close()', function(done) {
+  it('ends the agent stream on call to agent.close()', function(done) {
+    var isDone = false;
+    var finish = function() {
+      if (!isDone) done();
+    };
+
     this.backend.use('connect', function(request, next) {
-      request.agent.stream.on('end', function() {
-        done();
-      });
+      request.agent.stream.on('close', finish);
+      request.agent.stream.on('end', finish);
       request.agent.close();
       next();
     });
@@ -88,6 +92,38 @@ describe('client connection', function() {
     connection.socket.onerror({message: 'Test'});
   });
 
+  it('connects to a Backend that binds its socket late', function(done) {
+    var backend = this.backend;
+    var socket = new StreamSocket();
+    var connection = new Connection(socket);
+    socket._open();
+    var doc = connection.get('test', '123');
+    doc.fetch(done);
+
+    socket.stream.on('data', function() {
+      // Registering a stream triggers data to get flushed in our tests.
+      // In production, aw web socket might lose messages any time between
+      // connection and calling backend.listen()
+    });
+
+    process.nextTick(function() {
+      backend.listen(socket.stream);
+    });
+  });
+
+  it('connects when binding the Connection late', function(done) {
+    var backend = this.backend;
+    var socket = new StreamSocket();
+    socket._open();
+    backend.listen(socket.stream);
+
+    process.nextTick(function() {
+      var connection = new Connection(socket);
+      var doc = connection.get('test', '123');
+      doc.fetch(done);
+    });
+  });
+
   describe('backend.agentsCount', function() {
     it('updates after connect and connection.close()', function(done) {
       var backend = this.backend;