index.js

var inherits     = require('util').inherits
var EventEmitter = require('events').EventEmitter
var GenericPool  = require('generic-pool').Pool
var chain        = require('./lib/chain')

module.exports = ConnectionPool

inherits(ConnectionPool, EventEmitter)

function ConnectionPool(adapter, connParams, options) {
  if (!(this instanceof ConnectionPool)) {
    return new ConnectionPool(adapter, connParams, options)
  }
  EventEmitter.call(this)

  options    = options    || {}
  connParams = connParams || {}
  if (options.create) {
    console.warn("PoolConfig.create ignored, use PoolConfig.onConnect instead")
  }
  if (options.destroy) {
    console.warn("PoolConfig.destroy ignored")
  }

  if (adapter.name == 'sqlite3'
      && /:memory:$/i.test(connParams.database)
      && (options.min > 1 || options.max > 1))
  {
    console.warn(
      "Pools of more than 1 connection do not work for in-memory SQLite3 databases\n" +
      "The specified minimum (%d) and maximum (%d) connections have been overridden",
      options.min, options.max
    )
    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);

        onConnect(conn, function (err, connection) {
          if (err) return ready(err);
          conn.on('error', self._handleIdleError)
          ready(null, connection)
        })
      })
    },

    destroy: function (connection) {
      connection.removeAllListeners()
      connection.on('error', function () {})
      connection.end()
    },

    log: options.log,

    idleTimeoutMillis: options.idleTimeout,
    reapIntervalMillis: options.reapInterval,
  }

  if (options.hasOwnProperty('refreshIdle')) {
    poolOpts.refreshIdle = options.refreshIdle
  }

  this._pool = new GenericPool(poolOpts)
  this._cancelledQueries = []

  var resetSteps = []
  if (adapter.reset) resetSteps.unshift(adapter.reset)
  if (options.reset) resetSteps.unshift(options.reset)
  this.adapter = adapter
  this._reset = chain(resetSteps)

  this._shouldDestroyConnection = function (err) {
    if (err instanceof CancelledQueryError) {
      return false
    }
    return options.shouldDestroyConnection ? options.shouldDestroyConnection(err) : 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)
    , connection = null

  if (query.callback) {
    callback = query.callback
    query.callback = function (err, result) {
      self._maybeDestroy(connection, err)
      callback(err, result)
    }
  } 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) {
        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.
   */
  function handleConnectionError (error) {
    self._maybeDestroy(connection, error)
    forwardError(error)
  }

  function forwardError (error) {
    if (query.callback) {
      query.callback(error)
    } else {
      query.emit('error', error)
    }
  }

  this.acquire(function (err, connection_) {
    if (err) {
      return handleConnectionError(err)
    }
    err = self._checkCancellation(query)
    if (err) {
      return forwardError(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.cancel = function (query) {
  this._cancelledQueries.push(query)
}

ConnectionPool.prototype.acquire = function (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) {
  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)
  })
}

ConnectionPool.prototype.destroy = function (connection) {
  this._pool.destroy(connection)
}

ConnectionPool.prototype.close = function (callback) {
  var self = this
  this._pool.drain(function () {
    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._checkCancellation = function (query) {
  var cancelIndex = this._cancelledQueries.indexOf(query)
  if (cancelIndex >= 0) {
    this._cancelledQueries.splice(cancelIndex, 1)
    return new CancelledQueryError()
  }
}

inherits(CancelledQueryError, Error)
function CancelledQueryError() {
  Error.call(this)
  this.message = "Query was cancelled before connection was acquired"
  this.name = "CancelledQueryError"
}