var Parser = require('./Parser'); var Sequences = require('./sequences'); var Packets = require('./packets'); var Auth = require('./Auth'); var Stream = require('stream').Stream; var Util = require('util'); var PacketWriter = require('./PacketWriter'); module.exports = Protocol; Util.inherits(Protocol, Stream); function Protocol(options) { Stream.call(this); options = options || {}; this.readable = true; this.writable = true; this._config = options.config || {}; this._connection = options.connection; this._callback = null; this._fatalError = null; this._quitSequence = null; this._handshakeSequence = null; this._destroyed = false; this._queue = []; this._handshakeInitializationPacket = null; this._parser = new Parser({ onPacket : this._parsePacket.bind(this), config : this._config }); } Protocol.prototype.write = function(buffer) { this._parser.write(buffer); return true; }; Protocol.prototype.handshake = function(cb) { return this._handshakeSequence = this._enqueue(new Sequences.Handshake(this._config, cb)); }; Protocol.prototype.query = function(options, cb) { return this._enqueue(new Sequences.Query(options, cb)); }; Protocol.prototype.changeUser = function(options, cb) { return this._enqueue(new Sequences.ChangeUser(options, cb)); }; Protocol.prototype.ping = function(cb) { return this._enqueue(new Sequences.Ping(cb)); }; Protocol.prototype.stats = function(cb) { return this._enqueue(new Sequences.Statistics(cb)); }; Protocol.prototype.quit = function(cb) { return this._quitSequence = this._enqueue(new Sequences.Quit(cb)); }; Protocol.prototype.end = function() { var expected = (this._quitSequence && this._queue[0] === this._quitSequence); if (expected) { this._quitSequence.end(); this.emit('end'); return; } var err = new Error('Connection lost: The server closed the connection.'); err.fatal = true; err.code = 'PROTOCOL_CONNECTION_LOST'; this._delegateError(err); }; Protocol.prototype.pause = function() { this._parser.pause(); }; Protocol.prototype.resume = function() { this._parser.resume(); }; Protocol.prototype._enqueue = function(sequence) { if (!this._validateEnqueue(sequence)) { return sequence; } this._queue.push(sequence); var self = this; sequence .on('error', function(err) { self._delegateError(err, sequence); }) .on('packet', function(packet) { self._emitPacket(packet); }) .on('end', function() { self._dequeue(); }); if (this._queue.length === 1) { this._parser.resetPacketNumber(); sequence.start(); } return sequence; }; Protocol.prototype._validateEnqueue = function(sequence) { var err; var prefix = 'Cannot enqueue ' + sequence.constructor.name + ' after '; if (this._quitSequence) { err = new Error(prefix + 'invoking quit.'); err.code = 'PROTOCOL_ENQUEUE_AFTER_QUIT'; } else if (this._destroyed) { err = new Error(prefix + 'being destroyed.'); err.code = 'PROTOCOL_ENQUEUE_AFTER_DESTROY'; } else if (this._handshakeSequence && sequence.constructor === Sequences.Handshake) { err = new Error(prefix + 'already enqueuing a Handshake.'); err.code = 'PROTOCOL_ENQUEUE_HANDSHAKE_TWICE'; } else { return true; } var self = this; err.fatal = false; sequence .on('error', function(err) { self._delegateError(err, sequence); }) .end(err); return false; }; Protocol.prototype._parsePacket = function() { var sequence = this._queue[0]; var Packet = this._determinePacket(sequence); var packet = new Packet(); // Special case: Faster dispatch, and parsing done inside sequence if (Packet === Packets.RowDataPacket) { sequence.RowDataPacket(packet, this._parser, this._connection); if (this._config.debug) { this._debugPacket(true, packet); } return; } packet.parse(this._parser); if (this._config.debug) { this._debugPacket(true, packet); } if (Packet === Packets.HandshakeInitializationPacket) { this._handshakeInitializationPacket = packet; } sequence[Packet.name](packet); }; Protocol.prototype._emitPacket = function(packet) { var packetWriter = new PacketWriter(); packet.write(packetWriter); this.emit('data', packetWriter.toBuffer(this._parser)); if (this._config.debug) { this._debugPacket(false, packet) } }; Protocol.prototype._determinePacket = function(sequence) { var firstByte = this._parser.peak(); if (sequence.determinePacket) { var Packet = sequence.determinePacket(firstByte, this._parser); if (Packet) { return Packet; } } switch (firstByte) { case 0x00: return Packets.OkPacket; case 0xfe: return Packets.EofPacket; case 0xff: return Packets.ErrorPacket; } throw new Error('Could not determine packet, firstByte = ' + firstByte); }; Protocol.prototype._dequeue = function() { // No point in advancing the queue, we are dead if (this._fatalError) { return; } this._queue.shift(); var sequence = this._queue[0]; if (!sequence) { this.emit('drain'); return; } this._parser.resetPacketNumber(); if (sequence.constructor == Sequences.ChangeUser) { sequence.start(this._handshakeInitializationPacket); return; } sequence.start(); }; Protocol.prototype.handleNetworkError = function(err) { err.fatal = true; var sequence = this._queue[0]; if (sequence) { sequence.end(err) } else { this._delegateError(err); } }; Protocol.prototype._delegateError = function(err, sequence) { // Stop delegating errors after the first fatal error if (this._fatalError) { return; } if (err.fatal) { this._fatalError = err; } if (this._shouldErrorBubbleUp(err, sequence)) { // Can't use regular 'error' event here as that always destroys the pipe // between socket and protocol which is not what we want (unless the // exception was fatal). this.emit('unhandledError', err); } else if (err.fatal) { this._queue.forEach(function(sequence) { sequence.end(err); }); } // Make sure the stream we are piping to is getting closed if (err.fatal) { this.emit('end', err); } }; Protocol.prototype._shouldErrorBubbleUp = function(err, sequence) { if (sequence) { if (sequence.hasErrorHandler()) { return false; } else if (!err.fatal) { return true; } } return (err.fatal && !this._hasPendingErrorHandlers()); }; Protocol.prototype._hasPendingErrorHandlers = function() { return this._queue.some(function(sequence) { return sequence.hasErrorHandler(); }); }; Protocol.prototype.destroy = function() { this._destroyed = true; this._parser.pause(); }; Protocol.prototype._debugPacket = function(incoming, packet) { var headline = (incoming) ? '<-- ' : '--> '; headline = headline + packet.constructor.name; console.log(headline); console.log(packet); console.log(''); };