Skip to content

Commit

Permalink
Fix stream becoming readable again
Browse files Browse the repository at this point in the history
  • Loading branch information
johanneswuerbach committed Oct 29, 2019
1 parent a051869 commit eaebbd9
Show file tree
Hide file tree
Showing 2 changed files with 40 additions and 21 deletions.
27 changes: 6 additions & 21 deletions lib/mux.js
Original file line number Diff line number Diff line change
Expand Up @@ -55,26 +55,11 @@ Mux.prototype._readIncoming = function() {
// are empty, or we exhaust our ability to accept chunks.
function roundrobin(streams) {
var s;
// if there's just one incoming stream we don't have to
// go through all the dequeue/enqueueing
if (streams.length === 1) {
s = streams.shift();
while (accepting) {
var chunk = s.read();
if (chunk !== null) {
accepting = out.write(chunk);
}
else break;
}
if (!accepting) streams.push(s);
}
else {
while (accepting && (s = streams.shift())) {
var chunk = s.read();
if (chunk !== null) {
accepting = out.write(chunk);
streams.push(s);
}
while (accepting && (s = streams.shift())) {
var chunk = s.read();
if (chunk !== null) {
accepting = out.write(chunk);
streams.push(s);
}
}
}
Expand Down Expand Up @@ -105,7 +90,7 @@ Mux.prototype._readIncoming = function() {

Mux.prototype._scheduleRead = function() {
var self = this;

if (!self.scheduledRead) {
schedule(function() {
self.scheduledRead = false;
Expand Down
34 changes: 34 additions & 0 deletions test/mux.js
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,40 @@ test("single input", function(done) {
input.end();
});

test("single input, resuming stream", function(done) {
var input = stream();
var output = stream();
input.on('end', function() { output.end() });

var mux = new Mux(output);
mux.pipeFrom(input);

// Streams might be blocked and become readable again, simulate this
// using a special read function and a marker
var data = [1,2,3,4,'skip',6,7,8,9];

var oldRead = input.read;
input.read = function(size) {
var val = oldRead.call(input, size)

if (val === 'skip') {
input.emit('readable');
return null
}

return val;
}

data.forEach(input.write.bind(input));

readAllObjects(output, function(vals) {
assert.deepEqual([1,2,3,4,6,7,8,9], vals);
done();
});

input.end();
});

test("two sequential inputs", function(done) {
var input1 = stream();
var input2 = stream();
Expand Down

0 comments on commit eaebbd9

Please sign in to comment.