Make query callback exceptions not break client

If you throw an exception in a query callback the client will not pulse
its internal query queue and therefor will never process any
more queries or emit its own 'drain' event.

I don't find this to be an issue in production code since I restart
the process on exceptions, but it can break tests and cause things
to 'hang'.  My crude benchmarks show no noticable impact in perf
from the try/catch/rethrow.

:q

Author: brianc

lib/client.js

@@ -143,12 +143,22 @@ Client.prototype.connect = function(callback) {
   });
 
   con.on('readyForQuery', function() {
+    var error;
     if(self.activeQuery) {
-      self.activeQuery.handleReadyForQuery();
+      //try/catch/rethrow to ensure exceptions don't prevent the queryQueue from
+      //being processed
+      try{
+        self.activeQuery.handleReadyForQuery();
+      } catch(e) {
+        error = e;
+      }
     }
     self.activeQuery = null;
     self.readyForQuery = true;
     self._pulseQueryQueue();
+    if(error) {
+      throw error;
+    }
   });
 
   con.on('error', function(error) {

lib/native/index.js

@@ -208,15 +208,23 @@ var clientBuilder = function(config) {
   });
 
   connection.on('_readyForQuery', function() {
+    var error;
     var q = this._activeQuery;
     //a named query finished being prepared
     if(this._namedQuery) {
       this._namedQuery = false;
       this._sendQueryPrepared(q.name, q.values||[]);
     } else {
-      connection._activeQuery.handleReadyForQuery(connection._lastMeta);
+      //try/catch/rethrow to ensure exceptions don't prevent the queryQueue from
+      //being processed
+      try{
+        connection._activeQuery.handleReadyForQuery(connection._lastMeta);
+      } catch(e) {
+        error = e;
+      }
       connection._activeQuery = null;
       connection._pulseQueryQueue();
+      if(error) throw error;
     }
   });
   connection.on('copyInResponse', function () {

test/integration/client/query-callback-error-tests.js

@@ -0,0 +1,22 @@
+var helper = require(__dirname + '/test-helper');
+var util = require('util');
+
+test('error during query execution', function() {
+  var client = new Client(helper.args);
+  process.removeAllListeners('uncaughtException');
+  assert.emits(process, 'uncaughtException', function() {
+    assert.equal(client.activeQuery, null, 'should remove active query even if error happens in callback');
+    client.query('SELECT * FROM blah', assert.success(function(result) {
+      assert.equal(result.rows.length, 1);
+      client.end();
+    }));
+  });
+  client.connect(assert.success(function() {
+    client.query('CREATE TEMP TABLE "blah"(data text)', assert.success(function() {
+      var q = client.query('INSERT INTO blah(data) VALUES($1)', ['yo'], assert.success(function() {
+        assert.emits(client, 'drain');
+        throw new Error('WHOOOAAAHH!!');
+      }));
+    }));
+  }));
+});