query.js 4.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165
  1. 'use strict'
  2. var EventEmitter = require('events').EventEmitter
  3. var util = require('util')
  4. var utils = require('../utils')
  5. var NativeQuery = (module.exports = function (config, values, callback) {
  6. EventEmitter.call(this)
  7. config = utils.normalizeQueryConfig(config, values, callback)
  8. this.text = config.text
  9. this.values = config.values
  10. this.name = config.name
  11. this.callback = config.callback
  12. this.state = 'new'
  13. this._arrayMode = config.rowMode === 'array'
  14. // if the 'row' event is listened for
  15. // then emit them as they come in
  16. // without setting singleRowMode to true
  17. // this has almost no meaning because libpq
  18. // reads all rows into memory befor returning any
  19. this._emitRowEvents = false
  20. this.on(
  21. 'newListener',
  22. function (event) {
  23. if (event === 'row') this._emitRowEvents = true
  24. }.bind(this)
  25. )
  26. })
  27. util.inherits(NativeQuery, EventEmitter)
  28. var errorFieldMap = {
  29. /* eslint-disable quote-props */
  30. sqlState: 'code',
  31. statementPosition: 'position',
  32. messagePrimary: 'message',
  33. context: 'where',
  34. schemaName: 'schema',
  35. tableName: 'table',
  36. columnName: 'column',
  37. dataTypeName: 'dataType',
  38. constraintName: 'constraint',
  39. sourceFile: 'file',
  40. sourceLine: 'line',
  41. sourceFunction: 'routine',
  42. }
  43. NativeQuery.prototype.handleError = function (err) {
  44. // copy pq error fields into the error object
  45. var fields = this.native.pq.resultErrorFields()
  46. if (fields) {
  47. for (var key in fields) {
  48. var normalizedFieldName = errorFieldMap[key] || key
  49. err[normalizedFieldName] = fields[key]
  50. }
  51. }
  52. if (this.callback) {
  53. this.callback(err)
  54. } else {
  55. this.emit('error', err)
  56. }
  57. this.state = 'error'
  58. }
  59. NativeQuery.prototype.then = function (onSuccess, onFailure) {
  60. return this._getPromise().then(onSuccess, onFailure)
  61. }
  62. NativeQuery.prototype.catch = function (callback) {
  63. return this._getPromise().catch(callback)
  64. }
  65. NativeQuery.prototype._getPromise = function () {
  66. if (this._promise) return this._promise
  67. this._promise = new Promise(
  68. function (resolve, reject) {
  69. this._once('end', resolve)
  70. this._once('error', reject)
  71. }.bind(this)
  72. )
  73. return this._promise
  74. }
  75. NativeQuery.prototype.submit = function (client) {
  76. this.state = 'running'
  77. var self = this
  78. this.native = client.native
  79. client.native.arrayMode = this._arrayMode
  80. var after = function (err, rows, results) {
  81. client.native.arrayMode = false
  82. setImmediate(function () {
  83. self.emit('_done')
  84. })
  85. // handle possible query error
  86. if (err) {
  87. return self.handleError(err)
  88. }
  89. // emit row events for each row in the result
  90. if (self._emitRowEvents) {
  91. if (results.length > 1) {
  92. rows.forEach((rowOfRows, i) => {
  93. rowOfRows.forEach((row) => {
  94. self.emit('row', row, results[i])
  95. })
  96. })
  97. } else {
  98. rows.forEach(function (row) {
  99. self.emit('row', row, results)
  100. })
  101. }
  102. }
  103. // handle successful result
  104. self.state = 'end'
  105. self.emit('end', results)
  106. if (self.callback) {
  107. self.callback(null, results)
  108. }
  109. }
  110. if (process.domain) {
  111. after = process.domain.bind(after)
  112. }
  113. // named query
  114. if (this.name) {
  115. if (this.name.length > 63) {
  116. /* eslint-disable no-console */
  117. console.error('Warning! Postgres only supports 63 characters for query names.')
  118. console.error('You supplied %s (%s)', this.name, this.name.length)
  119. console.error('This can cause conflicts and silent errors executing queries')
  120. /* eslint-enable no-console */
  121. }
  122. var values = (this.values || []).map(utils.prepareValue)
  123. // check if the client has already executed this named query
  124. // if so...just execute it again - skip the planning phase
  125. if (client.namedQueries[this.name]) {
  126. if (this.text && client.namedQueries[this.name] !== this.text) {
  127. const err = new Error(`Prepared statements must be unique - '${this.name}' was used for a different statement`)
  128. return after(err)
  129. }
  130. return client.native.execute(this.name, values, after)
  131. }
  132. // plan the named query the first time, then execute it
  133. return client.native.prepare(this.name, this.text, values.length, function (err) {
  134. if (err) return after(err)
  135. client.namedQueries[self.name] = self.text
  136. return self.native.execute(self.name, values, after)
  137. })
  138. } else if (this.values) {
  139. if (!Array.isArray(this.values)) {
  140. const err = new Error('Query values must be an array')
  141. return after(err)
  142. }
  143. var vals = this.values.map(utils.prepareValue)
  144. client.native.query(this.text, vals, after)
  145. } else {
  146. client.native.query(this.text, after)
  147. }
  148. }