Merge pull request #40 from mvachhar/issue30

Fix Issue #30 where ReadableStreamBuffer stalls in certain cases.
samcday · May 21, 2018 · 802857a · 802857a
2 parents 78d2e1a + a2eb31d
commit 802857a
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 3 deletions.
diff --git a/lib/readable_streambuffer.js b/lib/readable_streambuffer.js
@@ -19,6 +19,7 @@ var ReadableStreamBuffer = module.exports = function(opts) {
 
   var size = 0;
   var buffer = new Buffer(initialSize);
+  var allowPush = false;
 
   var sendData = function() {
     var amount = Math.min(chunkSize, size);
@@ -30,6 +31,7 @@ var ReadableStreamBuffer = module.exports = function(opts) {
       buffer.copy(chunk, 0, 0, amount);
 
       sendMore = that.push(chunk) !== false;
+      allowPush = sendMore;
 
       buffer.copy(buffer, 0, amount, size);
       size -= amount;
@@ -76,6 +78,12 @@ var ReadableStreamBuffer = module.exports = function(opts) {
     }
   };
 
+  var kickSendDataTask = function () {
+    if (!sendData.timeout && allowPush) {
+      sendData.timeout = setTimeout(sendData, frequency);
+    }
+  }
+
   this.put = function(data, encoding) {
     if (that.stopped) {
       throw new Error('Tried to write data to a stopped ReadableStreamBuffer');
@@ -93,12 +101,13 @@ var ReadableStreamBuffer = module.exports = function(opts) {
       buffer.write(data, size, encoding || 'utf8');
       size += dataSizeInBytes;
     }
+
+    kickSendDataTask();
   };
 
   this._read = function() {
-    if (!sendData.timeout) {
-      sendData.timeout = setTimeout(sendData, frequency);
-    }
+    allowPush = true;
+    kickSendDataTask();
   };
 };
 

diff --git a/test/readable_streambuffer.test.js b/test/readable_streambuffer.test.js
@@ -55,6 +55,23 @@ describe('A default ReadableStreamBuffer', function() {
     this.buffer.stop();
   });
 
+  it('pushes new data even if read when empty', function(done) {
+    var that = this;
+    var str = '';
+    this.buffer.on('readable', function() {
+      str += (that.buffer.read() || new Buffer(0)).toString('utf8');
+    });
+    this.buffer.on('end', function() {
+      expect(str).to.equal(fixtures.unicodeString);
+      done();
+    });
+
+    setTimeout(function() {
+      that.buffer.put(fixtures.unicodeString);
+      that.buffer.stop();
+    }, streamBuffer.DEFAULT_FREQUENCY + 1);
+  });
+
   describe('when writing binary data', function() {
     beforeEach(function(done) {
       var that = this;