| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587 |
- 'use strict'
- const debug = require('debug')('mssql:base')
- const { EventEmitter } = require('events')
- const { IDS, objectHasProperty } = require('../utils')
- const globalConnection = require('../global-connection')
- const { RequestError, ConnectionError } = require('../error')
- const { TYPES } = require('../datatypes')
- const shared = require('../shared')
- /**
- * Class Request.
- *
- * @property {Transaction} transaction Reference to transaction when request was created in transaction.
- * @property {*} parameters Collection of input and output parameters.
- * @property {Boolean} canceled `true` if request was canceled.
- *
- * @fires Request#recordset
- * @fires Request#row
- * @fires Request#done
- * @fires Request#error
- */
- class Request extends EventEmitter {
- /**
- * Create new Request.
- *
- * @param {Connection|ConnectionPool|Transaction|PreparedStatement} parent If ommited, global connection is used instead.
- */
- constructor (parent) {
- super()
- IDS.add(this, 'Request')
- debug('request(%d): created', IDS.get(this))
- this.canceled = false
- this._paused = false
- this.parent = parent || globalConnection.pool
- this.parameters = {}
- }
- /**
- * Fetch request from tagged template string.
- *
- * @private
- * @param {Array} strings
- * @param {Array} values
- * @param {String} [method] If provided, method is automtically called with serialized command on this object.
- * @return {Request}
- */
- _template (strings, values, method) {
- const command = [strings[0]]
- for (let index = 0; index < values.length; index++) {
- const value = values[index]
- // if value is an array, prepare each items as it's own comma separated parameter
- if (Array.isArray(value)) {
- for (let parameterIndex = 0; parameterIndex < value.length; parameterIndex++) {
- this.input(`param${index + 1}_${parameterIndex}`, value[parameterIndex])
- command.push(`@param${index + 1}_${parameterIndex}`)
- if (parameterIndex < value.length - 1) {
- command.push(', ')
- } else {
- command.push(strings[index + 1])
- }
- }
- } else {
- this.input(`param${index + 1}`, value)
- command.push(`@param${index + 1}`, strings[index + 1])
- }
- }
- if (method) {
- return this[method](command.join(''))
- } else {
- return command.join('')
- }
- }
- /**
- * Add an input parameter to the request.
- *
- * @param {String} name Name of the input parameter without @ char.
- * @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.
- * @param {*} value Input parameter value. `undefined` and `NaN` values are automatically converted to `null` values.
- * @return {Request}
- */
- input (name, type, value) {
- if ((/(--| |\/\*|\*\/|')/).test(name)) {
- throw new RequestError(`SQL injection warning for param '${name}'`, 'EINJECT')
- }
- if (arguments.length < 2) {
- throw new RequestError('Invalid number of arguments. At least 2 arguments expected.', 'EARGS')
- } else if (arguments.length === 2) {
- value = type
- type = shared.getTypeByValue(value)
- }
- // support for custom data types
- if (value && typeof value.valueOf === 'function' && !(value instanceof Date)) value = value.valueOf()
- if (value === undefined) value = null // undefined to null
- if (typeof value === 'number' && isNaN(value)) value = null // NaN to null
- if (type instanceof Function) type = type()
- if (objectHasProperty(this.parameters, name)) {
- throw new RequestError(`The parameter name ${name} has already been declared. Parameter names must be unique`, 'EDUPEPARAM')
- }
- this.parameters[name] = {
- name,
- type: type.type,
- io: 1,
- value,
- length: type.length,
- scale: type.scale,
- precision: type.precision,
- tvpType: type.tvpType
- }
- return this
- }
- /**
- * Replace an input parameter on the request.
- *
- * @param {String} name Name of the input parameter without @ char.
- * @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.
- * @param {*} value Input parameter value. `undefined` and `NaN` values are automatically converted to `null` values.
- * @return {Request}
- */
- replaceInput (name, type, value) {
- delete this.parameters[name]
- return this.input(name, type, value)
- }
- /**
- * Add an output parameter to the request.
- *
- * @param {String} name Name of the output parameter without @ char.
- * @param {*} type SQL data type of output parameter.
- * @param {*} [value] Output parameter value initial value. `undefined` and `NaN` values are automatically converted to `null` values. Optional.
- * @return {Request}
- */
- output (name, type, value) {
- if (!type) { type = TYPES.NVarChar }
- if ((/(--| |\/\*|\*\/|')/).test(name)) {
- throw new RequestError(`SQL injection warning for param '${name}'`, 'EINJECT')
- }
- if ((type === TYPES.Text) || (type === TYPES.NText) || (type === TYPES.Image)) {
- throw new RequestError('Deprecated types (Text, NText, Image) are not supported as OUTPUT parameters.', 'EDEPRECATED')
- }
- // support for custom data types
- if (value && typeof value.valueOf === 'function' && !(value instanceof Date)) value = value.valueOf()
- if (value === undefined) value = null // undefined to null
- if (typeof value === 'number' && isNaN(value)) value = null // NaN to null
- if (type instanceof Function) type = type()
- if (objectHasProperty(this.parameters, name)) {
- throw new RequestError(`The parameter name ${name} has already been declared. Parameter names must be unique`, 'EDUPEPARAM')
- }
- this.parameters[name] = {
- name,
- type: type.type,
- io: 2,
- value,
- length: type.length,
- scale: type.scale,
- precision: type.precision
- }
- return this
- }
- /**
- * Replace an output parameter on the request.
- *
- * @param {String} name Name of the output parameter without @ char.
- * @param {*} type SQL data type of output parameter.
- * @param {*} [value] Output parameter value initial value. `undefined` and `NaN` values are automatically converted to `null` values. Optional.
- * @return {Request}
- */
- replaceOutput (name, type, value) {
- delete this.parameters[name]
- return this.output(name, type, value)
- }
- /**
- * Execute the SQL batch.
- *
- * @param {String} batch T-SQL batch to be executed.
- * @param {Request~requestCallback} [callback] A callback which is called after execution has completed, or an error has occurred. If omited, method returns Promise.
- * @return {Request|Promise}
- */
- batch (batch, callback) {
- if (this.stream == null && this.connection) this.stream = this.connection.config.stream
- this.rowsAffected = 0
- if (typeof callback === 'function') {
- this._batch(batch, (err, recordsets, output, rowsAffected) => {
- if (this.stream) {
- if (err) this.emit('error', err)
- err = null
- this.emit('done', {
- output,
- rowsAffected
- })
- }
- if (err) return callback(err)
- callback(null, {
- recordsets,
- recordset: recordsets && recordsets[0],
- output,
- rowsAffected
- })
- })
- return this
- }
- // Check is method was called as tagged template
- if (typeof batch === 'object') {
- const values = Array.prototype.slice.call(arguments)
- const strings = values.shift()
- batch = this._template(strings, values)
- }
- return new shared.Promise((resolve, reject) => {
- this._batch(batch, (err, recordsets, output, rowsAffected) => {
- if (this.stream) {
- if (err) this.emit('error', err)
- err = null
- this.emit('done', {
- output,
- rowsAffected
- })
- }
- if (err) return reject(err)
- resolve({
- recordsets,
- recordset: recordsets && recordsets[0],
- output,
- rowsAffected
- })
- })
- })
- }
- /**
- * @private
- * @param {String} batch
- * @param {Request~requestCallback} callback
- */
- _batch (batch, callback) {
- if (!this.connection) {
- return setImmediate(callback, new RequestError('No connection is specified for that request.', 'ENOCONN'))
- }
- if (!this.connection.connected) {
- return setImmediate(callback, new ConnectionError('Connection is closed.', 'ECONNCLOSED'))
- }
- this.canceled = false
- setImmediate(callback)
- }
- /**
- * Bulk load.
- *
- * @param {Table} table SQL table.
- * @param {object} [options] Options to be passed to the underlying driver (tedious only).
- * @param {Request~bulkCallback} [callback] A callback which is called after bulk load has completed, or an error has occurred. If omited, method returns Promise.
- * @return {Request|Promise}
- */
- bulk (table, options, callback) {
- if (typeof options === 'function') {
- callback = options
- options = {}
- } else if (typeof options === 'undefined') {
- options = {}
- }
- if (this.stream == null && this.connection) this.stream = this.connection.config.stream
- if (this.stream || typeof callback === 'function') {
- this._bulk(table, options, (err, rowsAffected) => {
- if (this.stream) {
- if (err) this.emit('error', err)
- return this.emit('done', {
- rowsAffected
- })
- }
- if (err) return callback(err)
- callback(null, {
- rowsAffected
- })
- })
- return this
- }
- return new shared.Promise((resolve, reject) => {
- this._bulk(table, options, (err, rowsAffected) => {
- if (err) return reject(err)
- resolve({
- rowsAffected
- })
- })
- })
- }
- /**
- * @private
- * @param {Table} table
- * @param {object} options
- * @param {Request~bulkCallback} callback
- */
- _bulk (table, options, callback) {
- if (!this.parent) {
- return setImmediate(callback, new RequestError('No connection is specified for that request.', 'ENOCONN'))
- }
- if (!this.parent.connected) {
- return setImmediate(callback, new ConnectionError('Connection is closed.', 'ECONNCLOSED'))
- }
- this.canceled = false
- setImmediate(callback)
- }
- /**
- * Sets request to `stream` mode and pulls all rows from all recordsets to a given stream.
- *
- * @param {Stream} stream Stream to pipe data into.
- * @return {Stream}
- */
- pipe (stream) {
- this.stream = true
- this.on('row', stream.write.bind(stream))
- this.on('error', stream.emit.bind(stream, 'error'))
- this.on('done', () => {
- setImmediate(() => stream.end())
- })
- stream.emit('pipe', this)
- return stream
- }
- /**
- * Execute the SQL command.
- *
- * @param {String} command T-SQL command to be executed.
- * @param {Request~requestCallback} [callback] A callback which is called after execution has completed, or an error has occurred. If omited, method returns Promise.
- * @return {Request|Promise}
- */
- query (command, callback) {
- if (this.stream == null && this.connection) this.stream = this.connection.config.stream
- this.rowsAffected = 0
- if (typeof callback === 'function') {
- this._query(command, (err, recordsets, output, rowsAffected) => {
- if (this.stream) {
- if (err) this.emit('error', err)
- err = null
- this.emit('done', {
- output,
- rowsAffected
- })
- }
- if (err) return callback(err)
- callback(null, {
- recordsets,
- recordset: recordsets && recordsets[0],
- output,
- rowsAffected
- })
- })
- return this
- }
- // Check is method was called as tagged template
- if (typeof command === 'object') {
- const values = Array.prototype.slice.call(arguments)
- const strings = values.shift()
- command = this._template(strings, values)
- }
- return new shared.Promise((resolve, reject) => {
- this._query(command, (err, recordsets, output, rowsAffected) => {
- if (this.stream) {
- if (err) this.emit('error', err)
- err = null
- this.emit('done', {
- output,
- rowsAffected
- })
- }
- if (err) return reject(err)
- resolve({
- recordsets,
- recordset: recordsets && recordsets[0],
- output,
- rowsAffected
- })
- })
- })
- }
- /**
- * @private
- * @param {String} command
- * @param {Request~bulkCallback} callback
- */
- _query (command, callback) {
- if (!this.parent) {
- return setImmediate(callback, new RequestError('No connection is specified for that request.', 'ENOCONN'))
- }
- if (!this.parent.connected) {
- return setImmediate(callback, new ConnectionError('Connection is closed.', 'ECONNCLOSED'))
- }
- this.canceled = false
- setImmediate(callback)
- }
- /**
- * Call a stored procedure.
- *
- * @param {String} procedure Name of the stored procedure to be executed.
- * @param {Request~requestCallback} [callback] A callback which is called after execution has completed, or an error has occurred. If omited, method returns Promise.
- * @return {Request|Promise}
- */
- execute (command, callback) {
- if (this.stream == null && this.connection) this.stream = this.connection.config.stream
- this.rowsAffected = 0
- if (typeof callback === 'function') {
- this._execute(command, (err, recordsets, output, returnValue, rowsAffected) => {
- if (this.stream) {
- if (err) this.emit('error', err)
- err = null
- this.emit('done', {
- output,
- rowsAffected,
- returnValue
- })
- }
- if (err) return callback(err)
- callback(null, {
- recordsets,
- recordset: recordsets && recordsets[0],
- output,
- rowsAffected,
- returnValue
- })
- })
- return this
- }
- return new shared.Promise((resolve, reject) => {
- this._execute(command, (err, recordsets, output, returnValue, rowsAffected) => {
- if (this.stream) {
- if (err) this.emit('error', err)
- err = null
- this.emit('done', {
- output,
- rowsAffected,
- returnValue
- })
- }
- if (err) return reject(err)
- resolve({
- recordsets,
- recordset: recordsets && recordsets[0],
- output,
- rowsAffected,
- returnValue
- })
- })
- })
- }
- /**
- * @private
- * @param {String} procedure
- * @param {Request~bulkCallback} callback
- */
- _execute (procedure, callback) {
- if (!this.parent) {
- return setImmediate(callback, new RequestError('No connection is specified for that request.', 'ENOCONN'))
- }
- if (!this.parent.connected) {
- return setImmediate(callback, new ConnectionError('Connection is closed.', 'ECONNCLOSED'))
- }
- this.canceled = false
- setImmediate(callback)
- }
- /**
- * Cancel currently executed request.
- *
- * @return {Boolean}
- */
- cancel () {
- this._cancel()
- return true
- }
- /**
- * @private
- */
- _cancel () {
- this.canceled = true
- }
- pause () {
- if (this.stream) {
- this._pause()
- return true
- }
- return false
- }
- _pause () {
- this._paused = true
- }
- resume () {
- if (this.stream) {
- this._resume()
- return true
- }
- return false
- }
- _resume () {
- this._paused = false
- }
- _setCurrentRequest (request) {
- this._currentRequest = request
- if (this._paused) {
- this.pause()
- }
- return this
- }
- }
- module.exports = Request
|