var EventEmitter = require('events').EventEmitter; var defaults = require('./defaults'); var genericPool = require('generic-pool'); module.exports = function(Client) { var pools = { Client: Client, //dictionary of all key:pool pairs all: {}, //reference to the client constructor - can override in tests or for require('pg').native getOrCreate: function(clientConfig) { clientConfig = clientConfig || {}; var name = JSON.stringify(clientConfig); var pool = pools.all[name]; if(pool) { return pool; } pool = genericPool.Pool({ name: name, max: clientConfig.poolSize || defaults.poolSize, idleTimeoutMillis: clientConfig.poolIdleTimeout || defaults.poolIdleTimeout, reapIntervalMillis: clientConfig.reapIntervalMillis || defaults.reapIntervalMillis, returnToHead: clientConfig.returnToHead || defaults.returnToHead, log: clientConfig.poolLog || defaults.poolLog, create: function(cb) { var client = new pools.Client(clientConfig); // Ignore errors on pooled clients until they are connected. client.on('error', Function.prototype); client.connect(function(err) { if(err) return cb(err, null); // Remove the noop error handler after a connection has been established. client.removeListener('error', Function.prototype); //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) { pool.emit('error', e, client); // If the client is already being destroyed, the error // occurred during stream ending. Do not attempt to destroy // the client again. if (!client._destroying) { pool.destroy(client); } }); // Remove connection from pool on disconnect client.on('end', function(e) { // Do not enter infinite loop between pool.destroy // and client 'end' event... if ( ! client._destroying ) { pool.destroy(client); } }); client.poolCount = 0; return cb(null, client); }); }, destroy: function(client) { client._destroying = true; client.poolCount = undefined; client.end(); } }); pools.all[name] = pool; //mixin EventEmitter to pool EventEmitter.call(pool); for(var key in EventEmitter.prototype) { if(EventEmitter.prototype.hasOwnProperty(key)) { pool[key] = EventEmitter.prototype[key]; } } //monkey-patch with connect method pool.connect = function(cb) { var domain = process.domain; pool.acquire(function(err, client) { if(domain) { cb = domain.bind(cb); } if(err) return cb(err, null, function() {/*NOOP*/}); client.poolCount++; cb(null, client, function(err) { if(err) { pool.destroy(client); } else { pool.release(client); } }); }); }; return pool; } }; return pools; };