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

511 lines
16 KiB
JavaScript

var net = require('net');
var util = require('util');
var EventEmitter = require('events').EventEmitter;
var Queue = require('fastqueue');
var PacketParser = require('./packet_parser');
var Packet = require('./packets/packet');
var Packets = require('./packets/index.js');
var Commands = require('./commands/index.js');
var SqlString = require('./sql_string');
var _connectionId = 0;
function Connection(opts)
{
EventEmitter.call(this);
this.config = opts.config;
// TODO: fill defaults
// if no params, connect to /var/lib/mysql/mysql.sock ( /tmp/mysql.sock on OSX )
// if host is given, connect to host:3306
// TODO: use `/usr/local/mysql/bin/mysql_config --socket` output? as default socketPath
// if there is no host/port and no socketPath parameters?
if (!opts.config.stream) {
if (opts.config.socketPath)
this.stream = net.connect(opts.config.socketPath);
else
this.stream = net.connect(opts.config.port, opts.config.host);
} else {
this.stream = opts.config.stream;
}
this._internalId = _connectionId++;
this._commands = new Queue();
this._command = null;
this.statements = {};
// TODO: make it lru cache
// https://github.com/mercadolibre/node-simple-lru-cache
// or https://github.com/rsms/js-lru
// or https://github.com/monsur/jscache
// or https://github.com/isaacs/node-lru-cache
//
// key is field.name + ':' + field.columnType + ':' field.flags + '/'
this.textProtocolParsers = {};
// TODO: not sure if cache should be separate (same key as with textProtocolParsers)
// or part of prepared statements cache (key is sql query)
this.binaryProtocolParsers = {};
this.serverCapabilityFlags = 0;
this.authorized = false;
var connection = this;
this.sequenceId = 0;
this.stream.on('error', function(err) {
connection.emit('error', err);
});
// big TODO: benchmark if it all worth using 'ondata' and onPacket callbacks directly
// compositing streams would be much more easier.
// also, look for existing length-prefixed streams to reuse instead of packet_parser
// https://github.com/squaremo/node-spb - currently only fixed 4 byte prefix
// ...?
// see https://gist.github.com/khoomeister/4985691#use-that-instead-of-bind
this.packetParser = new PacketParser(function(p) { connection.handlePacket(p) });
// TODO: this code used to be an optimized version of handler
// DOES NOT WORK IN NODE 11
// TODO: measure if we actually get something here
// if yes, re-enable for node 10
//if (this.stream instanceof net.Stream) {
// this.stream.ondata = function(data, start, end) {
// connection.packetParser.execute(data, start, end);
// };
//} else {
this.stream.on('data', function(data) {
connection.packetParser.execute(data, 0, data.length);
});
//}
this._protocolError = null;
this.stream.on('end', function() {
// we need to set this flag everywhere where we want connaction to close
if (connection._closing)
return;
if (!connection._protocolError) // no particular error message before disconnect
connection._protocolError = 'PROTOCOL_CONNECTION_LOST'
var err = new Error('Connection lost: The server closed the connection.');
err.fatal = true;
err.code = connection._protocolError;
connection.emit('error', err);
});
var handshakeCommand;
if (!this.config.isServer) {
handshakeCommand = new Commands.ClientHandshake(this.config.clientFlags);
handshakeCommand.on('error', function(e) { connection.emit('error', e); });
this.addCommand(handshakeCommand);
}
}
util.inherits(Connection, EventEmitter);
Connection.prototype.write = function(buffer) {
this.stream.write(buffer);
};
// TODO: replace function in runtime instead of having if() here
// Needs benchmark.
Connection.prototype.writePacket = function(packet) {
packet.writeHeader(this.sequenceId);
if (this.config.debug) {
console.log(this._internalId + ' ' + this.connectionId + ' <== ' + this._command._commandName + '#' + this._command.stateName() + '(' + [this.sequenceId, packet._name, packet.length()].join(',') + ')');
}
this.sequenceId++;
if (this.sequenceId == 256)
this.sequenceId = 0
if (!this.config.compress || !this.authorized) {
this.write(packet.buffer);
} else {
var packetLen = packet.length();
var compressHeader = new Buffer(7);
// TODO: currently all outgoing packets are sent uncompressed (header + deflated length=0 as uncompressed flag)
// Need to implement deflation of outgoing packet. Also need to decide when not to compress small packets
// http://dev.mysql.com/doc/internals/en/compression.html#uncompressed-payload suggest not to compress packets less than 50 bytes
// Write uncompressed packet
compressHeader.fill(0);
compressHeader.writeUInt8(packetLen & 0xff, 0);
compressHeader.writeUInt16LE(packetLen >> 8, 1);
this.write(compressHeader);
this.write(packet.buffer);
}
};
Connection.prototype.startTLS = function(onSecure) {
var connection = this;
var crypto = require('crypto');
var tls = require('tls');
var config = this.config;
var stream = this.stream;
// special case for Amazon RDS: use http://s3.amazonaws.com/rds-downloads/mysql-ssl-ca-cert.pem CA cert if ssl option set to "Amazon RDS"
if (config.ssl === 'Amazon RDS') {
var fs = require('fs');
var path = require('path');
fs.readFile(path.resolve(__dirname, '../fixtures/mysql-ssl-ca-cert.pem'), function(err, ca) {
if (err) throw err;
config.ssl = { ca: ca };
after();
});
} else
after();
function after() {
var credentials = crypto.createCredentials({
key: config.ssl.key,
cert: config.ssl.cert,
passphrase: config.ssl.passphrase,
ca: config.ssl.ca
});
var securePair = tls.createSecurePair(credentials, false);
if (stream.ondata)
stream.ondata = null;
stream.removeAllListeners('data');
stream.pipe(securePair.encrypted);
securePair.encrypted.pipe(stream);
securePair.cleartext.on('data', function(data) {
connection.packetParser.execute(data.parent, data.offset, data.offset + data.length);
});
connection.write = function(buffer) {
securePair.cleartext.write(buffer);
};
securePair.on('secure', onSecure);
}
};
// TODO: this does not work if uncompressed packet is split by compressed
// packet boundary.
// My assumption about compressedPacket to contain one or more complete
// compressed packets was wrong. It can wrap any chunk of data.
// This will be rmoved in favor of connection.startInflate
// currently Handshake command overwrites connection.handlePacket with handleCompressedPacket
// before expecting first compressed packet
var zlib = require('zlib');
Connection.prototype.handleCompressedPacket = function(packet) {
var connection = this;
var inflatedLength = packet.readInt24();
if (inflatedLength !== 0) {
var compressedBody = packet.readBuffer(packet.length() - 3);
zlib.inflate(compressedBody, function(err, packets) {
if (err)
return connection.emit('error', err);
var offset = packets.offset;
var end = offset + packets.length;
var buffer = packets.parent;
var len = 0;
var id = 0;
// single compressed packet can contain multiple uncompressed
while (offset < end) {
len = buffer.readUInt16LE(offset) + (buffer[offset+2] << 16);
id = buffer[offset+3];
connection.handlePacket(new Packet(id, buffer, offset + 4, offset + 4 + len));
offset += 4 + len;
}
});
} else {
inflatedLength = packet.readInt24();
var sequenceId = packet.readInt8();
connection.handlePacket(new Packet(sequenceId, packet.buffer, packet.offset, packet.offset + inflatedLength));
}
};
// TODO: consider using @creationix simple-streams
// https://gist.github.com/creationix/5498108
// https://github.com/creationix/min-stream-uv
// https://github.com/creationix/min-stream-helpers
// TODO: try with Stream2 streams
//
// changes stream -> packetParser to
// stream -> compressedPacketParser -> inflateStream -> packetParser
// note that in the caseof ssl this should become
// stream -> securePair.encrypted -> securePair.cleartext -> compressedPacketParser -> inflateStream -> packetParser
Connection.prototype.startInflate = function() {
var connection = this;
var zlib = require('zlib');
var inflateStream = zlib.createInflate();
var uncompressedPacketParser = connection.packetParser;
connection.packetParser = new PacketParser(function(compressedPacket) {
var inflatedLength = packet.readInt24();
if (inflatedLength !== 0) {
inflateStream.write(packet.readBuffer(packet.length() - 3));
} else {
uncompressedPacketParser.execute(packet.buffer, packet.offset, packet.end);
}
});
inflateStream.on('data', function(buff) {
uncompressedPacketParser.execute(buff.parent, buff.offset, buff.offset + buff.length);
});
if (this.stream.ondata)
this.stream.ondata = null;
this.stream.removeAllListeners('data');
this.pipe();
};
Connection.prototype.pipe = function() {
var connection = this;
if (this.stream instanceof net.Stream) {
this.stream.ondata = function(data, start, end) {
connection.packetParser.execute(data, start, end);
};
} else {
this.stream.on('data', function(data) {
connection.packetParser.execute(data.parent, data.offset, data.offset + data.length);
});
}
};
Connection.prototype.handlePacket = function(packet) {
// TODO: check packet sequenceId here
var packetType = '';
if (packet)
this.sequenceId = packet.sequenceId + 1;
if (this.config.debug) {
if (packet) {
console.log(this._internalId + ' ' + this.connectionId + ' ==> ' + this._command._commandName + '#' + this._command.stateName() + '(' + [packet.sequenceId, packet.type(), packet.length()].join(',') + ')');
}
}
var done = this._command.execute(packet, this);
if (done) {
this.sequenceId = 0;
this._command = this._commands.shift();
if (this._command)
this.handlePacket();
}
};
Connection.prototype.addCommand = function(cmd) {
if (this.config.debug) {
console.log('Add command: ' + arguments.callee.caller.name);
cmd._commandName = arguments.callee.caller.name;
}
if (!this._command) {
this._command = cmd;
this.handlePacket();
} else {
this._commands.push(cmd);
}
return cmd;
};
Connection.prototype.format = function(sql, values) {
if (typeof this.config.queryFormat == "function") {
return this.config.queryFormat.call(this, sql, values, this.config.timezone);
}
return SqlString.format(sql, values, this.config.timezone);
};
Connection.prototype.escape = function(value) {
return SqlString.escape(value, false, this.config.timezone);
};
function _domainify(callback) {
var domain = process.domain;
if (domain && callback)
return process.domain.bind(callback);
else
return callback;
}
Connection.prototype.query = function query(sql, values, cb) {
// copy-paste from node-mysql/lib/Connection.js:createQuery
var options = {};
if (typeof sql === 'object') {
// query(options, cb)
options = sql;
if (typeof values === 'function') {
cb = values;
} else {
options.values = values;
}
} else if (typeof values === 'function') {
// query(sql, cb)
cb = values;
options.sql = sql;
options.values = undefined;
} else {
// query(sql, values, cb)
options.sql = sql;
options.values = values;
}
var rawSql = this.format(options.sql, options.values || []);
return this.addCommand(new Commands.Query(rawSql, options, _domainify(cb)));
};
Connection.prototype.execute = function execute(sql, values, cb) {
var options = {};
if (typeof sql === 'object') {
// execute(options, cb)
options = sql;
if (typeof values === 'function') {
cb = values;
} else {
options.values = values;
}
} else if (typeof values === 'function') {
// execute(sql, cb)
cb = values;
options.sql = sql;
options.values = undefined;
} else {
// execute(sql, values, cb)
options.sql = sql;
options.values = values;
}
return this.addCommand(new Commands.Execute(options, _domainify(cb)));
};
// transaction helpers
Connection.prototype.beginTransaction = function(cb) {
return this.query('START TRANSACTION', cb);
}
Connection.prototype.commit = function(cb) {
return this.query('COMMIT', cb);
}
Connection.prototype.rollback = function(cb) {
return this.query('ROLLBACK', cb);
}
Connection.prototype.ping = function ping(cb) {
return this.addCommand(new Commands.Ping(_domainify(cb)));
};
Connection.prototype._registerSlave = function registerSlave(opts, cb) {
return this.addCommand(new Commands.RegisterSlave(opts, _domainify(cb)));
};
Connection.prototype._binlogDump = function binlogDump(opts, cb) {
return this.addCommand(new Commands.BinlogDump(opts, _domainify(cb)));
};
Connection.prototype.createBinlogStream = function(opts) {
// TODO: create proper stream class
// TODO: use through2
var test = 1;
var Readable = require('stream').Readable;
var stream = new Readable({objectMode: true});
stream._read = function() {
return {
data: test++
}
};
var connection = this;
connection._registerSlave(opts, function(err) {
var dumpCmd = connection._binlogDump(opts);
dumpCmd.on('event', function(ev) {
stream.push(ev);
});
dumpCmd.on('eof', function() {
stream.push(null);
// if non-blocking, then close stream to prevent errors
if (opts.flags && (opts.flags & 0x01)) {
connection._closing = true;
connection.stream.end();
}
});
// TODO: pipe errors as well
})
return stream;
}
Connection.prototype.connect = function(cb) {
if (!cb) return;
var connectCalled = 0;
// TODO domainify this callback as well. Note that domain has to be captured
// at the top of function due to nested callback
function callbackOnce(isErrorHandler) {
return function(param) {
if (!connectCalled) {
if (isErrorHandler)
cb(param);
else
cb(null, param);
}
connectCalled = 1;
};
}
this.once('error', callbackOnce(true) );
this.once('connect', callbackOnce(false));
};
// ===================================
// outgoing server connection methods
// ===================================
Connection.prototype.writeColumns = function(columns) {
var connection = this;
this.writePacket(Packets.ResultSetHeader.toPacket(columns.length));
columns.forEach(function(column) {
connection.writePacket(Packets.ColumnDefinition.toPacket(column));
});
this.writeEof();
};
// row is array of columns, not hash
Connection.prototype.writeTextRow = function(column) {
this.writePacket(Packets.TextRow.toPacket(column));
};
Connection.prototype.writeTextResult = function(rows, columns) {
var connection = this;
connection.writeColumns(columns);
rows.forEach(function(row) {
var arrayRow = new Array(columns.length);
columns.forEach(function(column) {
arrayRow.push(row[column.name]);
});
connection.writeTextRow(arrayRow);
});
connection.writeEof();
};
Connection.prototype.writeEof = function(warnings, statusFlags) {
this.writePacket(Packets.EOF.toPacket(warnings, statusFlags));
};
Connection.prototype.writeOk = function(args) {
if (!args)
args = { affectedRows: 0 };
this.writePacket(Packets.OK.toPacket(args));
};
Connection.prototype.writeError = function(args) {
this.writePacket(Packets.Error.toPacket(args));
};
Connection.prototype.serverHandshake = function serverHandshake(args) {
return this.addCommand(new Commands.ServerHandshake(args));
};
// ===============================================================
// TODO: domainify
Connection.prototype.end = function(callback) {
// TODO: implement COM_QUIT command
var self = this;
var endCmd = { connection: this };
endCmd.execute = function() {
self._closing = true;
this.connection.stream.end();
if (callback)
callback();
};
return this.addCommand(endCmd);
//return this.addCommand(new Commands.Quit(callback));
};
module.exports = Connection;