pg object emit error event on idle pooled client errors

Author: brianc

lib/index.js

@@ -1,4 +1,5 @@
 var EventEmitter = require('events').EventEmitter;
+var sys = require('sys');
 var Client = require(__dirname+'/client');
 var defaults =  require(__dirname + '/defaults');
 var genericPool = require('generic-pool');
@@ -7,7 +8,8 @@ var genericPool = require('generic-pool');
 var pools = {};
 
 //returns connect function using supplied client constructor
-var makeConnectFunction = function(ClientConstructor) {
+var makeConnectFunction = function(pg) {
+  var ClientConstructor = pg.Client;
   return function(config, callback) {
     var c = config;
     var cb = callback;
@@ -31,6 +33,13 @@ var makeConnectFunction = function(ClientConstructor) {
         };
         var connectSuccess = function() {
           client.removeListener('error', connectError);
+
+          //handle connected client background errors by emitting event
+          //via the pg object and then removing errored client from the pool
+          client.on('error', function(e) {
+            pg.emit('error', e, client);
+            pool.destroy(client);
+          });
           callback(null, client);
         };
         client.once('connect', connectSuccess);
@@ -58,25 +67,25 @@ var end = function() {
   })
 };
 
-module.exports = {
-  Client: Client,
-  Connection: require(__dirname + '/connection'),
-  connect: makeConnectFunction(Client),
-  end: end,
-  defaults: defaults
-}
+var PG = function(clientConstructor) {
+  EventEmitter.call(this);
+  this.Client = clientConstructor;
+  this.Connection = require(__dirname + '/connection');
+  this.connect = makeConnectFunction(this);
+  this.end = end;
+  this.defaults = defaults;
+};
+
+sys.inherits(PG, EventEmitter);
+
+module.exports = new PG(Client);
 
 var nativeExport = null;
 //lazy require native module...the c++ may not have been compiled
 module.exports.__defineGetter__("native", function() {
   if(nativeExport === null) {
     var NativeClient = require(__dirname + '/native');
-    nativeExport = {
-      Client: NativeClient,
-      connect: makeConnectFunction(NativeClient),
-      end: end,
-      defaults: defaults
-    }
+    nativeExport = new PG(NativeClient);
   }
   return nativeExport;
 })

test/integration/connection-pool/error-tests.js

@@ -0,0 +1,30 @@
+var helper = require(__dirname + "/../test-helper");
+var pg = require(__dirname + "/../../../lib");
+helper.pg = pg;
+
+var conString = helper.connectionString();
+
+//first make pool hold 2 clients
+pg.defaults.poolSize = 2;
+
+var killIdleQuery = 'SELECT procpid, (SELECT pg_terminate_backend(procpid)) AS killed FROM pg_stat_activity WHERE current_query LIKE \'<IDLE>\'';
+
+//get first client
+pg.connect(conString, assert.success(function(client) {
+  client.id = 1;
+  pg.connect(conString, assert.success(function(client2) {
+    client2.id = 2;
+    //subscribe to the pg error event
+    assert.emits(pg, 'error', function(error, brokenClient) {
+      assert.ok(error);
+      assert.ok(client);
+      assert.equal(client.id, 1);
+      pg.end();
+    });
+    //kill the connection from client
+    client2.query(killIdleQuery, assert.success(function(res) {
+      //check to make sure client connection actually was killed
+      assert.length(res.rows, 1);
+    }));
+  }));
+}));