'use strict' /** * Copyright (c) 2010-2017 Brian Carlson (brian.m.carlson@gmail.com) * All rights reserved. * * This source code is licensed under the MIT license found in the * README.md file in the root directory of this source tree. */ const { EventEmitter } = require('events') const Result = require('./result') const utils = require('./utils') class Query extends EventEmitter { constructor(config, values, callback) { super() config = utils.normalizeQueryConfig(config, values, callback) this.text = config.text this.values = config.values this.rows = config.rows this.types = config.types this.name = config.name this.binary = config.binary // use unique portal name each time this.portal = config.portal || '' this.callback = config.callback this._rowMode = config.rowMode if (process.domain && config.callback) { this.callback = process.domain.bind(config.callback) } this._result = new Result(this._rowMode, this.types) // potential for multiple results this._results = this._result this.isPreparedStatement = false this._canceledDueToError = false this._promise = null } requiresPreparation() { // named queries must always be prepared if (this.name) { return true } // always prepare if there are max number of rows expected per // portal execution if (this.rows) { return true } // don't prepare empty text queries if (!this.text) { return false } // prepare if there are values if (!this.values) { return false } return this.values.length > 0 } _checkForMultirow() { // if we already have a result with a command property // then we've already executed one query in a multi-statement simple query // turn our results into an array of results if (this._result.command) { if (!Array.isArray(this._results)) { this._results = [this._result] } this._result = new Result(this._rowMode, this.types) this._results.push(this._result) } } // associates row metadata from the supplied // message with this query object // metadata used when parsing row results handleRowDescription(msg) { this._checkForMultirow() this._result.addFields(msg.fields) this._accumulateRows = this.callback || !this.listeners('row').length } handleDataRow(msg) { let row if (this._canceledDueToError) { return } try { row = this._result.parseRow(msg.fields) } catch (err) { this._canceledDueToError = err return } this.emit('row', row, this._result) if (this._accumulateRows) { this._result.addRow(row) } } handleCommandComplete(msg, con) { this._checkForMultirow() this._result.addCommandComplete(msg) // need to sync after each command complete of a prepared statement if (this.isPreparedStatement) { con.sync() } } // if a named prepared statement is created with empty query text // the backend will send an emptyQuery message but *not* a command complete message // execution on the connection will hang until the backend receives a sync message handleEmptyQuery(con) { if (this.isPreparedStatement) { con.sync() } } handleReadyForQuery(con) { if (this._canceledDueToError) { return this.handleError(this._canceledDueToError, con) } if (this.callback) { this.callback(null, this._results) } this.emit('end', this._results) } handleError(err, connection) { // need to sync after error during a prepared statement if (this.isPreparedStatement) { connection.sync() } if (this._canceledDueToError) { err = this._canceledDueToError this._canceledDueToError = false } // if callback supplied do not emit error event as uncaught error // events will bubble up to node process if (this.callback) { return this.callback(err) } this.emit('error', err) } submit(connection) { if (typeof this.text !== 'string' && typeof this.name !== 'string') { return new Error('A query must have either text or a name. Supplying neither is unsupported.') } const previous = connection.parsedStatements[this.name] if (this.text && previous && this.text !== previous) { return new Error(`Prepared statements must be unique - '${this.name}' was used for a different statement`) } if (this.values && !Array.isArray(this.values)) { return new Error('Query values must be an array') } if (this.requiresPreparation()) { this.prepare(connection) } else { connection.query(this.text) } return null } hasBeenParsed(connection) { return this.name && connection.parsedStatements[this.name] } handlePortalSuspended(connection) { this._getRows(connection, this.rows) } _getRows(connection, rows) { connection.execute( { portal: this.portal, rows: rows, }, true ) connection.flush() } prepare(connection) { // prepared statements need sync to be called after each command // complete or when an error is encountered this.isPreparedStatement = true // TODO refactor this poor encapsulation if (!this.hasBeenParsed(connection)) { connection.parse( { text: this.text, name: this.name, types: this.types, }, true ) } if (this.values) { try { this.values = this.values.map(utils.prepareValue) } catch (err) { this.handleError(err, connection) return } } // http://developer.postgresql.org/pgdocs/postgres/protocol-flow.html#PROTOCOL-FLOW-EXT-QUERY connection.bind( { portal: this.portal, statement: this.name, values: this.values, binary: this.binary, }, true ) connection.describe( { type: 'P', name: this.portal || '', }, true ) this._getRows(connection, this.rows) } handleCopyInResponse(connection) { connection.sendCopyFail('No source stream defined') } // eslint-disable-next-line no-unused-vars handleCopyData(msg, connection) { // noop } } module.exports = Query