'use strict' const EventEmitter = require('events').EventEmitter const NOOP = function () {} const removeWhere = (list, predicate) => { const i = list.findIndex(predicate) return i === -1 ? undefined : list.splice(i, 1)[0] } class IdleItem { constructor(client, idleListener, timeoutId) { this.client = client this.idleListener = idleListener this.timeoutId = timeoutId } } class PendingItem { constructor(callback) { this.callback = callback } } function throwOnDoubleRelease() { throw new Error('Release called on client which has already been released to the pool.') } function promisify(Promise, callback) { if (callback) { return { callback: callback, result: undefined } } let rej let res const cb = function (err, client) { err ? rej(err) : res(client) } const result = new Promise(function (resolve, reject) { res = resolve rej = reject }) return { callback: cb, result: result } } function makeIdleListener(pool, client) { return function idleListener(err) { err.client = client client.removeListener('error', idleListener) client.on('error', () => { pool.log('additional client error after disconnection due to error', err) }) pool._remove(client) // TODO - document that once the pool emits an error // the client has already been closed & purged and is unusable pool.emit('error', err, client) } } class Pool extends EventEmitter { constructor(options, Client) { super() this.options = Object.assign({}, options) if (options != null && 'password' in options) { // "hiding" the password so it doesn't show up in stack traces // or if the client is console.logged Object.defineProperty(this.options, 'password', { configurable: true, enumerable: false, writable: true, value: options.password, }) } this.options.max = this.options.max || this.options.poolSize || 10 this.options.maxUses = this.options.maxUses || Infinity this.log = this.options.log || function () {} this.Client = this.options.Client || Client || require('pg').Client this.Promise = this.options.Promise || global.Promise if (typeof this.options.idleTimeoutMillis === 'undefined') { this.options.idleTimeoutMillis = 10000 } this._clients = [] this._idle = [] this._pendingQueue = [] this._endCallback = undefined this.ending = false this.ended = false } _isFull() { return this._clients.length >= this.options.max } _pulseQueue() { this.log('pulse queue') if (this.ended) { this.log('pulse queue ended') return } if (this.ending) { this.log('pulse queue on ending') if (this._idle.length) { this._idle.slice().map((item) => { this._remove(item.client) }) } if (!this._clients.length) { this.ended = true this._endCallback() } return } // if we don't have any waiting, do nothing if (!this._pendingQueue.length) { this.log('no queued requests') return } // if we don't have any idle clients and we have no more room do nothing if (!this._idle.length && this._isFull()) { return } const pendingItem = this._pendingQueue.shift() if (this._idle.length) { const idleItem = this._idle.pop() clearTimeout(idleItem.timeoutId) const client = idleItem.client const idleListener = idleItem.idleListener return this._acquireClient(client, pendingItem, idleListener, false) } if (!this._isFull()) { return this.newClient(pendingItem) } throw new Error('unexpected condition') } _remove(client) { const removed = removeWhere(this._idle, (item) => item.client === client) if (removed !== undefined) { clearTimeout(removed.timeoutId) } this._clients = this._clients.filter((c) => c !== client) client.end() this.emit('remove', client) } connect(cb) { if (this.ending) { const err = new Error('Cannot use a pool after calling end on the pool') return cb ? cb(err) : this.Promise.reject(err) } const response = promisify(this.Promise, cb) const result = response.result // if we don't have to connect a new client, don't do so if (this._clients.length >= this.options.max || this._idle.length) { // if we have idle clients schedule a pulse immediately if (this._idle.length) { process.nextTick(() => this._pulseQueue()) } if (!this.options.connectionTimeoutMillis) { this._pendingQueue.push(new PendingItem(response.callback)) return result } const queueCallback = (err, res, done) => { clearTimeout(tid) response.callback(err, res, done) } const pendingItem = new PendingItem(queueCallback) // set connection timeout on checking out an existing client const tid = setTimeout(() => { // remove the callback from pending waiters because // we're going to call it with a timeout error removeWhere(this._pendingQueue, (i) => i.callback === queueCallback) pendingItem.timedOut = true response.callback(new Error('timeout exceeded when trying to connect')) }, this.options.connectionTimeoutMillis) this._pendingQueue.push(pendingItem) return result } this.newClient(new PendingItem(response.callback)) return result } newClient(pendingItem) { const client = new this.Client(this.options) this._clients.push(client) const idleListener = makeIdleListener(this, client) this.log('checking client timeout') // connection timeout logic let tid let timeoutHit = false if (this.options.connectionTimeoutMillis) { tid = setTimeout(() => { this.log('ending client due to timeout') timeoutHit = true // force kill the node driver, and let libpq do its teardown client.connection ? client.connection.stream.destroy() : client.end() }, this.options.connectionTimeoutMillis) } this.log('connecting new client') client.connect((err) => { if (tid) { clearTimeout(tid) } client.on('error', idleListener) if (err) { this.log('client failed to connect', err) // remove the dead client from our list of clients this._clients = this._clients.filter((c) => c !== client) if (timeoutHit) { err.message = 'Connection terminated due to connection timeout' } // this client won’t be released, so move on immediately this._pulseQueue() if (!pendingItem.timedOut) { pendingItem.callback(err, undefined, NOOP) } } else { this.log('new client connected') return this._acquireClient(client, pendingItem, idleListener, true) } }) } // acquire a client for a pending work item _acquireClient(client, pendingItem, idleListener, isNew) { if (isNew) { this.emit('connect', client) } this.emit('acquire', client) client.release = this._releaseOnce(client, idleListener) client.removeListener('error', idleListener) if (!pendingItem.timedOut) { if (isNew && this.options.verify) { this.options.verify(client, (err) => { if (err) { client.release(err) return pendingItem.callback(err, undefined, NOOP) } pendingItem.callback(undefined, client, client.release) }) } else { pendingItem.callback(undefined, client, client.release) } } else { if (isNew && this.options.verify) { this.options.verify(client, client.release) } else { client.release() } } } // returns a function that wraps _release and throws if called more than once _releaseOnce(client, idleListener) { let released = false return (err) => { if (released) { throwOnDoubleRelease() } released = true this._release(client, idleListener, err) } } // release a client back to the poll, include an error // to remove it from the pool _release(client, idleListener, err) { client.on('error', idleListener) client._poolUseCount = (client._poolUseCount || 0) + 1 // TODO(bmc): expose a proper, public interface _queryable and _ending if (err || this.ending || !client._queryable || client._ending || client._poolUseCount >= this.options.maxUses) { if (client._poolUseCount >= this.options.maxUses) { this.log('remove expended client') } this._remove(client) this._pulseQueue() return } // idle timeout let tid if (this.options.idleTimeoutMillis) { tid = setTimeout(() => { this.log('remove idle client') this._remove(client) }, this.options.idleTimeoutMillis) } this._idle.push(new IdleItem(client, idleListener, tid)) this._pulseQueue() } query(text, values, cb) { // guard clause against passing a function as the first parameter if (typeof text === 'function') { const response = promisify(this.Promise, text) setImmediate(function () { return response.callback(new Error('Passing a function as the first parameter to pool.query is not supported')) }) return response.result } // allow plain text query without values if (typeof values === 'function') { cb = values values = undefined } const response = promisify(this.Promise, cb) cb = response.callback this.connect((err, client) => { if (err) { return cb(err) } let clientReleased = false const onError = (err) => { if (clientReleased) { return } clientReleased = true client.release(err) cb(err) } client.once('error', onError) this.log('dispatching query') client.query(text, values, (err, res) => { this.log('query dispatched') client.removeListener('error', onError) if (clientReleased) { return } clientReleased = true client.release(err) if (err) { return cb(err) } else { return cb(undefined, res) } }) }) return response.result } end(cb) { this.log('ending') if (this.ending) { const err = new Error('Called end on pool more than once') return cb ? cb(err) : this.Promise.reject(err) } this.ending = true const promised = promisify(this.Promise, cb) this._endCallback = promised.callback this._pulseQueue() return promised.result } get waitingCount() { return this._pendingQueue.length } get idleCount() { return this._idle.length } get totalCount() { return this._clients.length } } module.exports = Pool