Merge pull request #546 from CartoDB/2.11-ensure-connect-callback

Ensure connect callback is invoked on premature socket hangup

lib/client.js

@@ -170,9 +170,17 @@ Client.prototype.connect = function(callback) {
       return self.emit('error', error);
     }
     callback(error);
+    callback = null;
   });
 
   con.once('end', function() {
+    if ( callback ) {
+      // haven't received a connection message yet !
+      var err = new Error("Stream unexpectedly ended before getting ready for query execution");
+      callback(err);
+      callback = null;
+      return;
+    }
     if(self.activeQuery) {
       var disconnectError = new Error('Stream unexpectedly ended during query execution');
       self.activeQuery.handleError(disconnectError, con);

lib/connection.js

@@ -61,7 +61,10 @@ Connection.prototype.connect = function(port, host) {
     self.emit('error', error);
   });
 
-  this.stream.on('end', function() {
+  this.stream.on('close', function() {
+    // NOTE: node-0.10 emits both 'end' and 'close'
+    //       for streams closed by the peer, while
+    //       node-0.8 only emits 'close'
     self.emit('end');
   });
 

test/unit/client/early-disconnect-tests.js

@@ -0,0 +1,22 @@
+var helper = require(__dirname + '/test-helper');
+var net = require('net');
+var pg = require('../../..//lib/index.js');
+ 
+var server = net.createServer(function(c) {
+  console.log('server connected');
+  c.destroy();
+  console.log('server socket destroyed.');
+  server.close(function() { console.log('server closed'); });
+});
+
+server.listen(7777, function() {
+  console.log('server listening');
+  var client = new pg.Client('postgres://localhost:7777');
+  console.log('client connecting');
+  client.connect(assert.calls(function(err) {
+    if (err) console.log("Error on connect: "+err);
+    else console.log('client connected');
+    assert(err);
+  }));
+   
+});

test/unit/client/stream-and-query-error-interaction-tests.js

@@ -20,7 +20,7 @@ test('emits end when not in query', function() {
     assert.equal(client.queryQueue.length, 0);
     assert(client.activeQuery, 'client should have issued query');
     process.nextTick(function() {
-      stream.emit('end');
+      stream.emit('close');
     });
   });
 });