back port connection leak fix from 2.0

Author: jshepard

index.js

@@ -1,8 +1,7 @@
 var inherits     = require('util').inherits
 var EventEmitter = require('events').EventEmitter
 var Transaction  = require('any-db-transaction')
-var Pool         = require('generic-pool').Pool
-var once         = require('once')
+var GenericPool  = require('generic-pool').Pool
 var chain        = require('./lib/chain')
 
 module.exports = ConnectionPool
@@ -17,12 +16,14 @@ function ConnectionPool(adapter, connParams, options) {
 
   options    = options    || {}
   connParams = connParams || {}
-
-  if (options.create || options.destroy) {
-    throw new Error("Use onConnect/reset options instead of create/destroy.")
+  if (options.create) {
+    console.warn("PoolConfig.create ignored, use PoolConfig.onConnect instead")
+  }
+  if (options.destroy) {
+    console.warn("PoolConfig.destroy ignored")
   }
 
-  if (connParams.adapter == 'sqlite3'
+  if (adapter.name == 'sqlite3'
       && /:memory:$/i.test(connParams.database)
       && (options.min > 1 || options.max > 1))
   {
@@ -34,74 +35,142 @@ function ConnectionPool(adapter, connParams, options) {
     if (options.min) options.min = 1
     options.max = 1
   }
-  
+
+  var onConnect = options.onConnect || function (c, done) { done(null, c) }
+
   var poolOpts = {
     min: options.min || 0,
     max: options.max || 10,
     create: function (ready) {
       adapter.createConnection(connParams, function (err, conn) {
         if (err) return ready(err);
-        else if (options.onConnect) options.onConnect(conn, ready)
-        else ready(null, conn)
+
+        onConnect(conn, function (err, connection) {
+          if (err) return ready(err);
+          conn.on('error', self._handleIdleError)
+          ready(null, connection)
+        })
       })
     },
-    destroy: function (conn) {
-      conn.end()
-      conn._events = {}
+
+    destroy: function (connection) {
+      connection.removeAllListeners()
+      connection.on('error', function () {})
+      connection.end()
     },
 
-    log: options.log
+    log: options.log,
+
+    idleTimeoutMillis: options.idleTimeout,
+    reapIntervalMillis: options.reapInterval,
+  }
+
+  if (options.hasOwnProperty('refreshIdle')) {
+    poolOpts.refreshIdle = options.refreshIdle
   }
 
-  var resetSteps = [];
+  this._pool = new GenericPool(poolOpts)
+
+  var resetSteps = []
   if (adapter.reset) resetSteps.unshift(adapter.reset)
   if (options.reset) resetSteps.unshift(options.reset)
-  this.adapter = adapter.name
-  this._adapter = adapter
+  this.adapter = adapter
   this._reset = chain(resetSteps)
-  this._pool = new Pool(poolOpts)
+
+  this._shouldDestroyConnection = options.shouldDestroyConnection || function (err) { return true }
+
+  var self = this
+  self._handleIdleError = function (err) {
+    var connection = this
+    self._maybeDestroy(connection, err)
+    self.emit('error', err, connection)
+  }
 }
 
 ConnectionPool.prototype.query = function (statement, params, callback) {
   var self = this
-    , query = this._adapter.createQuery(statement, params, callback)
-
-  this.acquire(function (err, conn) {
-    if (err) {
-      if (typeof params === 'function') {
-        return params(err)
-      } else if (callback) {
-        return callback(err);
-      } else {
-        return query.emit('error', err);
-      }
+    , query = this.adapter.createQuery(statement, params, callback)
+    , connection = null
+
+  if (query.callback) {
+    callback = query.callback
+    query.callback = function (err, result) {
+      self._maybeDestroy(connection, err)
+      callback(err, result)
     }
-    conn.query(query);
-    self.emit('query', query)
-    var release = once(self.release.bind(self, conn))
-    query.once('end', release).once('error', function (err) {
-      release()
-      // If this was the only error listener, re-emit the error.
+  } else {
+    var finished = false
+    query.once('end', function () {
+      if (!finished) {
+        finished = true
+        self.release(connection)
+      }
+    })
+    query.once('error', function (err) {
+      if (!finished) {
+        finished = true
+        self._maybeDestroy(connection, err)
+      }
+      // If this was the only error listener, re-emit the error from the pool.
       if (!this.listeners('error').length) {
-        this.emit('error', err)
+        self.emit('error', err, query)
       }
     })
+  }
+
+  /**
+   * if a connection cannot be acquired, or emits an 'error' event while a
+   * query is in progress, the error should be handled by the query object.
+   */
+  var handleConnectionError = function (error) {
+    self._maybeDestroy(connection, error)
+    if (query.callback) {
+      query.callback(error)
+    } else {
+      query.emit('error', error)
+    }
+  }
+
+  this.acquire(function (err, connection_) {
+    if (err) {
+      return handleConnectionError(err)
+    }
+
+
+    // expose the connection to everything else in the outer scope
+    connection = connection_
+
+    // attach error event listener to the connection
+    connection.on('error', handleConnectionError)
+    query.once('end', function () {
+      connection.removeListener('error', handleConnectionError)
+    })
+
+    connection.query(query);
+    self.emit('query', query)
   })
 
   return query
 }
 
 ConnectionPool.prototype.acquire = function (callback) {
-  this.emit('acquire')
-  this._pool.acquire(callback);
+  var self = this
+  self._pool.acquire(function (err, connection) {
+    if (err) return callback(err);
+    connection.removeListener('error', self._handleIdleError)
+    self.emit('acquire', connection)
+    callback(null, connection)
+  });
 }
 
 ConnectionPool.prototype.release = function (connection) {
-  this.emit('release')
-  var pool = this._pool
-  this._reset(connection, function (err) {
-    if (err) return pool.destroy(connection)
-    pool.release(connection)
+  var self = this
+  self.emit('release', connection)
+  connection.removeAllListeners()
+  self._reset(connection, function (err) {
+    if (err) return self.destroy(connection);
+    connection.on('error', self._handleIdleError)
+    self._pool.release(connection)
   })
 }
 
@@ -112,14 +181,25 @@ ConnectionPool.prototype.destroy = function (connection) {
 ConnectionPool.prototype.close = function (callback) {
   var self = this
   this._pool.drain(function () {
-    self._pool.destroyAllNow()
-    self.emit('close')
-    if (callback) callback()
+    self._pool.destroyAllNow(function () {
+      self.emit('close')
+      if (callback) callback()
+    })
   })
 }
 
+ConnectionPool.prototype._maybeDestroy = function (connection, error) {
+  if (connection) {
+    if (error && this._shouldDestroyConnection(error)) {
+      this.destroy(connection)
+    } else {
+      this.release(connection)
+    }
+  }
+}
+
 ConnectionPool.prototype.begin = function (beginStatement, callback) {
-  var tx = Transaction.begin(this._adapter.createQuery, beginStatement, callback)
+  var tx = Transaction.begin(this.adapter.createQuery, beginStatement, callback)
 
   var pool = this
   this.acquire(function (err, connection) {