request.js 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942
  1. 'use strict'
  2. const tds = require('tedious')
  3. const debug = require('debug')('mssql:tedi')
  4. const BaseRequest = require('../base/request')
  5. const RequestError = require('../error/request-error')
  6. const { IDS, objectHasProperty } = require('../utils')
  7. const { TYPES, DECLARATIONS, declare, cast } = require('../datatypes')
  8. const Table = require('../table')
  9. const { PARSERS: UDT } = require('../udt')
  10. const JSON_COLUMN_ID = 'JSON_F52E2B61-18A1-11d1-B105-00805F49916B'
  11. const XML_COLUMN_ID = 'XML_F52E2B61-18A1-11d1-B105-00805F49916B'
  12. const N_TYPES = {
  13. BitN: 0x68,
  14. DateTimeN: 0x6F,
  15. DecimalN: 0x6A,
  16. FloatN: 0x6D,
  17. IntN: 0x26,
  18. MoneyN: 0x6E,
  19. NumericN: 0x6C
  20. }
  21. const getTediousType = function (type) {
  22. switch (type) {
  23. case TYPES.VarChar: return tds.TYPES.VarChar
  24. case TYPES.NVarChar: return tds.TYPES.NVarChar
  25. case TYPES.Text: return tds.TYPES.Text
  26. case TYPES.Int: return tds.TYPES.Int
  27. case TYPES.BigInt: return tds.TYPES.BigInt
  28. case TYPES.TinyInt: return tds.TYPES.TinyInt
  29. case TYPES.SmallInt: return tds.TYPES.SmallInt
  30. case TYPES.Bit: return tds.TYPES.Bit
  31. case TYPES.Float: return tds.TYPES.Float
  32. case TYPES.Decimal: return tds.TYPES.Decimal
  33. case TYPES.Numeric: return tds.TYPES.Numeric
  34. case TYPES.Real: return tds.TYPES.Real
  35. case TYPES.Money: return tds.TYPES.Money
  36. case TYPES.SmallMoney: return tds.TYPES.SmallMoney
  37. case TYPES.Time: return tds.TYPES.Time
  38. case TYPES.Date: return tds.TYPES.Date
  39. case TYPES.DateTime: return tds.TYPES.DateTime
  40. case TYPES.DateTime2: return tds.TYPES.DateTime2
  41. case TYPES.DateTimeOffset: return tds.TYPES.DateTimeOffset
  42. case TYPES.SmallDateTime: return tds.TYPES.SmallDateTime
  43. case TYPES.UniqueIdentifier: return tds.TYPES.UniqueIdentifier
  44. case TYPES.Xml: return tds.TYPES.NVarChar
  45. case TYPES.Char: return tds.TYPES.Char
  46. case TYPES.NChar: return tds.TYPES.NChar
  47. case TYPES.NText: return tds.TYPES.NVarChar
  48. case TYPES.Image: return tds.TYPES.Image
  49. case TYPES.Binary: return tds.TYPES.Binary
  50. case TYPES.VarBinary: return tds.TYPES.VarBinary
  51. case TYPES.UDT: case TYPES.Geography: case TYPES.Geometry: return tds.TYPES.UDT
  52. case TYPES.TVP: return tds.TYPES.TVP
  53. case TYPES.Variant: return tds.TYPES.Variant
  54. default: return type
  55. }
  56. }
  57. const getMssqlType = function (type, length) {
  58. if (typeof type !== 'object') return undefined
  59. switch (type) {
  60. case tds.TYPES.Char: return TYPES.Char
  61. case tds.TYPES.NChar: return TYPES.NChar
  62. case tds.TYPES.VarChar: return TYPES.VarChar
  63. case tds.TYPES.NVarChar: return TYPES.NVarChar
  64. case tds.TYPES.Text: return TYPES.Text
  65. case tds.TYPES.NText: return TYPES.NText
  66. case tds.TYPES.Int: return TYPES.Int
  67. case tds.TYPES.BigInt: return TYPES.BigInt
  68. case tds.TYPES.TinyInt: return TYPES.TinyInt
  69. case tds.TYPES.SmallInt: return TYPES.SmallInt
  70. case tds.TYPES.Bit: return TYPES.Bit
  71. case tds.TYPES.Float: return TYPES.Float
  72. case tds.TYPES.Real: return TYPES.Real
  73. case tds.TYPES.Money: return TYPES.Money
  74. case tds.TYPES.SmallMoney: return TYPES.SmallMoney
  75. case tds.TYPES.Numeric: return TYPES.Numeric
  76. case tds.TYPES.Decimal: return TYPES.Decimal
  77. case tds.TYPES.DateTime: return TYPES.DateTime
  78. case tds.TYPES.Time: return TYPES.Time
  79. case tds.TYPES.Date: return TYPES.Date
  80. case tds.TYPES.DateTime2: return TYPES.DateTime2
  81. case tds.TYPES.DateTimeOffset: return TYPES.DateTimeOffset
  82. case tds.TYPES.SmallDateTime: return TYPES.SmallDateTime
  83. case tds.TYPES.UniqueIdentifier: return TYPES.UniqueIdentifier
  84. case tds.TYPES.Image: return TYPES.Image
  85. case tds.TYPES.Binary: return TYPES.Binary
  86. case tds.TYPES.VarBinary: return TYPES.VarBinary
  87. case tds.TYPES.Xml: return TYPES.Xml
  88. case tds.TYPES.UDT: return TYPES.UDT
  89. case tds.TYPES.TVP: return TYPES.TVP
  90. case tds.TYPES.Variant: return TYPES.Variant
  91. default:
  92. switch (type.id) {
  93. case N_TYPES.BitN: return TYPES.Bit
  94. case N_TYPES.NumericN: return TYPES.Numeric
  95. case N_TYPES.DecimalN: return TYPES.Decimal
  96. case N_TYPES.IntN:
  97. if (length === 8) return TYPES.BigInt
  98. if (length === 4) return TYPES.Int
  99. if (length === 2) return TYPES.SmallInt
  100. return TYPES.TinyInt
  101. case N_TYPES.FloatN:
  102. if (length === 8) return TYPES.Float
  103. return TYPES.Real
  104. case N_TYPES.MoneyN:
  105. if (length === 8) return TYPES.Money
  106. return TYPES.SmallMoney
  107. case N_TYPES.DateTimeN:
  108. if (length === 8) return TYPES.DateTime
  109. return TYPES.SmallDateTime
  110. }
  111. }
  112. }
  113. const createColumns = function (metadata) {
  114. const out = {}
  115. for (let index = 0, length = metadata.length; index < length; index++) {
  116. const column = metadata[index]
  117. out[column.colName] = {
  118. index,
  119. name: column.colName,
  120. length: column.dataLength,
  121. type: getMssqlType(column.type, column.dataLength),
  122. scale: column.scale,
  123. precision: column.precision,
  124. nullable: !!(column.flags & 0x01),
  125. caseSensitive: !!(column.flags & 0x02),
  126. identity: !!(column.flags & 0x10),
  127. readOnly: !(column.flags & 0x0C)
  128. }
  129. if (column.udtInfo) {
  130. out[column.colName].udt = {
  131. name: column.udtInfo.typeName,
  132. database: column.udtInfo.dbname,
  133. schema: column.udtInfo.owningSchema,
  134. assembly: column.udtInfo.assemblyName
  135. }
  136. if (DECLARATIONS[column.udtInfo.typeName]) {
  137. out[column.colName].type = DECLARATIONS[column.udtInfo.typeName]
  138. }
  139. }
  140. }
  141. return out
  142. }
  143. const valueCorrection = function (value, metadata) {
  144. if ((metadata.type === tds.TYPES.UDT) && (value != null)) {
  145. if (UDT[metadata.udtInfo.typeName]) {
  146. return UDT[metadata.udtInfo.typeName](value)
  147. } else {
  148. return value
  149. }
  150. } else {
  151. return value
  152. }
  153. }
  154. const parameterCorrection = function (value) {
  155. if (value instanceof Table) {
  156. const tvp = {
  157. name: value.name,
  158. schema: value.schema,
  159. columns: [],
  160. rows: value.rows
  161. }
  162. for (const col of value.columns) {
  163. tvp.columns.push({
  164. name: col.name,
  165. type: getTediousType(col.type),
  166. length: col.length,
  167. scale: col.scale,
  168. precision: col.precision
  169. })
  170. }
  171. return tvp
  172. } else {
  173. return value
  174. }
  175. }
  176. class Request extends BaseRequest {
  177. /*
  178. Execute specified sql batch.
  179. */
  180. _batch (batch, callback) {
  181. this._isBatch = true
  182. this._query(batch, callback)
  183. }
  184. /*
  185. Bulk load.
  186. */
  187. _bulk (table, options, callback) {
  188. super._bulk(table, options, err => {
  189. if (err) return callback(err)
  190. table._makeBulk()
  191. if (!table.name) {
  192. return callback(new RequestError('Table name must be specified for bulk insert.', 'ENAME'))
  193. }
  194. if (table.name.charAt(0) === '@') {
  195. return callback(new RequestError("You can't use table variables for bulk insert.", 'ENAME'))
  196. }
  197. const errors = []
  198. const errorHandlers = {}
  199. let hasReturned = false
  200. const handleError = (doReturn, connection, info) => {
  201. let err = new Error(info.message)
  202. err.info = info
  203. err = new RequestError(err, 'EREQUEST')
  204. if (this.stream) {
  205. this.emit('error', err)
  206. } else {
  207. if (doReturn && !hasReturned) {
  208. if (connection) {
  209. for (const event in errorHandlers) {
  210. connection.removeListener(event, errorHandlers[event])
  211. }
  212. this.parent.release(connection)
  213. }
  214. hasReturned = true
  215. callback(err)
  216. }
  217. }
  218. // we must collect errors even in stream mode
  219. errors.push(err)
  220. }
  221. const handleInfo = msg => {
  222. this.emit('info', {
  223. message: msg.message,
  224. number: msg.number,
  225. state: msg.state,
  226. class: msg.class,
  227. lineNumber: msg.lineNumber,
  228. serverName: msg.serverName,
  229. procName: msg.procName
  230. })
  231. }
  232. this.parent.acquire(this, (err, connection) => {
  233. if (err) return callback(err)
  234. debug('connection(%d): borrowed to request #%d', IDS.get(connection), IDS.get(this))
  235. if (this.canceled) {
  236. debug('request(%d): canceled', IDS.get(this))
  237. this.parent.release(connection)
  238. return callback(new RequestError('Canceled.', 'ECANCEL'))
  239. }
  240. this._cancel = () => {
  241. debug('request(%d): cancel', IDS.get(this))
  242. connection.cancel()
  243. }
  244. // attach handler to handle multiple error messages
  245. connection.on('infoMessage', errorHandlers.infoMessage = handleInfo)
  246. connection.on('errorMessage', errorHandlers.errorMessage = handleError.bind(null, false, connection))
  247. connection.on('error', errorHandlers.error = handleError.bind(null, true, connection))
  248. const done = (err, rowCount) => {
  249. // to make sure we handle no-sql errors as well
  250. if (err && (!errors.length || (errors.length && err.message !== errors[errors.length - 1].message))) {
  251. err = new RequestError(err, 'EREQUEST')
  252. if (this.stream) this.emit('error', err)
  253. errors.push(err)
  254. }
  255. delete this._cancel
  256. let error
  257. if (errors.length && !this.stream) {
  258. error = errors.pop()
  259. error.precedingErrors = errors
  260. }
  261. if (!hasReturned) {
  262. for (const event in errorHandlers) {
  263. connection.removeListener(event, errorHandlers[event])
  264. }
  265. this.parent.release(connection)
  266. hasReturned = true
  267. if (this.stream) {
  268. callback(null, rowCount)
  269. } else {
  270. callback(error, rowCount)
  271. }
  272. }
  273. }
  274. const bulk = connection.newBulkLoad(table.path, options, done)
  275. for (const col of table.columns) {
  276. bulk.addColumn(col.name, getTediousType(col.type), { nullable: col.nullable, length: col.length, scale: col.scale, precision: col.precision })
  277. }
  278. for (const row of table.rows) {
  279. bulk.addRow(row)
  280. }
  281. if (table.create) {
  282. const objectid = table.temporary ? `tempdb..[${table.name}]` : table.path
  283. const req = new tds.Request(`if object_id('${objectid.replace(/'/g, '\'\'')}') is null ${table.declare()}`, err => {
  284. if (err) return done(err)
  285. connection.execBulkLoad(bulk)
  286. })
  287. this._setCurrentRequest(req)
  288. connection.execSqlBatch(req)
  289. } else {
  290. connection.execBulkLoad(bulk)
  291. }
  292. })
  293. })
  294. }
  295. /*
  296. Execute specified sql command.
  297. */
  298. _query (command, callback) {
  299. super._query(command, err => {
  300. if (err) return callback(err)
  301. const recordsets = []
  302. const errors = []
  303. const errorHandlers = {}
  304. const output = {}
  305. const rowsAffected = []
  306. let columns = {}
  307. let recordset = []
  308. let batchLastRow = null
  309. let batchHasOutput = false
  310. let isChunkedRecordset = false
  311. let chunksBuffer = null
  312. let hasReturned = false
  313. const handleError = (doReturn, connection, info) => {
  314. let err = new Error(info.message)
  315. err.info = info
  316. err = new RequestError(err, 'EREQUEST')
  317. if (this.stream) {
  318. this.emit('error', err)
  319. } else {
  320. if (doReturn && !hasReturned) {
  321. if (connection) {
  322. for (const event in errorHandlers) {
  323. connection.removeListener(event, errorHandlers[event])
  324. }
  325. this.parent.release(connection)
  326. }
  327. hasReturned = true
  328. callback(err)
  329. }
  330. }
  331. // we must collect errors even in stream mode
  332. errors.push(err)
  333. }
  334. const handleInfo = msg => {
  335. this.emit('info', {
  336. message: msg.message,
  337. number: msg.number,
  338. state: msg.state,
  339. class: msg.class,
  340. lineNumber: msg.lineNumber,
  341. serverName: msg.serverName,
  342. procName: msg.procName
  343. })
  344. }
  345. this.parent.acquire(this, (err, connection, config) => {
  346. if (err) return callback(err)
  347. debug('connection(%d): borrowed to request #%d', IDS.get(connection), IDS.get(this))
  348. let row
  349. if (this.canceled) {
  350. debug('request(%d): canceled', IDS.get(this))
  351. this.parent.release(connection)
  352. return callback(new RequestError('Canceled.', 'ECANCEL'))
  353. }
  354. this._cancel = () => {
  355. debug('request(%d): cancel', IDS.get(this))
  356. connection.cancel()
  357. }
  358. // attach handler to handle multiple error messages
  359. connection.on('infoMessage', errorHandlers.infoMessage = handleInfo)
  360. connection.on('errorMessage', errorHandlers.errorMessage = handleError.bind(null, false, connection))
  361. connection.on('error', errorHandlers.error = handleError.bind(null, true, connection))
  362. debug('request(%d): query', IDS.get(this), command)
  363. const req = new tds.Request(command, err => {
  364. // to make sure we handle no-sql errors as well
  365. if (err && (!errors.length || (errors.length && err.message !== errors[errors.length - 1].message))) {
  366. err = new RequestError(err, 'EREQUEST')
  367. if (this.stream) this.emit('error', err)
  368. errors.push(err)
  369. }
  370. // process batch outputs
  371. if (batchHasOutput) {
  372. if (!this.stream) batchLastRow = recordsets.pop()[0]
  373. for (const name in batchLastRow) {
  374. const value = batchLastRow[name]
  375. if (name !== '___return___') {
  376. output[name] = value
  377. }
  378. }
  379. }
  380. delete this._cancel
  381. let error
  382. if (errors.length && !this.stream) {
  383. error = errors.pop()
  384. error.precedingErrors = errors
  385. }
  386. if (!hasReturned) {
  387. for (const event in errorHandlers) {
  388. connection.removeListener(event, errorHandlers[event])
  389. }
  390. this.parent.release(connection)
  391. hasReturned = true
  392. if (error) {
  393. debug('request(%d): failed', IDS.get(this), error)
  394. } else {
  395. debug('request(%d): completed', IDS.get(this))
  396. }
  397. if (this.stream) {
  398. callback(null, null, output, rowsAffected)
  399. } else {
  400. callback(error, recordsets, output, rowsAffected)
  401. }
  402. }
  403. })
  404. this._setCurrentRequest(req)
  405. req.on('columnMetadata', metadata => {
  406. columns = createColumns(metadata)
  407. isChunkedRecordset = false
  408. if (metadata.length === 1 && (metadata[0].colName === JSON_COLUMN_ID || metadata[0].colName === XML_COLUMN_ID)) {
  409. isChunkedRecordset = true
  410. chunksBuffer = []
  411. }
  412. if (this.stream) {
  413. if (this._isBatch) {
  414. // don't stream recordset with output values in batches
  415. if (!columns.___return___) {
  416. this.emit('recordset', columns)
  417. }
  418. } else {
  419. this.emit('recordset', columns)
  420. }
  421. }
  422. })
  423. const doneHandler = (rowCount, more) => {
  424. if (rowCount != null) rowsAffected.push(rowCount)
  425. // this function is called even when select only set variables so we should skip adding a new recordset
  426. if (Object.keys(columns).length === 0) return
  427. if (isChunkedRecordset) {
  428. const concatenatedChunks = chunksBuffer.join('')
  429. if (columns[JSON_COLUMN_ID] && config.parseJSON === true) {
  430. try {
  431. if (concatenatedChunks === '') {
  432. row = null
  433. } else {
  434. row = JSON.parse(concatenatedChunks)
  435. }
  436. } catch (ex) {
  437. row = null
  438. const ex2 = new RequestError(new Error(`Failed to parse incoming JSON. ${ex.message}`), 'EJSON')
  439. if (this.stream) this.emit('error', ex2)
  440. // we must collect errors even in stream mode
  441. errors.push(ex2)
  442. }
  443. } else {
  444. row = {}
  445. row[Object.keys(columns)[0]] = concatenatedChunks
  446. }
  447. chunksBuffer = null
  448. if (this.stream) {
  449. this.emit('row', row)
  450. } else {
  451. recordset.push(row)
  452. }
  453. }
  454. if (!this.stream) {
  455. // all rows of current recordset loaded
  456. Object.defineProperty(recordset, 'columns', {
  457. enumerable: false,
  458. configurable: true,
  459. value: columns
  460. })
  461. Object.defineProperty(recordset, 'toTable', {
  462. enumerable: false,
  463. configurable: true,
  464. value (name) { return Table.fromRecordset(this, name) }
  465. })
  466. recordsets.push(recordset)
  467. }
  468. recordset = []
  469. columns = {}
  470. }
  471. req.on('doneInProc', doneHandler) // doneInProc handlers are used in both queries and batches
  472. req.on('done', doneHandler) // done handlers are used in batches
  473. req.on('returnValue', (parameterName, value, metadata) => {
  474. output[parameterName] = value
  475. })
  476. req.on('row', columns => {
  477. if (!recordset) recordset = []
  478. if (isChunkedRecordset) {
  479. return chunksBuffer.push(columns[0].value)
  480. }
  481. row = {}
  482. for (const col of columns) {
  483. col.value = valueCorrection(col.value, col.metadata)
  484. const exi = row[col.metadata.colName]
  485. if (exi != null) {
  486. if (exi instanceof Array) {
  487. exi.push(col.value)
  488. } else {
  489. row[col.metadata.colName] = [exi, col.value]
  490. }
  491. } else {
  492. row[col.metadata.colName] = col.value
  493. }
  494. }
  495. if (this.stream) {
  496. if (this._isBatch) {
  497. // dont stream recordset with output values in batches
  498. if (row.___return___) {
  499. batchLastRow = row
  500. } else {
  501. this.emit('row', row)
  502. }
  503. } else {
  504. this.emit('row', row)
  505. }
  506. } else {
  507. recordset.push(row)
  508. }
  509. })
  510. if (this._isBatch) {
  511. if (Object.keys(this.parameters).length) {
  512. for (const name in this.parameters) {
  513. if (!objectHasProperty(this.parameters, name)) {
  514. continue
  515. }
  516. const param = this.parameters[name]
  517. let value = getTediousType(param.type).validate(param.value)
  518. if (value instanceof TypeError) {
  519. value = new RequestError(`Validation failed for parameter '${name}'. ${value.message}`, 'EPARAM')
  520. this.parent.release(connection)
  521. return callback(value)
  522. }
  523. param.value = value
  524. }
  525. const declarations = []
  526. for (const name in this.parameters) {
  527. if (!objectHasProperty(this.parameters, name)) {
  528. continue
  529. }
  530. const param = this.parameters[name]
  531. declarations.push(`@${name} ${declare(param.type, param)}`)
  532. }
  533. const assigns = []
  534. for (const name in this.parameters) {
  535. if (!objectHasProperty(this.parameters, name)) {
  536. continue
  537. }
  538. const param = this.parameters[name]
  539. assigns.push(`@${name} = ${cast(param.value, param.type, param)}`)
  540. }
  541. const selects = []
  542. for (const name in this.parameters) {
  543. if (!objectHasProperty(this.parameters, name)) {
  544. continue
  545. }
  546. const param = this.parameters[name]
  547. if (param.io === 2) {
  548. selects.push(`@${name} as [${name}]`)
  549. }
  550. }
  551. batchHasOutput = selects.length > 0
  552. req.sqlTextOrProcedure = `declare ${declarations.join(', ')};select ${assigns.join(', ')};${req.sqlTextOrProcedure};${batchHasOutput ? (`select 1 as [___return___], ${selects.join(', ')}`) : ''}`
  553. }
  554. } else {
  555. for (const name in this.parameters) {
  556. if (!objectHasProperty(this.parameters, name)) {
  557. continue
  558. }
  559. const param = this.parameters[name]
  560. if (param.io === 1) {
  561. req.addParameter(param.name, getTediousType(param.type), parameterCorrection(param.value), { length: param.length, scale: param.scale, precision: param.precision })
  562. } else {
  563. req.addOutputParameter(param.name, getTediousType(param.type), parameterCorrection(param.value), { length: param.length, scale: param.scale, precision: param.precision })
  564. }
  565. }
  566. }
  567. try {
  568. connection[this._isBatch ? 'execSqlBatch' : 'execSql'](req)
  569. } catch (error) {
  570. handleError(true, connection, error)
  571. }
  572. })
  573. })
  574. }
  575. /*
  576. Execute stored procedure with specified parameters.
  577. */
  578. _execute (procedure, callback) {
  579. super._execute(procedure, err => {
  580. if (err) return callback(err)
  581. const recordsets = []
  582. const errors = []
  583. const errorHandlers = {}
  584. const output = {}
  585. const rowsAffected = []
  586. let columns = {}
  587. let recordset = []
  588. let returnValue = 0
  589. let isChunkedRecordset = false
  590. let chunksBuffer = null
  591. let hasReturned = false
  592. const handleError = (doReturn, connection, info) => {
  593. let err = new Error(info.message)
  594. err.info = info
  595. err = new RequestError(err, 'EREQUEST')
  596. if (this.stream) {
  597. this.emit('error', err)
  598. } else {
  599. if (doReturn && !hasReturned) {
  600. if (connection) {
  601. for (const event in errorHandlers) {
  602. connection.removeListener(event, errorHandlers[event])
  603. }
  604. this.parent.release(connection)
  605. }
  606. hasReturned = true
  607. callback(err)
  608. }
  609. }
  610. // we must collect errors even in stream mode
  611. errors.push(err)
  612. }
  613. const handleInfo = msg => {
  614. this.emit('info', {
  615. message: msg.message,
  616. number: msg.number,
  617. state: msg.state,
  618. class: msg.class,
  619. lineNumber: msg.lineNumber,
  620. serverName: msg.serverName,
  621. procName: msg.procName
  622. })
  623. }
  624. this.parent.acquire(this, (err, connection, config) => {
  625. if (err) return callback(err)
  626. debug('connection(%d): borrowed to request #%d', IDS.get(connection), IDS.get(this))
  627. let row
  628. if (this.canceled) {
  629. debug('request(%d): canceled', IDS.get(this))
  630. this.parent.release(connection)
  631. return callback(new RequestError('Canceled.', 'ECANCEL'))
  632. }
  633. this._cancel = () => {
  634. debug('request(%d): cancel', IDS.get(this))
  635. connection.cancel()
  636. }
  637. // attach handler to handle multiple error messages
  638. connection.on('infoMessage', errorHandlers.infoMessage = handleInfo)
  639. connection.on('errorMessage', errorHandlers.errorMessage = handleError.bind(null, false, connection))
  640. connection.on('error', errorHandlers.error = handleError.bind(null, true, connection))
  641. debug('request(%d): execute', IDS.get(this), procedure)
  642. const req = new tds.Request(procedure, err => {
  643. // to make sure we handle no-sql errors as well
  644. if (err && (!errors.length || (errors.length && err.message !== errors[errors.length - 1].message))) {
  645. err = new RequestError(err, 'EREQUEST')
  646. if (this.stream) this.emit('error', err)
  647. errors.push(err)
  648. }
  649. delete this._cancel
  650. let error
  651. if (errors.length && !this.stream) {
  652. error = errors.pop()
  653. error.precedingErrors = errors
  654. }
  655. if (!hasReturned) {
  656. for (const event in errorHandlers) {
  657. connection.removeListener(event, errorHandlers[event])
  658. }
  659. this.parent.release(connection)
  660. hasReturned = true
  661. if (error) {
  662. debug('request(%d): failed', IDS.get(this), error)
  663. } else {
  664. debug('request(%d): complete', IDS.get(this))
  665. }
  666. if (this.stream) {
  667. callback(null, null, output, returnValue, rowsAffected)
  668. } else {
  669. callback(error, recordsets, output, returnValue, rowsAffected)
  670. }
  671. }
  672. })
  673. this._setCurrentRequest(req)
  674. req.on('columnMetadata', metadata => {
  675. columns = createColumns(metadata)
  676. isChunkedRecordset = false
  677. if ((metadata.length === 1) && (metadata[0].colName === JSON_COLUMN_ID || metadata[0].colName === XML_COLUMN_ID)) {
  678. isChunkedRecordset = true
  679. chunksBuffer = []
  680. }
  681. if (this.stream) this.emit('recordset', columns)
  682. })
  683. req.on('row', columns => {
  684. if (!recordset) recordset = []
  685. if (isChunkedRecordset) {
  686. return chunksBuffer.push(columns[0].value)
  687. }
  688. row = {}
  689. for (const col of columns) {
  690. col.value = valueCorrection(col.value, col.metadata)
  691. const exi = row[col.metadata.colName]
  692. if (exi != null) {
  693. if (exi instanceof Array) {
  694. exi.push(col.value)
  695. } else {
  696. row[col.metadata.colName] = [exi, col.value]
  697. }
  698. } else {
  699. row[col.metadata.colName] = col.value
  700. }
  701. }
  702. if (this.stream) {
  703. this.emit('row', row)
  704. } else {
  705. recordset.push(row)
  706. }
  707. })
  708. req.on('doneInProc', (rowCount, more) => {
  709. if (rowCount != null) rowsAffected.push(rowCount)
  710. // filter empty recordsets when NOCOUNT is OFF
  711. if (Object.keys(columns).length === 0) return
  712. if (isChunkedRecordset) {
  713. if (columns[JSON_COLUMN_ID] && config.parseJSON === true) {
  714. try {
  715. if (chunksBuffer.length === 0) {
  716. row = null
  717. } else {
  718. row = JSON.parse(chunksBuffer.join(''))
  719. }
  720. } catch (ex) {
  721. row = null
  722. const ex2 = new RequestError(new Error(`Failed to parse incoming JSON. ${ex.message}`), 'EJSON')
  723. if (this.stream) this.emit('error', ex2)
  724. // we must collect errors even in stream mode
  725. errors.push(ex2)
  726. }
  727. } else {
  728. row = {}
  729. row[Object.keys(columns)[0]] = chunksBuffer.join('')
  730. }
  731. chunksBuffer = null
  732. if (this.stream) {
  733. this.emit('row', row)
  734. } else {
  735. recordset.push(row)
  736. }
  737. }
  738. if (!this.stream) {
  739. // all rows of current recordset loaded
  740. Object.defineProperty(recordset, 'columns', {
  741. enumerable: false,
  742. configurable: true,
  743. value: columns
  744. })
  745. Object.defineProperty(recordset, 'toTable', {
  746. enumerable: false,
  747. configurable: true,
  748. value (name) { return Table.fromRecordset(this, name) }
  749. })
  750. recordsets.push(recordset)
  751. }
  752. recordset = []
  753. columns = {}
  754. })
  755. req.on('doneProc', (rowCount, more, returnStatus) => {
  756. returnValue = returnStatus
  757. })
  758. req.on('returnValue', (parameterName, value, metadata) => {
  759. output[parameterName] = value
  760. })
  761. for (const name in this.parameters) {
  762. if (!objectHasProperty(this.parameters, name)) {
  763. continue
  764. }
  765. const param = this.parameters[name]
  766. if (param.io === 1) {
  767. req.addParameter(param.name, getTediousType(param.type), parameterCorrection(param.value), { length: param.length, scale: param.scale, precision: param.precision })
  768. } else {
  769. req.addOutputParameter(param.name, getTediousType(param.type), parameterCorrection(param.value), { length: param.length, scale: param.scale, precision: param.precision })
  770. }
  771. }
  772. connection.callProcedure(req)
  773. })
  774. })
  775. }
  776. _pause () {
  777. super._pause()
  778. if (this._currentRequest) {
  779. this._currentRequest.pause()
  780. }
  781. }
  782. _resume () {
  783. super._resume()
  784. if (this._currentRequest) {
  785. this._currentRequest.resume()
  786. }
  787. }
  788. }
  789. module.exports = Request