Merge pull request #174 from roblabla/rewrite-internals

Rewrite internal pipeline to use transform stremas

package.json

@@ -1,6 +1,6 @@
 {
   "name": "minecraft-protocol",
-  "version": "0.13.5-GH",
+  "version": "0.14.0-GH",
   "description": "Parse and serialize minecraft packets, plus authentication and encryption.",
   "main": "index.js",
   "repository": {
@@ -44,6 +44,7 @@
     "buffer-equal": "0.0.0",
     "node-uuid": "~1.4.1",
     "prismarine-nbt": "0.0.1",
+    "readable-stream": "^1.1.0",
     "superagent": "~0.10.0",
     "ursa-purejs": "0.0.3"
   },

src/client.js

@@ -11,6 +11,10 @@ var EventEmitter = require('events').EventEmitter
   , packetNames = protocol.packetNames
   , states = protocol.states
   , debug = require('./debug')
+  , serializer = require('./transforms/serializer')
+  , compression = require('./transforms/compression')
+  , framing = require('./transforms/framing')
+  , crypto = require('crypto')
   ;
 
 module.exports = Client;
@@ -18,24 +22,38 @@ module.exports = Client;
 function Client(isServer) {
   EventEmitter.call(this);
 
+  var socket;
+  this.packetsToParse = {};
+
+  this.serializer = serializer.createSerializer({ isServer });
+  this.compressor = null;
+  this.framer = framing.createFramer();
+  this.cipher = null;
+
+  this.decipher = null;
+  this.splitter = framing.createSplitter();
+  this.decompressor = null;
+  this.deserializer = serializer.createDeserializer({ isServer, packetsToParse: this.packetsToParse });
+
   this._state = states.HANDSHAKING;
   Object.defineProperty(this, "state", {
     get: function() {
-      return this._state;
+      return this.serializer.protocolState;
     },
     set: function(newProperty) {
-      var oldProperty = this._state;
+      var oldProperty = this.serializer.protocolState;
-      this._state = newProperty;
+      this.serializer.protocolState = newProperty;
+      this.deserializer.protocolState = newProperty;
       this.emit('state', newProperty, oldProperty);
     }
   });
+  Object.defineProperty(this, "compressionThreshold", {
+    get: () => this.compressor == null ? -2 : this.compressor.compressionThreshold,
+    set: (threshold) => this.setCompressionThreshold(threshold)
+  });
+
   this.isServer = !!isServer;
-  this.socket = null;
+
-  this.encryptionEnabled = false;
-  this.cipher = null;
-  this.decipher = null;
-  this.compressionThreshold = -2;
-  this.packetsToParse = {};
   this.on('newListener', function(event, listener) {
     var direction = this.isServer ? 'toServer' : 'toClient';
     if(protocol.packetStates[direction].hasOwnProperty(event) || event === "packet") {
@@ -78,77 +96,46 @@ Client.prototype.onRaw = function(type, func) {
 };
 
 Client.prototype.setSocket = function(socket) {
-  var self = this;
-
-  function afterParse(err, parsed) {
-    if(err || (parsed && parsed.error)) {
-      self.emit('error', err || parsed.error);
-      self.end("ProtocolError");
-      return;
-    }
-    if(!parsed) {
-      return;
-    }
-    var packet = parsed.results;
-    //incomingBuffer = incomingBuffer.slice(parsed.size); TODO: Already removed in prepare
-
-    var packetName = protocol.packetNames[self.state][self.isServer ? 'toServer' : 'toClient'][packet.id];
-    var packetState = self.state;
-    self.emit(packetName, packet);
-    self.emit('packet', packet);
-    self.emit('raw.' + packetName, parsed.buffer, packetState);
-    self.emit('raw', parsed.buffer, packetState);
-    prepareParse();
-  }
-
-  function prepareParse() {
-    var packetLengthField = protocol.types["varint"][0](incomingBuffer, 0);
-    if(packetLengthField && packetLengthField.size + packetLengthField.value <= incomingBuffer.length) {
-      var buf = incomingBuffer.slice(packetLengthField.size, packetLengthField.size + packetLengthField.value);
-      // TODO : Slice as early as possible to avoid processing same data twice.
-      incomingBuffer = incomingBuffer.slice(packetLengthField.size + packetLengthField.value);
-      if(self.compressionThreshold == -2) {
-        afterParse(null, parsePacketData(buf, self.state, self.isServer, self.packetsToParse));
-      } else {
-        parseNewStylePacket(buf, self.state, self.isServer, self.packetsToParse, afterParse);
-      }
-    }
-  }
-
-  self.socket = socket;
-  if(self.socket.setNoDelay)
-    self.socket.setNoDelay(true);
-  var incomingBuffer = new Buffer(0);
-  self.socket.on('data', function(data) {
-    if(self.encryptionEnabled) data = new Buffer(self.decipher.update(data), 'binary');
-    incomingBuffer = Buffer.concat([incomingBuffer, data]);
-    prepareParse()
-  });
-
-  self.socket.on('connect', function() {
-    self.emit('connect');
-  });
-
-  self.socket.on('error', onError);
-  self.socket.on('close', endSocket);
-  self.socket.on('end', endSocket);
-  self.socket.on('timeout', endSocket);
-
-  function onError(err) {
-    self.emit('error', err);
-    endSocket();
-  }
-
   var ended = false;
 
-  function endSocket() {
+  // TODO : A lot of other things needs to be done.
+  var endSocket = () => {
     if(ended) return;
     ended = true;
-    self.socket.removeListener('close', endSocket);
+    this.socket.removeListener('close', endSocket);
-    self.socket.removeListener('end', endSocket);
+    this.socket.removeListener('end', endSocket);
-    self.socket.removeListener('timeout', endSocket);
+    this.socket.removeListener('timeout', endSocket);
-    self.emit('end', self._endReason);
+    this.emit('end', this._endReason);
-  }
+  };
+
+  var onError = (err) => {
+    this.emit('error', err);
+    endSocket();
+  };
+
+  this.socket = socket;
+
+  if(this.socket.setNoDelay)
+    this.socket.setNoDelay(true);
+
+  this.socket.on('connect', () => this.emit('connect'));
+
+  this.socket.on('error', onError);
+  this.socket.on('close', endSocket);
+  this.socket.on('end', endSocket);
+  this.socket.on('timeout', endSocket);
+
+  this.socket.pipe(this.splitter).pipe(this.deserializer);
+  this.serializer.pipe(this.framer).pipe(this.socket);
+
+  this.deserializer.on('data', (parsed) => {
+    var packet = parsed.results;
+    var packetName = protocol.packetNames[packet.state][this.isServer ? 'toServer' : 'toClient'][packet.id];
+    this.emit('packet', packet);
+    this.emit(packetName, packet);
+    this.emit('raw.' + packetName, parsed.buffer, packet.state);
+    this.emit('raw', parsed.buffer, packet.state);
+  });
 };
 
 Client.prototype.end = function(reason) {
@@ -156,64 +143,52 @@ Client.prototype.end = function(reason) {
   this.socket.end();
 };
 
+Client.prototype.setEncryption = function(sharedSecret) {
+  if (this.cipher != null)
+    throw new Error("Set encryption twice !");
+  this.cipher = crypto.createCipheriv('aes-128-cfb8', sharedSecret, sharedSecret);
+  this.framer.unpipe(this.socket);
+  this.framer.pipe(this.cipher).pipe(this.socket);
+  this.decipher = crypto.createDecipheriv('aes-128-cfb8', sharedSecret, sharedSecret);
+  this.socket.unpipe(this.splitter);
+  this.socket.pipe(this.decipher).pipe(this.splitter);
+}
+
+Client.prototype.setCompressionThreshold = function(threshold) {
+  if (this.compressor == null) {
+    this.compressor = compression.createCompressor(threshold);
+    this.serializer.unpipe(this.framer);
+    this.serializer.pipe(this.compressor).pipe(this.framer);
+    this.decompressor = compression.createDecompressor(threshold);
+    this.splitter.unpipe(this.deserializer);
+    this.splitter.pipe(this.decompressor).pipe(this.deserializer);
+  } else {
+    this.decompressor.threshold = threshold;
+    this.compressor.threshold = threshold;
+  }
+}
+
 function noop(err) {
   if(err) throw err;
 }
 
-Client.prototype.write = function(packetId, params, cb) {
+Client.prototype.write = function(packetId, params, cb = noop) {
-  cb = cb || noop;
   if(Array.isArray(packetId)) {
     if(packetId[0] !== this.state)
       return false;
     packetId = packetId[1];
   }
   if(typeof packetId === "string")
-    packetId = packetIds[this.state][this.isServer ? "toClient" : "toServer"][packetId];
+    packetId = protocol.packetIds[this.state][this.isServer ? "toClient" : "toServer"][packetId];
-  var that = this;
+  var packetName = protocol.packetNames[this.state][this.isServer ? "toClient" : "toServer"][packetId];
-
+  debug("writing packetId " + this.state + "." + packetName + " (0x" + packetId.toString(16) + ")");
-  var finishWriting = function(err, buffer) {
+  debug(params);
-    if(err) {
+  this.serializer.write({ packetId, params }, cb);
-      console.log(err);
-      throw err; // TODO : Handle errors gracefully, if possible
-    }
-    var packetName = packetNames[that.state][that.isServer ? "toClient" : "toServer"][packetId];
-    debug("writing packetId " + that.state + "." + packetName + " (0x" + packetId.toString(16) + ")");
-    debug(params);
-    var out = that.encryptionEnabled ? new Buffer(that.cipher.update(buffer), 'binary') : buffer;
-    that.socket.write(out,cb);
-    return true;
-  };
-
-  var buffer = createPacketBuffer(packetId, this.state, params, this.isServer);
-  if(this.compressionThreshold >= 0 && buffer.length >= this.compressionThreshold) {
-    debug("Compressing packet");
-    compressPacketBuffer(buffer, finishWriting);
-  } else if(this.compressionThreshold >= -1) {
-    debug("New-styling packet");
-    newStylePacket(buffer, 0, finishWriting);
-  } else {
-    debug("Old-styling packet");
-    oldStylePacket(buffer, finishWriting);
-  }
 };
 
-// TODO : Perhaps this should only accept buffers without length, so we can
-// handle compression ourself ? Needs to ask peopl who actually use this feature
-// like @deathcap
 Client.prototype.writeRaw = function(buffer) {
-  var self = this;
+  if (this.compressor === null)
-
+    this.framer.write(buffer);
-  var finishWriting = function(error, buffer) {
+  else
-    if(error)
+    this.compressor.write(buffer);
-      throw error; // TODO : How do we handle this error ?
-    var out = self.encryptionEnabled ? new Buffer(self.cipher.update(buffer), 'binary') : buffer;
-    self.socket.write(out);
-  };
-  if(this.compressionThreshold >= 0 && buffer.length >= this.compressionThreshold) {
-    compressPacketBuffer(buffer, finishWriting);
-  } else if(this.compressionThreshold >= -1) {
-    newStylePacket(buffer, 0, finishWriting);
-  } else {
-    oldStylePacket(buffer, finishWriting);
-  }
 };

src/index.js

@@ -183,11 +183,9 @@ function createServer(options) {
         client.end('DidNotEncryptVerifyTokenProperly');
         return;
       }
-      client.cipher = crypto.createCipheriv('aes-128-cfb8', sharedSecret, sharedSecret);
+      client.setEncryption(sharedSecret);
-      client.decipher = crypto.createDecipheriv('aes-128-cfb8', sharedSecret, sharedSecret);
       hash.update(sharedSecret);
       hash.update(client.publicKey);
-      client.encryptionEnabled = true;
 
       var isException = !!server.onlineModeExceptions[client.username.toLowerCase()];
       var needToVerify = (onlineMode && !isException) || (!onlineMode && isException);
@@ -365,16 +363,14 @@ function createClient(options) {
         var pubKey = mcPubKeyToURsa(packet.publicKey);
         var encryptedSharedSecretBuffer = pubKey.encrypt(sharedSecret, undefined, undefined, ursa.RSA_PKCS1_PADDING);
         var encryptedVerifyTokenBuffer = pubKey.encrypt(packet.verifyToken, undefined, undefined, ursa.RSA_PKCS1_PADDING);
-        client.cipher = crypto.createCipheriv('aes-128-cfb8', sharedSecret, sharedSecret);
-        client.decipher = crypto.createDecipheriv('aes-128-cfb8', sharedSecret, sharedSecret);
         client.write(0x01, {
           sharedSecret: encryptedSharedSecretBuffer,
           verifyToken: encryptedVerifyTokenBuffer,
         },function(err){
           if(err)
             throw err;
-          client.encryptionEnabled = true;
         });
+        client.setEncryption(sharedSecret);
       }
     }
   }

src/protocol.js

@@ -133,35 +133,8 @@ function createPacketBuffer(packetId, state, params, isServer) {
   return buffer;
 }
 
-function compressPacketBuffer(buffer, callback) {
+// By default, parse every packets.
-  var dataLength = buffer.size;
+function parsePacketData(buffer, state, isServer, packetsToParse = {"packet": true}) {
-  zlib.deflate(buffer, function(err, buf) {
-    if(err)
-      callback(err);
-    else
-      newStylePacket(buf, buffer.length, callback);
-  });
-}
-
-function oldStylePacket(buffer, callback) {
-  var packet = new Buffer(utils.varint[2](buffer.length) + buffer.length);
-  var cursor = utils.varint[1](buffer.length, packet, 0);
-  utils.buffer[1](buffer, packet, cursor);
-  callback(null, packet);
-}
-
-function newStylePacket(buffer, dataSize, callback) {
-  var sizeOfDataLength = utils.varint[2](dataSize);
-  var sizeOfLength = utils.varint[2](buffer.length + sizeOfDataLength);
-  var size = sizeOfLength + sizeOfDataLength + buffer.length;
-  var packet = new Buffer(size);
-  var cursor = utils.varint[1](size - sizeOfLength, packet, 0);
-  cursor = utils.varint[1](dataSize, packet, cursor);
-  utils.buffer[1](buffer, packet, cursor);
-  callback(null, packet);
-}
-
-function parsePacketData(buffer, state, isServer, packetsToParse) {
   var cursor = 0;
   var packetIdField = utils.varint[0](buffer, cursor);
   var packetId = packetIdField.value;
@@ -207,6 +180,10 @@ function parsePacketData(buffer, state, isServer, packetsToParse) {
      results: results
      };
      }*/
+    // TODO : investigate readResults returning null : shouldn't happen.
+    // When there is not enough data to read, we should return an error.
+    // As a general rule, it would be a good idea to introduce a whole bunch
+    // of new error classes to differenciate the errors.
     if(readResults === null || readResults.value == null) continue;
     if(readResults.error) {
       return readResults;
@@ -223,33 +200,12 @@ function parsePacketData(buffer, state, isServer, packetsToParse) {
   };
 }
 
-function parseNewStylePacket(buffer, state, isServer, packetsToParse, cb) {
-  var dataLengthField = utils.varint[0](buffer, 0);
-  var buf = buffer.slice(dataLengthField.size);
-  if(dataLengthField.value != 0) {
-    zlib.inflate(buf, function(err, newbuf) {
-      if(err) {
-        console.log(err);
-        cb(err);
-      } else {
-        cb(null, parsePacketData(newbuf, state, isServer, packetsToParse));
-      }
-    });
-  } else {
-    cb(null, parsePacketData(buf, state, isServer, packetsToParse));
-  }
-}
-
 module.exports = {
   version: 47,
   minecraftVersion: '1.8.1',
   sessionVersion: 13,
   parsePacketData: parsePacketData,
-  parseNewStylePacket: parseNewStylePacket,
   createPacketBuffer: createPacketBuffer,
-  compressPacketBuffer: compressPacketBuffer,
-  oldStylePacket: oldStylePacket,
-  newStylePacket: newStylePacket,
   packetIds: packetIds,
   packetNames: packetNames,
   packetFields: packetFields,

src/transforms/compression.js

@@ -0,0 +1,69 @@
+var [readVarInt, writeVarInt, sizeOfVarInt] = require("../datatypes/utils").varint;
+var zlib = require("zlib");
+var Transform = require("readable-stream").Transform;
+
+module.exports.createCompressor = function(threshold) {
+  return new Compressor(threshold);
+}
+
+module.exports.createDecompressor = function(threshold) {
+  return new Decompressor(threshold);
+}
+
+class Compressor extends Transform {
+  constructor(compressionThreshold = -1) {
+    super();
+    this.compressionThreshold = compressionThreshold;
+  }
+
+  _transform(chunk, enc, cb) {
+    if (chunk.length >= this.compressionThreshold)
+    {
+      zlib.deflate(chunk, (err, newChunk) => {
+        if (err)
+          return cb(err);
+        var buf = new Buffer(sizeOfVarInt(chunk.length) + newChunk.length);
+        var offset = writeVarInt(chunk.length, buf, 0);
+        newChunk.copy(buf, offset);
+        this.push(buf);
+        return cb();
+      });
+    }
+    else
+    {
+      var buf = new Buffer(sizeOfVarInt(0) + chunk.length);
+      var offset = writeVarInt(0, buf, 0);
+      chunk.copy(buf, offset);
+      this.push(buf);
+      return cb();
+    }
+  }
+}
+
+class Decompressor extends Transform {
+  constructor(compressionThreshold = -1) {
+    super();
+    this.compressionThreshold = compressionThreshold;
+  }
+
+  _transform(chunk, enc, cb) {
+    var size, value, error;
+    ({ size, value, error } = readVarInt(chunk, 0));
+    if (error)
+      return cb(error);
+    if (value === 0)
+    {
+      this.push(chunk.slice(size));
+      return cb();
+    }
+    else
+    {
+      zlib.inflate(chunk.slice(size), (err, newBuf) => {
+        if (err)
+          return cb(err);
+        this.push(newBuf);
+        return cb();
+      });
+    }
+  }
+}

src/transforms/framing.js

@@ -0,0 +1,47 @@
+var [readVarInt, writeVarInt, sizeOfVarInt] = require("../datatypes/utils").varint;
+var Transform = require("readable-stream").Transform;
+
+module.exports.createSplitter = function() {
+  return new Splitter();
+}
+
+module.exports.createFramer = function() {
+  return new Framer();
+}
+
+class Framer extends Transform {
+  constructor() {
+    super();
+  }
+
+  _transform(chunk, enc, cb) {
+    var buffer = new Buffer(sizeOfVarInt(chunk.length));
+    writeVarInt(chunk.length, buffer, 0);
+    this.push(buffer);
+    this.push(chunk);
+    return cb();
+  }
+}
+
+class Splitter extends Transform {
+  constructor() {
+    super();
+    this.buffer = new Buffer(0);
+  }
+  _transform(chunk, enc, cb) {
+    this.buffer = Buffer.concat([this.buffer, chunk]);
+    var value, size, error;
+    var offset = 0;
+
+    ({ value, size, error } = readVarInt(this.buffer, offset) || { error: "Not enough data" });
+    while (!error && this.buffer.length >= offset + size + value)
+    {
+      this.push(this.buffer.slice(offset + size, offset + size + value));
+      offset += size + value;
+      ({ value, size, error } = readVarInt(this.buffer, offset) || { error: "Not enough data" });
+    }
+    this.buffer = this.buffer.slice(offset);
+    return cb();
+  }
+}
+

src/transforms/serializer.js

@@ -0,0 +1,58 @@
+var [readVarInt, writeVarInt, sizeOfVarInt] = require("../datatypes/utils").varint;
+var protocol = require("../protocol");
+var Transform = require("readable-stream").Transform;
+
+module.exports.createSerializer = function(obj) {
+  return new Serializer(obj);
+}
+
+module.exports.createDeserializer = function(obj) {
+  return new Deserializer(obj);
+}
+
+class Serializer extends Transform {
+  constructor({ state = protocol.states.HANDSHAKING, isServer = false } = {}) {
+    super({ writableObjectMode: true });
+    this.protocolState = state;
+    this.isServer = isServer;
+  }
+
+  // TODO : Might make sense to make createPacketBuffer async.
+  _transform(chunk, enc, cb) {
+    try {
+      var buf = protocol.createPacketBuffer(chunk.packetId, this.protocolState, chunk.params, this.isServer);
+      this.push(buf);
+      return cb();
+    } catch (e) {
+      return cb(e);
+    }
+  }
+}
+
+class Deserializer extends Transform {
+  constructor({ state = protocol.states.HANDSHAKING, isServer = false, packetsToParse = {"packet": true} } = {}) {
+    super({ readableObjectMode: true });
+    this.protocolState = state;
+    this.isServer = isServer;
+    this.packetsToParse = packetsToParse;
+  }
+
+  _transform(chunk, enc, cb) {
+    var packet;
+    try {
+      packet = protocol.parsePacketData(chunk, this.protocolState, this.isServer, this.packetsToParse);
+    } catch (e) {
+      return cb(e);
+    }
+    if (packet.error)
+    {
+      packet.error.packet = packet;
+      return cb(packet.error)
+    }
+    else
+    {
+      this.push(packet);
+      return cb();
+    }
+  }
+}

test/benchmark.js

@@ -1,21 +1,8 @@
 var ITERATIONS = 100000;
 
-var Client = require('../dist/client'),
+var protocol = require('../dist/protocol'),
-  EventEmitter = require('events').EventEmitter,
   util = require('util'),
-  states = require('../dist/protocol').states;
+  states = protocol.states;
-
-var FakeSocket = function() {
-  EventEmitter.call(this);
-};
-util.inherits(FakeSocket, EventEmitter);
-FakeSocket.prototype.write = function() {
-};
-
-var client = new Client();
-var socket = new FakeSocket();
-client.setSocket(socket);
-client.state = states.PLAY;
 
 var testDataWrite = [
   {id: 0x00, params: {keepAliveId: 957759560}},
@@ -24,37 +11,21 @@ var testDataWrite = [
   // TODO: add more packets for better quality data
 ];
 
+var inputData = [];
+
 var start, i, j;
 console.log('Beginning write test');
 start = Date.now();
 for(i = 0; i < ITERATIONS; i++) {
   for(j = 0; j < testDataWrite.length; j++) {
-    client.write(testDataWrite[j].id, testDataWrite[j].params);
+    inputData.push(protocol.createPacketBuffer(testDataWrite[j].id, states.PLAY, testDataWrite[j].params, false));
   }
 }
 console.log('Finished write test in ' + (Date.now() - start) / 1000 + ' seconds');
 
-var testDataRead = [
-  {id: 0x00, params: {keepAliveId: 957759560}},
-  {id: 0x02, params: {message: '<Bob> Hello World!', position: 0}},
-  {id: 0x08, params: {x: 6.5, y: 65.62, z: 7.5, yaw: 0, pitch: 0, flags: 0}},
-];
-
-client.isServer = true;
-
-var inputData = new Buffer(0);
-socket.write = function(data) {
-  inputData = Buffer.concat([inputData, data]);
-};
-for(i = 0; i < testDataRead.length; i++) {
-  client.write(testDataRead[i].id, testDataRead[i].params);
-}
-
-client.isServer = false;
-
 console.log('Beginning read test');
 start = Date.now();
-for(i = 0; i < ITERATIONS; i++) {
+for (j = 0; j < inputData.length; j++) {
-  socket.emit('data', inputData);
+  protocol.parsePacketData(inputData[j], states.PLAY, true);
 }
 console.log('Finished read test in ' + (Date.now() - start) / 1000 + ' seconds');