request.js 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587
  1. 'use strict'
  2. const debug = require('debug')('mssql:base')
  3. const { EventEmitter } = require('events')
  4. const { IDS, objectHasProperty } = require('../utils')
  5. const globalConnection = require('../global-connection')
  6. const { RequestError, ConnectionError } = require('../error')
  7. const { TYPES } = require('../datatypes')
  8. const shared = require('../shared')
  9. /**
  10. * Class Request.
  11. *
  12. * @property {Transaction} transaction Reference to transaction when request was created in transaction.
  13. * @property {*} parameters Collection of input and output parameters.
  14. * @property {Boolean} canceled `true` if request was canceled.
  15. *
  16. * @fires Request#recordset
  17. * @fires Request#row
  18. * @fires Request#done
  19. * @fires Request#error
  20. */
  21. class Request extends EventEmitter {
  22. /**
  23. * Create new Request.
  24. *
  25. * @param {Connection|ConnectionPool|Transaction|PreparedStatement} parent If ommited, global connection is used instead.
  26. */
  27. constructor (parent) {
  28. super()
  29. IDS.add(this, 'Request')
  30. debug('request(%d): created', IDS.get(this))
  31. this.canceled = false
  32. this._paused = false
  33. this.parent = parent || globalConnection.pool
  34. this.parameters = {}
  35. }
  36. /**
  37. * Fetch request from tagged template string.
  38. *
  39. * @private
  40. * @param {Array} strings
  41. * @param {Array} values
  42. * @param {String} [method] If provided, method is automtically called with serialized command on this object.
  43. * @return {Request}
  44. */
  45. _template (strings, values, method) {
  46. const command = [strings[0]]
  47. for (let index = 0; index < values.length; index++) {
  48. const value = values[index]
  49. // if value is an array, prepare each items as it's own comma separated parameter
  50. if (Array.isArray(value)) {
  51. for (let parameterIndex = 0; parameterIndex < value.length; parameterIndex++) {
  52. this.input(`param${index + 1}_${parameterIndex}`, value[parameterIndex])
  53. command.push(`@param${index + 1}_${parameterIndex}`)
  54. if (parameterIndex < value.length - 1) {
  55. command.push(', ')
  56. } else {
  57. command.push(strings[index + 1])
  58. }
  59. }
  60. } else {
  61. this.input(`param${index + 1}`, value)
  62. command.push(`@param${index + 1}`, strings[index + 1])
  63. }
  64. }
  65. if (method) {
  66. return this[method](command.join(''))
  67. } else {
  68. return command.join('')
  69. }
  70. }
  71. /**
  72. * Add an input parameter to the request.
  73. *
  74. * @param {String} name Name of the input parameter without @ char.
  75. * @param {*} [type] SQL data type of input parameter. If you omit type, module automaticaly decide which SQL data type should be used based on JS data type.
  76. * @param {*} value Input parameter value. `undefined` and `NaN` values are automatically converted to `null` values.
  77. * @return {Request}
  78. */
  79. input (name, type, value) {
  80. if ((/(--| |\/\*|\*\/|')/).test(name)) {
  81. throw new RequestError(`SQL injection warning for param '${name}'`, 'EINJECT')
  82. }
  83. if (arguments.length < 2) {
  84. throw new RequestError('Invalid number of arguments. At least 2 arguments expected.', 'EARGS')
  85. } else if (arguments.length === 2) {
  86. value = type
  87. type = shared.getTypeByValue(value)
  88. }
  89. // support for custom data types
  90. if (value && typeof value.valueOf === 'function' && !(value instanceof Date)) value = value.valueOf()
  91. if (value === undefined) value = null // undefined to null
  92. if (typeof value === 'number' && isNaN(value)) value = null // NaN to null
  93. if (type instanceof Function) type = type()
  94. if (objectHasProperty(this.parameters, name)) {
  95. throw new RequestError(`The parameter name ${name} has already been declared. Parameter names must be unique`, 'EDUPEPARAM')
  96. }
  97. this.parameters[name] = {
  98. name,
  99. type: type.type,
  100. io: 1,
  101. value,
  102. length: type.length,
  103. scale: type.scale,
  104. precision: type.precision,
  105. tvpType: type.tvpType
  106. }
  107. return this
  108. }
  109. /**
  110. * Replace an input parameter on the request.
  111. *
  112. * @param {String} name Name of the input parameter without @ char.
  113. * @param {*} [type] SQL data type of input parameter. If you omit type, module automaticaly decide which SQL data type should be used based on JS data type.
  114. * @param {*} value Input parameter value. `undefined` and `NaN` values are automatically converted to `null` values.
  115. * @return {Request}
  116. */
  117. replaceInput (name, type, value) {
  118. delete this.parameters[name]
  119. return this.input(name, type, value)
  120. }
  121. /**
  122. * Add an output parameter to the request.
  123. *
  124. * @param {String} name Name of the output parameter without @ char.
  125. * @param {*} type SQL data type of output parameter.
  126. * @param {*} [value] Output parameter value initial value. `undefined` and `NaN` values are automatically converted to `null` values. Optional.
  127. * @return {Request}
  128. */
  129. output (name, type, value) {
  130. if (!type) { type = TYPES.NVarChar }
  131. if ((/(--| |\/\*|\*\/|')/).test(name)) {
  132. throw new RequestError(`SQL injection warning for param '${name}'`, 'EINJECT')
  133. }
  134. if ((type === TYPES.Text) || (type === TYPES.NText) || (type === TYPES.Image)) {
  135. throw new RequestError('Deprecated types (Text, NText, Image) are not supported as OUTPUT parameters.', 'EDEPRECATED')
  136. }
  137. // support for custom data types
  138. if (value && typeof value.valueOf === 'function' && !(value instanceof Date)) value = value.valueOf()
  139. if (value === undefined) value = null // undefined to null
  140. if (typeof value === 'number' && isNaN(value)) value = null // NaN to null
  141. if (type instanceof Function) type = type()
  142. if (objectHasProperty(this.parameters, name)) {
  143. throw new RequestError(`The parameter name ${name} has already been declared. Parameter names must be unique`, 'EDUPEPARAM')
  144. }
  145. this.parameters[name] = {
  146. name,
  147. type: type.type,
  148. io: 2,
  149. value,
  150. length: type.length,
  151. scale: type.scale,
  152. precision: type.precision
  153. }
  154. return this
  155. }
  156. /**
  157. * Replace an output parameter on the request.
  158. *
  159. * @param {String} name Name of the output parameter without @ char.
  160. * @param {*} type SQL data type of output parameter.
  161. * @param {*} [value] Output parameter value initial value. `undefined` and `NaN` values are automatically converted to `null` values. Optional.
  162. * @return {Request}
  163. */
  164. replaceOutput (name, type, value) {
  165. delete this.parameters[name]
  166. return this.output(name, type, value)
  167. }
  168. /**
  169. * Execute the SQL batch.
  170. *
  171. * @param {String} batch T-SQL batch to be executed.
  172. * @param {Request~requestCallback} [callback] A callback which is called after execution has completed, or an error has occurred. If omited, method returns Promise.
  173. * @return {Request|Promise}
  174. */
  175. batch (batch, callback) {
  176. if (this.stream == null && this.connection) this.stream = this.connection.config.stream
  177. this.rowsAffected = 0
  178. if (typeof callback === 'function') {
  179. this._batch(batch, (err, recordsets, output, rowsAffected) => {
  180. if (this.stream) {
  181. if (err) this.emit('error', err)
  182. err = null
  183. this.emit('done', {
  184. output,
  185. rowsAffected
  186. })
  187. }
  188. if (err) return callback(err)
  189. callback(null, {
  190. recordsets,
  191. recordset: recordsets && recordsets[0],
  192. output,
  193. rowsAffected
  194. })
  195. })
  196. return this
  197. }
  198. // Check is method was called as tagged template
  199. if (typeof batch === 'object') {
  200. const values = Array.prototype.slice.call(arguments)
  201. const strings = values.shift()
  202. batch = this._template(strings, values)
  203. }
  204. return new shared.Promise((resolve, reject) => {
  205. this._batch(batch, (err, recordsets, output, rowsAffected) => {
  206. if (this.stream) {
  207. if (err) this.emit('error', err)
  208. err = null
  209. this.emit('done', {
  210. output,
  211. rowsAffected
  212. })
  213. }
  214. if (err) return reject(err)
  215. resolve({
  216. recordsets,
  217. recordset: recordsets && recordsets[0],
  218. output,
  219. rowsAffected
  220. })
  221. })
  222. })
  223. }
  224. /**
  225. * @private
  226. * @param {String} batch
  227. * @param {Request~requestCallback} callback
  228. */
  229. _batch (batch, callback) {
  230. if (!this.connection) {
  231. return setImmediate(callback, new RequestError('No connection is specified for that request.', 'ENOCONN'))
  232. }
  233. if (!this.connection.connected) {
  234. return setImmediate(callback, new ConnectionError('Connection is closed.', 'ECONNCLOSED'))
  235. }
  236. this.canceled = false
  237. setImmediate(callback)
  238. }
  239. /**
  240. * Bulk load.
  241. *
  242. * @param {Table} table SQL table.
  243. * @param {object} [options] Options to be passed to the underlying driver (tedious only).
  244. * @param {Request~bulkCallback} [callback] A callback which is called after bulk load has completed, or an error has occurred. If omited, method returns Promise.
  245. * @return {Request|Promise}
  246. */
  247. bulk (table, options, callback) {
  248. if (typeof options === 'function') {
  249. callback = options
  250. options = {}
  251. } else if (typeof options === 'undefined') {
  252. options = {}
  253. }
  254. if (this.stream == null && this.connection) this.stream = this.connection.config.stream
  255. if (this.stream || typeof callback === 'function') {
  256. this._bulk(table, options, (err, rowsAffected) => {
  257. if (this.stream) {
  258. if (err) this.emit('error', err)
  259. return this.emit('done', {
  260. rowsAffected
  261. })
  262. }
  263. if (err) return callback(err)
  264. callback(null, {
  265. rowsAffected
  266. })
  267. })
  268. return this
  269. }
  270. return new shared.Promise((resolve, reject) => {
  271. this._bulk(table, options, (err, rowsAffected) => {
  272. if (err) return reject(err)
  273. resolve({
  274. rowsAffected
  275. })
  276. })
  277. })
  278. }
  279. /**
  280. * @private
  281. * @param {Table} table
  282. * @param {object} options
  283. * @param {Request~bulkCallback} callback
  284. */
  285. _bulk (table, options, callback) {
  286. if (!this.parent) {
  287. return setImmediate(callback, new RequestError('No connection is specified for that request.', 'ENOCONN'))
  288. }
  289. if (!this.parent.connected) {
  290. return setImmediate(callback, new ConnectionError('Connection is closed.', 'ECONNCLOSED'))
  291. }
  292. this.canceled = false
  293. setImmediate(callback)
  294. }
  295. /**
  296. * Sets request to `stream` mode and pulls all rows from all recordsets to a given stream.
  297. *
  298. * @param {Stream} stream Stream to pipe data into.
  299. * @return {Stream}
  300. */
  301. pipe (stream) {
  302. this.stream = true
  303. this.on('row', stream.write.bind(stream))
  304. this.on('error', stream.emit.bind(stream, 'error'))
  305. this.on('done', () => {
  306. setImmediate(() => stream.end())
  307. })
  308. stream.emit('pipe', this)
  309. return stream
  310. }
  311. /**
  312. * Execute the SQL command.
  313. *
  314. * @param {String} command T-SQL command to be executed.
  315. * @param {Request~requestCallback} [callback] A callback which is called after execution has completed, or an error has occurred. If omited, method returns Promise.
  316. * @return {Request|Promise}
  317. */
  318. query (command, callback) {
  319. if (this.stream == null && this.connection) this.stream = this.connection.config.stream
  320. this.rowsAffected = 0
  321. if (typeof callback === 'function') {
  322. this._query(command, (err, recordsets, output, rowsAffected) => {
  323. if (this.stream) {
  324. if (err) this.emit('error', err)
  325. err = null
  326. this.emit('done', {
  327. output,
  328. rowsAffected
  329. })
  330. }
  331. if (err) return callback(err)
  332. callback(null, {
  333. recordsets,
  334. recordset: recordsets && recordsets[0],
  335. output,
  336. rowsAffected
  337. })
  338. })
  339. return this
  340. }
  341. // Check is method was called as tagged template
  342. if (typeof command === 'object') {
  343. const values = Array.prototype.slice.call(arguments)
  344. const strings = values.shift()
  345. command = this._template(strings, values)
  346. }
  347. return new shared.Promise((resolve, reject) => {
  348. this._query(command, (err, recordsets, output, rowsAffected) => {
  349. if (this.stream) {
  350. if (err) this.emit('error', err)
  351. err = null
  352. this.emit('done', {
  353. output,
  354. rowsAffected
  355. })
  356. }
  357. if (err) return reject(err)
  358. resolve({
  359. recordsets,
  360. recordset: recordsets && recordsets[0],
  361. output,
  362. rowsAffected
  363. })
  364. })
  365. })
  366. }
  367. /**
  368. * @private
  369. * @param {String} command
  370. * @param {Request~bulkCallback} callback
  371. */
  372. _query (command, callback) {
  373. if (!this.parent) {
  374. return setImmediate(callback, new RequestError('No connection is specified for that request.', 'ENOCONN'))
  375. }
  376. if (!this.parent.connected) {
  377. return setImmediate(callback, new ConnectionError('Connection is closed.', 'ECONNCLOSED'))
  378. }
  379. this.canceled = false
  380. setImmediate(callback)
  381. }
  382. /**
  383. * Call a stored procedure.
  384. *
  385. * @param {String} procedure Name of the stored procedure to be executed.
  386. * @param {Request~requestCallback} [callback] A callback which is called after execution has completed, or an error has occurred. If omited, method returns Promise.
  387. * @return {Request|Promise}
  388. */
  389. execute (command, callback) {
  390. if (this.stream == null && this.connection) this.stream = this.connection.config.stream
  391. this.rowsAffected = 0
  392. if (typeof callback === 'function') {
  393. this._execute(command, (err, recordsets, output, returnValue, rowsAffected) => {
  394. if (this.stream) {
  395. if (err) this.emit('error', err)
  396. err = null
  397. this.emit('done', {
  398. output,
  399. rowsAffected,
  400. returnValue
  401. })
  402. }
  403. if (err) return callback(err)
  404. callback(null, {
  405. recordsets,
  406. recordset: recordsets && recordsets[0],
  407. output,
  408. rowsAffected,
  409. returnValue
  410. })
  411. })
  412. return this
  413. }
  414. return new shared.Promise((resolve, reject) => {
  415. this._execute(command, (err, recordsets, output, returnValue, rowsAffected) => {
  416. if (this.stream) {
  417. if (err) this.emit('error', err)
  418. err = null
  419. this.emit('done', {
  420. output,
  421. rowsAffected,
  422. returnValue
  423. })
  424. }
  425. if (err) return reject(err)
  426. resolve({
  427. recordsets,
  428. recordset: recordsets && recordsets[0],
  429. output,
  430. rowsAffected,
  431. returnValue
  432. })
  433. })
  434. })
  435. }
  436. /**
  437. * @private
  438. * @param {String} procedure
  439. * @param {Request~bulkCallback} callback
  440. */
  441. _execute (procedure, callback) {
  442. if (!this.parent) {
  443. return setImmediate(callback, new RequestError('No connection is specified for that request.', 'ENOCONN'))
  444. }
  445. if (!this.parent.connected) {
  446. return setImmediate(callback, new ConnectionError('Connection is closed.', 'ECONNCLOSED'))
  447. }
  448. this.canceled = false
  449. setImmediate(callback)
  450. }
  451. /**
  452. * Cancel currently executed request.
  453. *
  454. * @return {Boolean}
  455. */
  456. cancel () {
  457. this._cancel()
  458. return true
  459. }
  460. /**
  461. * @private
  462. */
  463. _cancel () {
  464. this.canceled = true
  465. }
  466. pause () {
  467. if (this.stream) {
  468. this._pause()
  469. return true
  470. }
  471. return false
  472. }
  473. _pause () {
  474. this._paused = true
  475. }
  476. resume () {
  477. if (this.stream) {
  478. this._resume()
  479. return true
  480. }
  481. return false
  482. }
  483. _resume () {
  484. this._paused = false
  485. }
  486. _setCurrentRequest (request) {
  487. this._currentRequest = request
  488. if (this._paused) {
  489. this.pause()
  490. }
  491. return this
  492. }
  493. }
  494. module.exports = Request