diff --git a/packages/pg/lib/client.js b/packages/pg/lib/client.js index 2b13c1de7..4291c7603 100644 --- a/packages/pg/lib/client.js +++ b/packages/pg/lib/client.js @@ -701,21 +701,15 @@ class Client extends EventEmitter { const readTimeout = config.query_timeout || this.connectionParameters.query_timeout if (readTimeout) { - const queryCallback = query.callback || (() => {}) - const readTimeoutTimer = setTimeout(() => { const error = new Error('Query read timeout') - process.nextTick(() => { query.handleError(error, this.connection) + // we already returned an error, + // just do nothing if query completes + query.callback = () => {} }) - queryCallback(error) - - // we already returned an error, - // just do nothing if query completes - query.callback = () => {} - // Remove from queue (only safe if not yet sent) const index = this._queryQueue.indexOf(query) if (index > -1) { @@ -729,10 +723,15 @@ class Client extends EventEmitter { this._pulseQueryQueue() }, readTimeout) - - query.callback = (err, res) => { - clearTimeout(readTimeoutTimer) - queryCallback(err, res) + //if there is onEndRead === null, use it, otherwise fallback to overriding callback + if (query.onEndRead === null) { + query.onEndRead = () => clearTimeout(readTimeoutTimer) + } else { + const queryCallback = query.callback || (() => {}) + query.callback = (err, res) => { + clearTimeout(readTimeoutTimer) + queryCallback(err, res) + } } } diff --git a/packages/pg/lib/native/client.js b/packages/pg/lib/native/client.js index 9ec3c8c03..e21ba6ecf 100644 --- a/packages/pg/lib/native/client.js +++ b/packages/pg/lib/native/client.js @@ -190,21 +190,16 @@ Client.prototype.query = function (config, values, callback) { } if (readTimeout) { - queryCallback = query.callback || (() => {}) - readTimeoutTimer = setTimeout(() => { const error = new Error('Query read timeout') process.nextTick(() => { query.handleError(error, this.connection) + // we already returned an error, + // just do nothing if query completes + query.callback = () => {} }) - queryCallback(error) - - // we already returned an error, - // just do nothing if query completes - query.callback = () => {} - // Remove from queue const index = this._queryQueue.indexOf(query) if (index > -1) { @@ -214,9 +209,14 @@ Client.prototype.query = function (config, values, callback) { this._pulseQueryQueue() }, readTimeout) - query.callback = (err, res) => { - clearTimeout(readTimeoutTimer) - queryCallback(err, res) + if (query.onEndRead === null) { + query.onEndRead = () => clearTimeout(readTimeoutTimer) + } else { + queryCallback = query.callback || (() => {}) + query.callback = (err, res) => { + clearTimeout(readTimeoutTimer) + queryCallback(err, res) + } } } diff --git a/packages/pg/lib/native/query.js b/packages/pg/lib/native/query.js index 39dfcc20b..0f71001d7 100644 --- a/packages/pg/lib/native/query.js +++ b/packages/pg/lib/native/query.js @@ -12,6 +12,7 @@ const NativeQuery = (module.exports = function (config, values, callback) { this.name = config.name this.queryMode = config.queryMode this.callback = config.callback + this.onEndRead = null this.state = 'new' this._arrayMode = config.rowMode === 'array' @@ -57,6 +58,7 @@ NativeQuery.prototype.handleError = function (err) { err[normalizedFieldName] = fields[key] } } + if (this.onEndRead) this.onEndRead() if (this.callback) { this.callback(err) } else { @@ -116,6 +118,7 @@ NativeQuery.prototype.submit = function (client) { } } + if (self.onEndRead) self.onEndRead() // handle successful result self.state = 'end' self.emit('end', results) diff --git a/packages/pg/lib/query.js b/packages/pg/lib/query.js index 3b93cb5ba..f2b1f966a 100644 --- a/packages/pg/lib/query.js +++ b/packages/pg/lib/query.js @@ -30,6 +30,7 @@ class Query extends EventEmitter { // potential for multiple results this._results = this._result this._canceledDueToError = false + this.onEndRead = null } requiresPreparation() { @@ -125,6 +126,7 @@ class Query extends EventEmitter { err = this._canceledDueToError this._canceledDueToError = false } + if (this.onEndRead) this.onEndRead() // if callback supplied do not emit error event as uncaught error // events will bubble up to node process if (this.callback) { @@ -146,6 +148,7 @@ class Query extends EventEmitter { }) } } + if (this.onEndRead) this.onEndRead() this.emit('end', this._results) }