Add option to time out pg.connect() call

Makefile

@@ -4,10 +4,10 @@ connectionString=postgres://
 
 params := $(connectionString)
 
-node-command := xargs -n 1 -I file node file $(params)
+node-command := TZ=GMT xargs -n 1 -I file node file $(params)
 
 .PHONY : test test-connection test-integration bench test-native \
-	 jshint publish test-missing-native update-npm
+	 jshint publish test-missing-native update-npm clean
 
 all:
 	npm install
@@ -67,3 +67,6 @@ prepare-test-db:
 jshint:
 	@echo "***Starting jshint***"
 	@./node_modules/.bin/jshint lib
+
+clean:
+	@rm -rf node_modules

lib/defaults.js

@@ -30,6 +30,17 @@ var defaults = module.exports = {
   //from the pool and destroyed
   poolIdleTimeout: 30000,
 
+  // acquireTimeout is the amount of time in ms to wait to check out a database
+  // connection. By default, node-postgres will wait indefinitely for a
+  // connection to become available. If the acquireTimeout is negative,
+  // or NaN, pg.connect will return an error immediately. If the acquireTimeout
+  // is zero, node-postgres will return an error in the next tick if the pool
+  // is full.
+  //
+  // The error message on timeout will be "Cannot acquire resource because the
+  // pool is full".
+  acquireTimeout: undefined,
+
   //frequency to check for idle clients within the client pool
   reapIntervalMillis: 1000,
 

lib/pool.js

@@ -1,7 +1,7 @@
 var EventEmitter = require('events').EventEmitter;
 
 var defaults = require('./defaults');
-var genericPool = require('generic-pool');
+var genericPool = require('generic-pool-timeout');
 
 
 module.exports = function(Client) {
@@ -72,6 +72,17 @@ module.exports = function(Client) {
           pool[key] = EventEmitter.prototype[key];
         }
       }
+      var block = true;
+      if (defaults.block === false || clientConfig.block === false) {
+        block = false;
+      }
+      var timeout;
+      if (typeof defaults.acquireTimeout === "number" && !isNaN(defaults.acquireTimeout)) {
+        timeout = defaults.acquireTimeout;
+      } else if (typeof clientConfig.acquireTimeout === "number" && !isNaN(clientConfig.acquireTimeout)) {
+        timeout = clientConfig.acquireTimeout;
+      }
+
       //monkey-patch with connect method
       pool.connect = function(cb) {
         var domain = process.domain;
@@ -88,10 +99,10 @@ module.exports = function(Client) {
               pool.release(client);
             }
           });
-        });
+        }, 0, {timeout: timeout});
       };
       return pool;
-    }
+    },
   };
 
   return pools;

package.json

@@ -19,7 +19,7 @@
   "main": "./lib",
   "dependencies": {
     "buffer-writer": "1.0.1",
-    "generic-pool": "2.1.1",
+    "generic-pool-timeout": "2.7.1",
     "packet-reader": "0.2.0",
     "pg-connection-string": "0.1.3",
     "pg-types": "1.*",

script/test-connection.js

@@ -9,7 +9,7 @@ pg.connect(helper.config, function(err, client, done) {
     console.error(err);
     process.exit(255);
   }
-  console.log("Checking for existance of required test table 'person'")
+  console.log("Checking for existence of required test table 'person'")
   client.query("SELECT COUNT(name) FROM person", function(err, callback) {
     if(err != null) {
       console.error("Recieved error when executing query 'SELECT COUNT(name) FROM person'")

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

@@ -17,7 +17,7 @@ pg.connect(helper.config, assert.success(function(client, done) {
         var params = ['idle'];
         if(!isGreater) {
           killIdleQuery = 'SELECT procpid, (SELECT pg_terminate_backend(procpid)) AS killed FROM pg_stat_activity WHERE current_query LIKE $1';
-          params = ['%IDLE%']
+          params = ['%IDLE%'];
         }
 
         //subscribe to the pg error event
@@ -37,5 +37,5 @@ pg.connect(helper.config, assert.success(function(client, done) {
       }));
     }));
 
-  })
+  });
 }));

test/test-helper.js

@@ -3,7 +3,7 @@ assert = require('assert');
 
 var EventEmitter = require('events').EventEmitter;
 var sys = require('util');
-var BufferList = require(__dirname+'/buffer-list')
+var BufferList = require(__dirname+'/buffer-list');
 
 var Connection = require(__dirname + '/../lib/connection');
 
@@ -160,7 +160,7 @@ var expect = function(callback, timeout) {
       callback.apply(this, arguments)
     }
   } else {
-    throw new Error("Unsupported arrity " + callback.length);
+    throw new Error("Unsupported arity " + callback.length);
   }
 
 }
@@ -193,7 +193,7 @@ process.on('exit', function() {
 })
 
 process.on('uncaughtException', function(err) {
-  console.error("\n %s", err.stack || err.toString())
+  console.error("\n %s", err.stack || err.toString());
   //causes xargs to abort right away
   process.exit(255);
 });
@@ -248,5 +248,3 @@ module.exports = {
   setTimezoneOffset: setTimezoneOffset,
   resetTimezoneOffset: resetTimezoneOffset
 };
-
-

test/unit/client/query-queue-tests.js

@@ -46,7 +46,7 @@ test('drain', function() {
     test("emits drain", function() {
       process.nextTick(function() {
         assert.ok(raisedDrain);
-      })
+      });
     });
   });
 });

test/unit/pool/basic-tests.js

@@ -5,29 +5,30 @@ var libDir = __dirname + '/../../../lib';
 var poolsFactory = require(libDir + '/pool')
 var defaults = require(libDir + '/defaults');
 var poolId = 0;
+var pg = require(libDir);
 
 require(__dirname + '/../../test-helper');
 
 var FakeClient = function() {
   EventEmitter.call(this);
-}
+};
 
 util.inherits(FakeClient, EventEmitter);
 
 FakeClient.prototype.connect = function(cb) {
   process.nextTick(cb);
-}
+};
 
 FakeClient.prototype.end = function() {
   this.endCalled = true;
-}
+};
 var pools = poolsFactory(FakeClient);
 
 //Hangs the event loop until 'end' is called on client
 var HangingClient = function(config) {
   EventEmitter.call(this);
   this.config = config;
-}
+};
 
 util.inherits(HangingClient, EventEmitter);
 
@@ -36,11 +37,11 @@ HangingClient.prototype.connect = function(cb) {
     console.log('hung client...');
   }, 1000);
   process.nextTick(cb);
-}
+};
 
 HangingClient.prototype.end = function() {
   clearInterval(this.intervalId);
-}
+};
 
 test('no pools exist', function() {
   assert.empty(Object.keys(pools.all));
@@ -153,7 +154,7 @@ test('pool with connection error on connection', function() {
   });
 });
 
-test('returnning an error to done()', function() {
+test('returning an error to done()', function() {
   var p = pools.getOrCreate(poolId++);
   p.connect(function(err, client, done) {
     assert.equal(err, null);
@@ -178,7 +179,6 @@ test('fetching pool by object', function() {
   assert.equal(p, p2);
 });
 
-
 test('pool#connect client.poolCount', function() {
   var p = pools.getOrCreate(poolId++);
   var tid;
@@ -187,7 +187,7 @@ test('pool#connect client.poolCount', function() {
     tid = setTimeout(function() {
       throw new Error("Connection callback was never called");
     }, 100);
-  }
+  };
 
   setConnectTimeout();
   p.connect(function(err, client, done) {
@@ -209,6 +209,54 @@ test('pool#connect client.poolCount', function() {
         assert.equal(client.poolCount, undefined,
           'after pool is destroyed, count should be undefined');
       });
-    })
+    });
+  });
+});
+
+pg.defaults.poolSize = 1;
+
+test('pool#connect acquire errors if the pool is full', function() {
+  var p = pools.getOrCreate(poolId++);
+  p.connect(function(err, client, done1) {
+    p.connect(function(err, client, done2) {
+      assert.equal(err.message, "Cannot acquire resource because the pool is full");
+    });
+  });
+});
+
+pg.defaults.poolSize = 1;
+
+pg.defaults.poolSize = 1;
+pg.defaults.acquireTimeout = 10;
+pg.defaults.poolIdleTimeout = 10;
+
+test('pool#connect acquire errors if acquisition times out', function() {
+  var p = pools.getOrCreate(poolId++);
+  p.connect(function(err, client, done1) {
+    setTimeout(function() {
+      done1();
+    }, 40);
+
+    var start = Date.now();
+    p.connect(function(err, client, done2) {
+      assert.equal(err.message, "Cannot acquire resource because the pool is full");
+      assert.ok((Date.now() - start) < 15);
+    });
+  });
+});
+
+pg.defaults.acquireTimeout = 20;
+
+test('pool#connect acquire returns a resource if one becomes available before the timeout', function() {
+  var p = pools.getOrCreate(poolId++);
+  p.connect(function(err, client, done1) {
+    setTimeout(function() {
+      done1();
+    }, 10);
+
+    p.connect(function(err, client, done2) {
+      assert.equal(err, null);
+      done2();
+    });
   });
 });

test/unit/pool/timeout-tests.js

@@ -10,17 +10,17 @@ require(__dirname + '/../../test-helper');
 
 var FakeClient = function() {
   EventEmitter.call(this);
-}
+};
 
 util.inherits(FakeClient, EventEmitter);
 
 FakeClient.prototype.connect = function(cb) {
   process.nextTick(cb);
-}
+};
 
 FakeClient.prototype.end = function() {
   this.endCalled = true;
-}
+};
 
 defaults.poolIdleTimeout = 10;
 defaults.reapIntervalMillis = 10;