WIP compressed write

Author: dougwilson

lib/protocol/PacketWriter.js

@@ -7,15 +7,22 @@ var COMPRESSED_PACKET_HEADER_LENGTH = 7;
 var IEEE_754_BINARY_64_PRECISION = Math.pow(2, 53);
 var PACKET_HEADER_LENGTH         = 4;
 var MAX_PACKET_LENGTH            = Math.pow(2, 24) - 1;
-var Buffer                       = require('safe-buffer').Buffer;
+
+var Buffer       = require('safe-buffer').Buffer;
+var BufferList   = require('./BufferList');
+var EventEmitter = require('events').EventEmitter;
+var Util         = require('util');
+var Zlib         = require('zlib');
 
 module.exports = PacketWriter;
 function PacketWriter() {
   this._buffer = null;
   this._offset = 0;
+  this._sync   = false;
 }
+Util.inherits(PacketWriter, EventEmitter);
 
-PacketWriter.prototype.toBuffer = function toBuffer(parser) {
+PacketWriter.prototype.finalize = function finalize(parser) {
   if (!this._buffer) {
     this._buffer = Buffer.alloc(0);
     this._offset = 0;
@@ -36,6 +43,7 @@ PacketWriter.prototype.toBuffer = function toBuffer(parser) {
       : MAX_PACKET_LENGTH;
 
     var packetNumber = parser.incrementPacketNumber();
+    var packetOffset = this._offset;
 
     this.writeUnsignedNumber(3, packetLength);
     this.writeUnsignedNumber(1, packetNumber);
@@ -44,6 +52,10 @@ PacketWriter.prototype.toBuffer = function toBuffer(parser) {
     var end   = start + packetLength;
 
     this.writeBuffer(buffer.slice(start, end));
+
+    if (!parser._compressed) {
+      this.emit('data', this._buffer.slice(packetOffset, this._offset));
+    }
   }
 
   if (parser._compressed) {
@@ -52,9 +64,6 @@ PacketWriter.prototype.toBuffer = function toBuffer(parser) {
     packets = Math.floor(length / MAX_PACKET_LENGTH) + 1;
     size    = length + (packets * COMPRESSED_PACKET_HEADER_LENGTH);
 
-    this._buffer = Buffer.allocUnsafe(size);
-    this._offset = 0;
-
     for (packet = 0; packet < packets; packet++) {
       isLast = (packet + 1 === packets);
       packetLength = (isLast)
@@ -63,6 +72,9 @@ PacketWriter.prototype.toBuffer = function toBuffer(parser) {
 
       packetNumber = parser.incrementCompressedPacketNumber();
 
+      this._buffer = Buffer.allocUnsafe(packetLength + COMPRESSED_PACKET_HEADER_LENGTH);
+      this._offset = 0;
+
       this.writeUnsignedNumber(3, packetLength);
       this.writeUnsignedNumber(1, packetNumber);
       this.writeUnsignedNumber(3, 0);
@@ -71,8 +83,28 @@ PacketWriter.prototype.toBuffer = function toBuffer(parser) {
       var end   = start + packetLength;
 
       this.writeBuffer(buffer.slice(start, end));
+
+      this.emit('data', this._buffer);
     }
   }
+};
+
+PacketWriter.prototype.toBuffer = function toBuffer(parser) {
+  var bufs = new BufferList();
+
+  this.on('data', function (data) {
+    bufs.push(data);
+  });
+
+  this._sync = true;
+  this.finalize(parser);
+
+  this._buffer = Buffer.allocUnsafe(bufs.size);
+  this._offset = 0;
+
+  while (bufs.size > 0) {
+    this._offset += bufs.shift().copy(this._buffer, this._offset);
+  }
 
   return this._buffer;
 };

lib/protocol/Protocol.js

@@ -306,8 +306,14 @@ Protocol.prototype._parsePacketDebug = function _parsePacketDebug(packet) {
 
 Protocol.prototype._emitPacket = function(packet) {
   var packetWriter = new PacketWriter();
+  var protocol     = this;
+
+  packetWriter.on('data', function (data) {
+    protocol.emit('data', data);
+  });
+
   packet.write(packetWriter);
-  this.emit('data', packetWriter.toBuffer(this._parser));
+  packetWriter.toBuffer(this._parser);
 
   if (this._config.debug) {
     this._debugPacket(false, packet);