query.js 4.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165
  1. 'use strict'
  2. const EventEmitter = require('events').EventEmitter
  3. const util = require('util')
  4. const utils = require('../utils')
  5. const 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.queryMode = config.queryMode
  12. this.callback = config.callback
  13. this.state = 'new'
  14. this._arrayMode = config.rowMode === 'array'
  15. // if the 'row' event is listened for
  16. // then emit them as they come in
  17. // without setting singleRowMode to true
  18. // this has almost no meaning because libpq
  19. // reads all rows into memory befor returning any
  20. this._emitRowEvents = false
  21. this.on(
  22. 'newListener',
  23. function (event) {
  24. if (event === 'row') this._emitRowEvents = true
  25. }.bind(this)
  26. )
  27. })
  28. util.inherits(NativeQuery, EventEmitter)
  29. const errorFieldMap = {
  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. const fields = this.native.pq.resultErrorFields()
  46. if (fields) {
  47. for (const key in fields) {
  48. const 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. const self = this
  78. this.native = client.native
  79. client.native.arrayMode = this._arrayMode
  80. let 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. console.error('Warning! Postgres only supports 63 characters for query names.')
  117. console.error('You supplied %s (%s)', this.name, this.name.length)
  118. console.error('This can cause conflicts and silent errors executing queries')
  119. }
  120. const values = (this.values || []).map(utils.prepareValue)
  121. // check if the client has already executed this named query
  122. // if so...just execute it again - skip the planning phase
  123. if (client.namedQueries[this.name]) {
  124. if (this.text && client.namedQueries[this.name] !== this.text) {
  125. const err = new Error(`Prepared statements must be unique - '${this.name}' was used for a different statement`)
  126. return after(err)
  127. }
  128. return client.native.execute(this.name, values, after)
  129. }
  130. // plan the named query the first time, then execute it
  131. return client.native.prepare(this.name, this.text, values.length, function (err) {
  132. if (err) return after(err)
  133. client.namedQueries[self.name] = self.text
  134. return self.native.execute(self.name, values, after)
  135. })
  136. } else if (this.values) {
  137. if (!Array.isArray(this.values)) {
  138. const err = new Error('Query values must be an array')
  139. return after(err)
  140. }
  141. const vals = this.values.map(utils.prepareValue)
  142. client.native.query(this.text, vals, after)
  143. } else if (this.queryMode === 'extended') {
  144. client.native.query(this.text, [], after)
  145. } else {
  146. client.native.query(this.text, after)
  147. }
  148. }