From 2bf8f2352b03eb66696a1fe821d73f92b5791e4f Mon Sep 17 00:00:00 2001
From: Simo Kinnunen <simo@shoqolate.com>
Date: Tue, 1 Apr 2014 17:54:31 +0900
Subject: [PATCH] Fix major oversight in MessageStream that was causing
 multi-chunk messages to fail.

---
 lib/wire/messagestream.js  | 48 ++++++++++++++++++--------------------
 test/wire/messagestream.js |  8 +++++--
 2 files changed, 29 insertions(+), 27 deletions(-)

diff --git a/lib/wire/messagestream.js b/lib/wire/messagestream.js
index 08588e58..ccd672a7 100644
--- a/lib/wire/messagestream.js
+++ b/lib/wire/messagestream.js
@@ -14,35 +14,33 @@ util.inherits(DelimitedStream, stream.Transform)
 DelimitedStream.prototype._transform = function(chunk, encoding, done) {
   this._buffer = Buffer.concat([this._buffer, chunk])
 
-  var lo = 0
-  var hi = this._buffer.length
-
-  while (lo < hi) {
+  while (this._buffer.length) {
     if (this._readingLength) {
-      while (lo < hi) {
-        var byte = this._buffer[lo++]
-        if (byte & (1 << 7)) {
-          this._length += (byte & 0x7f) << (7 * this._lengthIndex)
-          this._lengthIndex += 1
-          this._readingLength = true
-        }
-        else {
-          this._length += (byte & 0x7f) << (7 * this._lengthIndex)
-          this._lengthIndex = 0
-          this._readingLength = false
-          break
-        }
+      var byte = this._buffer[0]
+      if (byte & (1 << 7)) {
+        this._length += (byte & 0x7f) << (7 * this._lengthIndex)
+        this._lengthIndex += 1
+        this._readingLength = true
       }
+      else {
+        this._length += (byte & 0x7f) << (7 * this._lengthIndex)
+        this._lengthIndex = 0
+        this._readingLength = false
+      }
+      this._buffer = this._buffer.slice(1)
     }
 
-    if (!this._readingLength && lo + this._length <= hi) {
-      this.push(chunk.slice(lo, lo + this._length))
-      lo += this._length
-      this._length = 0
-      this._readingLength = true
-    }
-    else {
-      break
+    if (!this._readingLength) {
+      if (this._length <= this._buffer.length) {
+        this.push(this._buffer.slice(0, this._length))
+        this._buffer = this._buffer.slice(this._length)
+        this._length = 0
+        this._readingLength = true
+      }
+      else {
+        // Wait for more chunks
+        break
+      }
     }
   }
 
diff --git a/test/wire/messagestream.js b/test/wire/messagestream.js
index 265ebc76..906de60b 100644
--- a/test/wire/messagestream.js
+++ b/test/wire/messagestream.js
@@ -21,11 +21,15 @@ describe('MessageStream', function() {
       var ds = new ms.DelimitedStream()
       var spy = sinon.spy()
       ds.on('data', spy)
-      ds.write(new Buffer([1]))
+      ds.write(new Buffer([3]))
       expect(spy).to.not.have.been.called
       ds.write(new Buffer([0x66]))
+      expect(spy).to.not.have.been.called
+      ds.write(new Buffer([0x65]))
+      expect(spy).to.not.have.been.called
+      ds.write(new Buffer([0x64]))
       expect(spy).to.have.been.calledOnce
-      expect(spy.firstCall.args).to.eql([new Buffer([0x66])])
+      expect(spy.firstCall.args).to.eql([new Buffer([0x66, 0x65, 0x64])])
     })
 
     it('should read varint32 properly', function() {