fix TFramedTransport residual continuation

There were many bugs in the current TFramedTransport.receiver caused by
merge mistakes and bad patches. Simplify the logic so it is easier to
reason about and prevent future issues.

- THRIFT-2194 Fixed one bug with residual not being set
- THRIFT-2205 Reverted the above fix (by accident) and broke it further
  by including InputBufferUnderrunError in TFramedTransport (which is
  incorrect)

This patch cleans up TFramedTransport.receiver by only have one hold
over buffer instead of two (frame + residual).

diff --git a/lib/nodejs/lib/thrift/transport.js b/lib/nodejs/lib/thrift/transport.js
index 8423f89..900472c 100644 (file)
--- a/lib/nodejs/lib/thrift/transport.js
+++ b/lib/nodejs/lib/thrift/transport.js
@@ -34,49 +34,36 @@ var TFramedTransport = exports.TFramedTransport = function(buffer, callback) {
   this.onFlush = callback;
 };
 TFramedTransport.receiver = function(callback) {
-  var frameLeft = 0,
-      framePos = 0,
-      frame = null;
   var residual = null;
 
   return function(data) {
     // Prepend any residual data from our previous read
     if (residual) {
-      var dat = new Buffer(data.length + residual.length);
-      residual.copy(dat, 0, 0);
-      data.copy(dat, residual.length, 0);
+      data = Buffer.concat([residual, data]);
       residual = null;
     }
 
     // framed transport
     while (data.length) {
-      if (frameLeft === 0) {
-        // TODO assumes we have all 4 bytes
-        if (data.length < 4) {
-          console.log("Expecting > 4 bytes, found only " + data.length);
-          residual = data;
-          throw new InputBufferUnderrunError();
-          //throw Error("Expecting > 4 bytes, found only " + data.length);
-        }
-        frameLeft = binary.readI32(data, 0);
-        frame = new Buffer(frameLeft);
-        framePos = 0;
-        data = data.slice(4, data.length);
+      if (data.length < 4) {
+        // Not enough bytes to continue, save and resume on next packet
+        residual = data;
+        return;
       }
-      
-      if (data.length >= frameLeft) {
-        data.copy(frame, framePos, 0, frameLeft);
-        data = data.slice(frameLeft, data.length);
-      
-        frameLeft = 0;
-        callback(new TFramedTransport(frame));
-      } else if (data.length) {
-        data.copy(frame, framePos, 0, data.length);
-        frameLeft -= data.length;
-        framePos += data.length;
-        data = data.slice(data.length, data.length);
-        throw new InputBufferUnderrunError();
+      var frameSize = binary.readI32(data, 0);
+      if (data.length < 4 + frameSize) {
+        // Not enough bytes to continue, save and resume on next packet
+        residual = data;
+        return;
       }
+
+      var frame = data.slice(4, 4 + frameSize);
+      residual = data.slice(4 + frameSize);
+
+      callback(new TFramedTransport(frame));
+
+      data = residual;
+      residual = null;
     }
   };
 };