fixup! WIP compressed write

Author: dougwilson

lib/protocol/PacketWriter.js

@@ -12,12 +12,15 @@ 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;
+  this._buffer       = null;
+  this._deflateQueue = [];
+  this._deflating    = false;
+  this._offset       = 0;
+  this._sync         = false;
 }
 Util.inherits(PacketWriter, EventEmitter);
 
@@ -48,10 +51,18 @@ PacketWriter.prototype.finalize = function finalize(parser) {
 
     if (parser._compressed) {
       num = parser.incrementCompressedPacketNumber();
-      buf = this._toCompressedPacket(num, buf);
+
+      if (this._sync) {
+        buf = this._toCompressedPacket(num, buf);
+      } else {
+        this._toCompressedPacketAsync(num, buf);
+        buf = null;
+      }
     }
 
-    this.emit('data', buf);
+    if (buf) {
+      this.emit('data', buf);
+    }
   }
 };
 
@@ -239,7 +250,31 @@ PacketWriter.prototype._allocate = function _allocate(bytes) {
   oldBuffer.copy(this._buffer);
 };
 
-PacketWriter.prototype._toCompressedPacket = function _toCompressedPacket(num, buf) {
+PacketWriter.prototype._deflateNextPacket = function _deflateNextPacket() {
+  if (this._deflating) {
+    return;
+  }
+
+  var item = this._deflateQueue.shift();
+  var buf  = item[1];
+  var num  = item[0];
+  var len  = buf.length;
+  var self = this;
+
+  this._deflating = true;
+  Zlib.deflate(buf, function (err, data) {
+    if (err) {
+      self.emit('error', err);
+      return;
+    }
+
+    self._deflating = false;
+    self.emit('data', self._toCompressedPacket(num, data, len));
+    self._deflateNextPacket();
+  });
+};
+
+PacketWriter.prototype._toCompressedPacket = function _toCompressedPacket(num, buf, len) {
   var origBuffer = this._buffer;
   var origOffset = this._offset;
 
@@ -248,7 +283,7 @@ PacketWriter.prototype._toCompressedPacket = function _toCompressedPacket(num, b
 
   this.writeUnsignedNumber(3, buf.length);
   this.writeUnsignedNumber(1, num);
-  this.writeUnsignedNumber(3, 0);
+  this.writeUnsignedNumber(3, (len || 0));
   this.writeBuffer(buf);
 
   var packet = this._buffer;
@@ -259,6 +294,11 @@ PacketWriter.prototype._toCompressedPacket = function _toCompressedPacket(num, b
   return packet;
 };
 
+PacketWriter.prototype._toCompressedPacketAsync = function _toCompressedPacketAsync(num, buf) {
+  this._deflateQueue.push(buf);
+  this._deflateNextPacket();
+};
+
 PacketWriter.prototype._toPacket = function _toPacket(num, buf) {
   var origBuffer = this._buffer;
   var origOffset = this._offset;

lib/protocol/Protocol.js

@@ -308,6 +308,7 @@ Protocol.prototype._emitPacket = function(packet) {
   var packetWriter = new PacketWriter();
   var protocol     = this;
 
+  packetWriter.on('error', this.handleParserError.bind(this));
   packetWriter.on('data', function (data) {
     protocol.emit('data', data);
   });