You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
165 lines
4.6 KiB
165 lines
4.6 KiB
'use strict' |
|
|
|
const EventEmitter = require('events').EventEmitter |
|
const util = require('util') |
|
const utils = require('../utils') |
|
|
|
const NativeQuery = (module.exports = function (config, values, callback) { |
|
EventEmitter.call(this) |
|
config = utils.normalizeQueryConfig(config, values, callback) |
|
this.text = config.text |
|
this.values = config.values |
|
this.name = config.name |
|
this.queryMode = config.queryMode |
|
this.callback = config.callback |
|
this.state = 'new' |
|
this._arrayMode = config.rowMode === 'array' |
|
|
|
// if the 'row' event is listened for |
|
// then emit them as they come in |
|
// without setting singleRowMode to true |
|
// this has almost no meaning because libpq |
|
// reads all rows into memory before returning any |
|
this._emitRowEvents = false |
|
this.on( |
|
'newListener', |
|
function (event) { |
|
if (event === 'row') this._emitRowEvents = true |
|
}.bind(this) |
|
) |
|
}) |
|
|
|
util.inherits(NativeQuery, EventEmitter) |
|
|
|
const errorFieldMap = { |
|
sqlState: 'code', |
|
statementPosition: 'position', |
|
messagePrimary: 'message', |
|
context: 'where', |
|
schemaName: 'schema', |
|
tableName: 'table', |
|
columnName: 'column', |
|
dataTypeName: 'dataType', |
|
constraintName: 'constraint', |
|
sourceFile: 'file', |
|
sourceLine: 'line', |
|
sourceFunction: 'routine', |
|
} |
|
|
|
NativeQuery.prototype.handleError = function (err) { |
|
// copy pq error fields into the error object |
|
const fields = this.native.pq.resultErrorFields() |
|
if (fields) { |
|
for (const key in fields) { |
|
const normalizedFieldName = errorFieldMap[key] || key |
|
err[normalizedFieldName] = fields[key] |
|
} |
|
} |
|
if (this.callback) { |
|
this.callback(err) |
|
} else { |
|
this.emit('error', err) |
|
} |
|
this.state = 'error' |
|
} |
|
|
|
NativeQuery.prototype.then = function (onSuccess, onFailure) { |
|
return this._getPromise().then(onSuccess, onFailure) |
|
} |
|
|
|
NativeQuery.prototype.catch = function (callback) { |
|
return this._getPromise().catch(callback) |
|
} |
|
|
|
NativeQuery.prototype._getPromise = function () { |
|
if (this._promise) return this._promise |
|
this._promise = new Promise( |
|
function (resolve, reject) { |
|
this._once('end', resolve) |
|
this._once('error', reject) |
|
}.bind(this) |
|
) |
|
return this._promise |
|
} |
|
|
|
NativeQuery.prototype.submit = function (client) { |
|
this.state = 'running' |
|
const self = this |
|
this.native = client.native |
|
client.native.arrayMode = this._arrayMode |
|
|
|
let after = function (err, rows, results) { |
|
client.native.arrayMode = false |
|
setImmediate(function () { |
|
self.emit('_done') |
|
}) |
|
|
|
// handle possible query error |
|
if (err) { |
|
return self.handleError(err) |
|
} |
|
|
|
// emit row events for each row in the result |
|
if (self._emitRowEvents) { |
|
if (results.length > 1) { |
|
rows.forEach((rowOfRows, i) => { |
|
rowOfRows.forEach((row) => { |
|
self.emit('row', row, results[i]) |
|
}) |
|
}) |
|
} else { |
|
rows.forEach(function (row) { |
|
self.emit('row', row, results) |
|
}) |
|
} |
|
} |
|
|
|
// handle successful result |
|
self.state = 'end' |
|
self.emit('end', results) |
|
if (self.callback) { |
|
self.callback(null, results) |
|
} |
|
} |
|
|
|
if (process.domain) { |
|
after = process.domain.bind(after) |
|
} |
|
|
|
// named query |
|
if (this.name) { |
|
if (this.name.length > 63) { |
|
console.error('Warning! Postgres only supports 63 characters for query names.') |
|
console.error('You supplied %s (%s)', this.name, this.name.length) |
|
console.error('This can cause conflicts and silent errors executing queries') |
|
} |
|
const values = (this.values || []).map(utils.prepareValue) |
|
|
|
// check if the client has already executed this named query |
|
// if so...just execute it again - skip the planning phase |
|
if (client.namedQueries[this.name]) { |
|
if (this.text && client.namedQueries[this.name] !== this.text) { |
|
const err = new Error(`Prepared statements must be unique - '${this.name}' was used for a different statement`) |
|
return after(err) |
|
} |
|
return client.native.execute(this.name, values, after) |
|
} |
|
// plan the named query the first time, then execute it |
|
return client.native.prepare(this.name, this.text, values.length, function (err) { |
|
if (err) return after(err) |
|
client.namedQueries[self.name] = self.text |
|
return self.native.execute(self.name, values, after) |
|
}) |
|
} else if (this.values) { |
|
if (!Array.isArray(this.values)) { |
|
const err = new Error('Query values must be an array') |
|
return after(err) |
|
} |
|
const vals = this.values.map(utils.prepareValue) |
|
client.native.query(this.text, vals, after) |
|
} else if (this.queryMode === 'extended') { |
|
client.native.query(this.text, [], after) |
|
} else { |
|
client.native.query(this.text, after) |
|
} |
|
}
|
|
|