Merge pull request #60 from Myztiq/demux-fix

Converted demux to work on.('data')

Author: apocas

lib/modem.js

@@ -314,22 +314,45 @@ Modem.prototype.buildPayload = function(err, isStream, statusCodes, openStdin, r
 };
 
 Modem.prototype.demuxStream = function(stream, stdout, stderr) {
-  var header = null;
-
-  stream.on('readable', function() {
-    header = header || stream.read(8);
-    while (header !== null) {
-      var type = header.readUInt8(0);
-      var payload = stream.read(header.readUInt32BE(4));
-      if (payload === null) break;
-      if (type == 2) {
-        stderr.write(payload);
-      } else {
-        stdout.write(payload);
+  var nextDataType = null;
+  var nextDataLength = null;
+  var buffer = new Buffer('');
+  function processData(data) {
+    if (data) {
+      buffer = Buffer.concat([buffer, data]);
+    }
+    if (!nextDataType) {
+      if (buffer.length >= 8) {
+        var header = bufferSlice(8);
+        nextDataType = header.readUInt8(0);
+        nextDataLength = header.readUInt32BE(4);
+        // It's possible we got a "data" that contains multiple messages
+        // Process the next one
+        processData();
+      }
+    } else {
+      if (buffer.length >= nextDataLength) {
+        var content = bufferSlice(nextDataLength);
+        if (nextDataType === 1) {
+          stdout.write(content);
+        } else {
+          stderr.write(content);
+        }
+        nextDataType = null;
+        // It's possible we got a "data" that contains multiple messages
+        // Process the next one
+        processData();
       }
-      header = stream.read(8);
     }
-  });
+  }
+
+  function bufferSlice (end) {
+    var out = buffer.slice(0, end);
+    buffer = new Buffer(buffer.slice(end, buffer.length));
+    return out;
+  }
+
+  stream.on('data', processData);
 };
 
 Modem.prototype.followProgress = function(stream, onFinished, onProgress) {