node-query/node_modules/pg/lib/connection.js

630 lines
16 KiB
JavaScript
Raw Normal View History

var net = require('net');
var EventEmitter = require('events').EventEmitter;
var util = require('util');
var Writer = require('buffer-writer');
var Reader = require('packet-reader');
var TEXT_MODE = 0;
var BINARY_MODE = 1;
var Connection = function(config) {
EventEmitter.call(this);
config = config || {};
this.stream = config.stream || new net.Stream();
this.lastBuffer = false;
this.lastOffset = 0;
this.buffer = null;
this.offset = null;
this.encoding = 'utf8';
this.parsedStatements = {};
this.writer = new Writer();
this.ssl = config.ssl || false;
this._ending = false;
this._mode = TEXT_MODE;
this._emitMessage = false;
this._reader = new Reader({
headerSize: 1,
lengthPadding: -4
});
var self = this;
this.on('newListener', function(eventName) {
if(eventName == 'message') {
self._emitMessage = true;
}
});
};
util.inherits(Connection, EventEmitter);
Connection.prototype.connect = function(port, host) {
if(this.stream.readyState === 'closed') {
this.stream.connect(port, host);
} else if(this.stream.readyState == 'open') {
this.emit('connect');
}
var self = this;
this.stream.on('connect', function() {
self.emit('connect');
});
this.stream.on('error', function(error) {
//don't raise ECONNRESET errors - they can & should be ignored
//during disconnect
if(self._ending && error.code == 'ECONNRESET') {
return;
}
self.emit('error', error);
});
this.stream.on('close', function() {
// NOTE: node-0.10 emits both 'end' and 'close'
// for streams closed by the peer, while
// node-0.8 only emits 'close'
self.emit('end');
});
if(!this.ssl) {
return this.attachListeners(this.stream);
}
this.stream.once('data', function(buffer) {
var responseCode = buffer.toString('utf8');
if(responseCode != 'S') {
return self.emit('error', new Error('The server does not support SSL connections'));
}
var tls = require('tls');
self.stream = tls.connect({
socket: self.stream,
servername: host,
rejectUnauthorized: self.ssl.rejectUnauthorized,
ca: self.ssl.ca,
pfx: self.ssl.pfx,
key: self.ssl.key,
passphrase: self.ssl.passphrase,
cert: self.ssl.cert,
NPNProtocols: self.ssl.NPNProtocols
});
self.attachListeners(self.stream);
self.emit('sslconnect');
self.stream.on('error', function(error){
self.emit('error', error);
});
});
};
Connection.prototype.attachListeners = function(stream) {
var self = this;
stream.on('data', function(buff) {
self._reader.addChunk(buff);
var packet = self._reader.read();
while(packet) {
var msg = self.parseMessage(packet);
if(self._emitMessage) {
self.emit('message', msg);
}
self.emit(msg.name, msg);
packet = self._reader.read();
}
});
};
Connection.prototype.requestSsl = function() {
this.checkSslResponse = true;
var bodyBuffer = this.writer
.addInt16(0x04D2)
.addInt16(0x162F).flush();
var length = bodyBuffer.length + 4;
var buffer = new Writer()
.addInt32(length)
.add(bodyBuffer)
.join();
this.stream.write(buffer);
};
Connection.prototype.startup = function(config) {
var writer = this.writer
.addInt16(3)
.addInt16(0)
;
Object.keys(config).forEach(function(key){
var val = config[key];
writer.addCString(key).addCString(val);
});
writer.addCString('client_encoding').addCString("'utf-8'");
var bodyBuffer = writer.addCString('').flush();
//this message is sent without a code
var length = bodyBuffer.length + 4;
var buffer = new Writer()
.addInt32(length)
.add(bodyBuffer)
.join();
this.stream.write(buffer);
};
Connection.prototype.cancel = function(processID, secretKey) {
var bodyBuffer = this.writer
.addInt16(1234)
.addInt16(5678)
.addInt32(processID)
.addInt32(secretKey)
.addCString('').flush();
var length = bodyBuffer.length + 4;
var buffer = new Writer()
.addInt32(length)
.add(bodyBuffer)
.join();
this.stream.write(buffer);
};
Connection.prototype.password = function(password) {
//0x70 = 'p'
this._send(0x70, this.writer.addCString(password));
};
Connection.prototype._send = function(code, more) {
if(!this.stream.writable) { return false; }
if(more === true) {
this.writer.addHeader(code);
} else {
return this.stream.write(this.writer.flush(code));
}
};
Connection.prototype.query = function(text) {
//0x51 = Q
this.stream.write(this.writer.addCString(text).flush(0x51));
};
//send parse message
//"more" === true to buffer the message until flush() is called
Connection.prototype.parse = function(query, more) {
//expect something like this:
// { name: 'queryName',
// text: 'select * from blah',
// types: ['int8', 'bool'] }
//normalize missing query names to allow for null
query.name = query.name || '';
//normalize null type array
query.types = query.types || [];
var len = query.types.length;
var buffer = this.writer
.addCString(query.name) //name of query
.addCString(query.text) //actual query text
.addInt16(len);
for(var i = 0; i < len; i++) {
buffer.addInt32(query.types[i]);
}
var code = 0x50;
this._send(code, more);
};
//send bind message
//"more" === true to buffer the message until flush() is called
Connection.prototype.bind = function(config, more) {
//normalize config
config = config || {};
config.portal = config.portal || '';
config.statement = config.statement || '';
config.binary = config.binary || false;
var values = config.values || [];
var len = values.length;
var useBinary = false;
for (var j = 0; j < len; j++)
useBinary |= values[j] instanceof Buffer;
var buffer = this.writer
.addCString(config.portal)
.addCString(config.statement);
if (!useBinary)
buffer.addInt16(0);
else {
buffer.addInt16(len);
for (j = 0; j < len; j++)
buffer.addInt16(values[j] instanceof Buffer);
}
buffer.addInt16(len);
for(var i = 0; i < len; i++) {
var val = values[i];
if(val === null || typeof val === "undefined") {
buffer.addInt32(-1);
} else if (val instanceof Buffer) {
buffer.addInt32(val.length);
buffer.add(val);
} else {
buffer.addInt32(Buffer.byteLength(val));
buffer.addString(val);
}
}
if(config.binary) {
buffer.addInt16(1); // format codes to use binary
buffer.addInt16(1);
}
else {
buffer.addInt16(0); // format codes to use text
}
//0x42 = 'B'
this._send(0x42, more);
};
//send execute message
//"more" === true to buffer the message until flush() is called
Connection.prototype.execute = function(config, more) {
config = config || {};
config.portal = config.portal || '';
config.rows = config.rows || '';
this.writer
.addCString(config.portal)
.addInt32(config.rows);
//0x45 = 'E'
this._send(0x45, more);
};
var emptyBuffer = Buffer(0);
Connection.prototype.flush = function() {
//0x48 = 'H'
this.writer.add(emptyBuffer);
this._send(0x48);
};
Connection.prototype.sync = function() {
//clear out any pending data in the writer
this.writer.flush(0);
this.writer.add(emptyBuffer);
this._ending = true;
this._send(0x53);
};
Connection.prototype.end = function() {
//0x58 = 'X'
this.writer.add(emptyBuffer);
this._ending = true;
this._send(0x58);
};
Connection.prototype.close = function(msg, more) {
this.writer.addCString(msg.type + (msg.name || ''));
this._send(0x43, more);
};
Connection.prototype.describe = function(msg, more) {
this.writer.addCString(msg.type + (msg.name || ''));
this._send(0x44, more);
};
Connection.prototype.sendCopyFromChunk = function (chunk) {
this.stream.write(this.writer.add(chunk).flush(0x64));
};
Connection.prototype.endCopyFrom = function () {
this.stream.write(this.writer.add(emptyBuffer).flush(0x63));
};
Connection.prototype.sendCopyFail = function (msg) {
//this.stream.write(this.writer.add(emptyBuffer).flush(0x66));
this.writer.addCString(msg);
this._send(0x66);
};
var Message = function(name, length) {
this.name = name;
this.length = length;
};
Connection.prototype.parseMessage = function(buffer) {
this.offset = 0;
var length = buffer.length + 4;
switch(this._reader.header)
{
case 0x52: //R
return this.parseR(buffer, length);
case 0x53: //S
return this.parseS(buffer, length);
case 0x4b: //K
return this.parseK(buffer, length);
case 0x43: //C
return this.parseC(buffer, length);
case 0x5a: //Z
return this.parseZ(buffer, length);
case 0x54: //T
return this.parseT(buffer, length);
case 0x44: //D
return this.parseD(buffer, length);
case 0x45: //E
return this.parseE(buffer, length);
case 0x4e: //N
return this.parseN(buffer, length);
case 0x31: //1
return new Message('parseComplete', length);
case 0x32: //2
return new Message('bindComplete', length);
case 0x33: //3
return new Message('closeComplete', length);
case 0x41: //A
return this.parseA(buffer, length);
case 0x6e: //n
return new Message('noData', length);
case 0x49: //I
return new Message('emptyQuery', length);
case 0x73: //s
return new Message('portalSuspended', length);
case 0x47: //G
return this.parseG(buffer, length);
case 0x48: //H
return this.parseH(buffer, length);
case 0x63: //c
return new Message('copyDone', length);
case 0x64: //d
return this.parsed(buffer, length);
}
};
Connection.prototype.parseR = function(buffer, length) {
var code = 0;
var msg = new Message('authenticationOk', length);
if(msg.length === 8) {
code = this.parseInt32(buffer);
if(code === 3) {
msg.name = 'authenticationCleartextPassword';
}
return msg;
}
if(msg.length === 12) {
code = this.parseInt32(buffer);
if(code === 5) { //md5 required
msg.name = 'authenticationMD5Password';
msg.salt = new Buffer(4);
buffer.copy(msg.salt, 0, this.offset, this.offset + 4);
this.offset += 4;
return msg;
}
}
throw new Error("Unknown authenticatinOk message type" + util.inspect(msg));
};
Connection.prototype.parseS = function(buffer, length) {
var msg = new Message('parameterStatus', length);
msg.parameterName = this.parseCString(buffer);
msg.parameterValue = this.parseCString(buffer);
return msg;
};
Connection.prototype.parseK = function(buffer, length) {
var msg = new Message('backendKeyData', length);
msg.processID = this.parseInt32(buffer);
msg.secretKey = this.parseInt32(buffer);
return msg;
};
Connection.prototype.parseC = function(buffer, length) {
var msg = new Message('commandComplete', length);
msg.text = this.parseCString(buffer);
return msg;
};
Connection.prototype.parseZ = function(buffer, length) {
var msg = new Message('readyForQuery', length);
msg.name = 'readyForQuery';
msg.status = this.readString(buffer, 1);
return msg;
};
var ROW_DESCRIPTION = 'rowDescription';
Connection.prototype.parseT = function(buffer, length) {
var msg = new Message(ROW_DESCRIPTION, length);
msg.fieldCount = this.parseInt16(buffer);
var fields = [];
for(var i = 0; i < msg.fieldCount; i++){
fields.push(this.parseField(buffer));
}
msg.fields = fields;
return msg;
};
var Field = function() {
this.name = null;
this.tableID = null;
this.columnID = null;
this.dataTypeID = null;
this.dataTypeSize = null;
this.dataTypeModifier = null;
this.format = null;
};
var FORMAT_TEXT = 'text';
var FORMAT_BINARY = 'binary';
Connection.prototype.parseField = function(buffer) {
var field = new Field();
field.name = this.parseCString(buffer);
field.tableID = this.parseInt32(buffer);
field.columnID = this.parseInt16(buffer);
field.dataTypeID = this.parseInt32(buffer);
field.dataTypeSize = this.parseInt16(buffer);
field.dataTypeModifier = this.parseInt32(buffer);
if(this.parseInt16(buffer) === TEXT_MODE) {
this._mode = TEXT_MODE;
field.format = FORMAT_TEXT;
} else {
this._mode = BINARY_MODE;
field.format = FORMAT_BINARY;
}
return field;
};
var DATA_ROW = 'dataRow';
var DataRowMessage = function(length, fieldCount) {
this.name = DATA_ROW;
this.length = length;
this.fieldCount = fieldCount;
this.fields = [];
};
//extremely hot-path code
Connection.prototype.parseD = function(buffer, length) {
var fieldCount = this.parseInt16(buffer);
var msg = new DataRowMessage(length, fieldCount);
for(var i = 0; i < fieldCount; i++) {
msg.fields.push(this._readValue(buffer));
}
return msg;
};
//extremely hot-path code
Connection.prototype._readValue = function(buffer) {
var length = this.parseInt32(buffer);
if(length === -1) return null;
if(this._mode === TEXT_MODE) {
return this.readString(buffer, length);
}
return this.readBytes(buffer, length);
};
//parses error
Connection.prototype.parseE = function(buffer, length) {
var fields = {};
var msg, item;
var input = new Message('error', length);
var fieldType = this.readString(buffer, 1);
while(fieldType != '\0') {
fields[fieldType] = this.parseCString(buffer);
fieldType = this.readString(buffer, 1);
}
if(input.name === 'error') {
// the msg is an Error instance
msg = new Error(fields.M);
for (item in input) {
// copy input properties to the error
if(input.hasOwnProperty(item)) {
msg[item] = input[item];
}
}
} else {
// the msg is an object literal
msg = input;
msg.message = fields.M;
}
msg.severity = fields.S;
msg.code = fields.C;
msg.detail = fields.D;
msg.hint = fields.H;
msg.position = fields.P;
msg.internalPosition = fields.p;
msg.internalQuery = fields.q;
msg.where = fields.W;
msg.file = fields.F;
msg.line = fields.L;
msg.routine = fields.R;
return msg;
};
//same thing, different name
Connection.prototype.parseN = function(buffer, length) {
var msg = this.parseE(buffer, length);
msg.name = 'notice';
return msg;
};
Connection.prototype.parseA = function(buffer, length) {
var msg = new Message('notification', length);
msg.processId = this.parseInt32(buffer);
msg.channel = this.parseCString(buffer);
msg.payload = this.parseCString(buffer);
return msg;
};
Connection.prototype.parseG = function (buffer, length) {
var msg = new Message('copyInResponse', length);
return this.parseGH(buffer, msg);
};
Connection.prototype.parseH = function(buffer, length) {
var msg = new Message('copyOutResponse', length);
return this.parseGH(buffer, msg);
};
Connection.prototype.parseGH = function (buffer, msg) {
var isBinary = buffer[this.offset] !== 0;
this.offset++;
msg.binary = isBinary;
var columnCount = this.parseInt16(buffer);
msg.columnTypes = [];
for(var i = 0; i<columnCount; i++) {
msg.columnTypes.push(this.parseInt16(buffer));
}
return msg;
};
Connection.prototype.parsed = function (buffer, length) {
var msg = new Message('copyData', length);
msg.chunk = this.readBytes(buffer, msg.length - 4);
return msg;
};
Connection.prototype.parseInt32 = function(buffer) {
var value = buffer.readInt32BE(this.offset, true);
this.offset += 4;
return value;
};
Connection.prototype.parseInt16 = function(buffer) {
var value = buffer.readInt16BE(this.offset, true);
this.offset += 2;
return value;
};
Connection.prototype.readString = function(buffer, length) {
return buffer.toString(this.encoding, this.offset, (this.offset += length));
};
Connection.prototype.readBytes = function(buffer, length) {
return buffer.slice(this.offset, this.offset += length);
};
Connection.prototype.parseCString = function(buffer) {
var start = this.offset;
while(buffer[this.offset++] !== 0) { }
return buffer.toString(this.encoding, start, this.offset - 1);
};
//end parsing methods
module.exports = Connection;