domain aware connection pool

lib/connection-parameters.js

@@ -1,3 +1,4 @@
+var url = require('url');
 var dns = require('dns');
 var path = require('path');
 
@@ -17,7 +18,6 @@ var val = function(key, config, envVar) {
     defaults[key];
 };
 
-var url = require('url');
 //parses a connection string
 var parse = function(str) {
   var config;

lib/pool.js

@@ -50,9 +50,27 @@ var pools = {
     }
     //monkey-patch with connect method
     pool.connect = function(cb) {
+      var domain = process.domain;
       pool.acquire(function(err, client) {
+        if(domain) {
+          cb = domain.bind(cb);
+          domain.add(client);
+          //native clients do not have a connection object
+          if(client.connection) {
+            domain.add(client.connection);
+            domain.add(client.connection.stream);
+          }
+        }
         if(err)  return cb(err, null, function() {/*NOOP*/});
         cb(null, client, function(err) {
+          if(domain) {
+            //native clients do not have a connection object
+            if(client.connection) {
+              domain.remove(client.connection.stream);
+              domain.remove(client.connection);
+            }
+            domain.remove(client);
+          }
           if(err) {
             pool.destroy(client);
           } else {

package.json

@@ -27,7 +27,8 @@
   },
   "devDependencies": {
     "jshint": "1.1.0",
-    "semver": "~1.1.4"
+    "semver": "~1.1.4",
+    "async": "0.2.10"
   },
   "scripts": {
     "test": "make test-travis connectionString=postgres://postgres@localhost:5432/postgres",

test/integration/domain-tests.js

@@ -0,0 +1,58 @@
+var helper = require('./test-helper')
+var async = require('async')
+
+var testWithoutDomain = function(cb) {
+  test('no domain', function() {
+    assert(!process.domain)
+    helper.pg.connect(helper.config, assert.success(function(client, done) {
+      assert(!process.domain)
+      done()
+      cb()
+    }))
+  })
+}
+
+var testWithDomain = function(cb) {
+  test('with domain', function() {
+    assert(!process.domain)
+    var domain = require('domain').create()
+    domain.run(function() {
+      var startingDomain = process.domain
+      assert(startingDomain)
+      helper.pg.connect(helper.config, assert.success(function(client, done) {
+        assert(process.domain, 'no domain exists in connect callback')
+        assert.equal(startingDomain, process.domain, 'domain was lost when checking out a client')
+        var query = client.query('SELECT NOW()', assert.success(function() {
+          assert(process.domain, 'no domain exists in query callback')
+          assert.equal(startingDomain, process.domain, 'domain was lost when checking out a client')
+          done(true)
+          process.domain.exit()
+          cb()
+        }))
+      }))
+    })
+  })
+}
+
+var testErrorWithDomain = function(cb) {
+  test('error on domain', function() {
+    var domain = require('domain').create()
+    domain.on('error', function() {
+      cb()
+    })
+    domain.run(function() {
+      helper.pg.connect(helper.config, assert.success(function(client, done) {
+        client.query('SELECT SLDKJFLSKDJF')
+        client.on('drain', done)
+      }))
+    })
+  })
+}
+
+async.series([
+  testWithoutDomain,
+  testWithDomain,
+  testErrorWithDomain
+], function() {
+  helper.pg.end()
+})