Skip to content

Commit

Permalink
stream: fix regression introduced in #26059
Browse files Browse the repository at this point in the history
In #26059, we introduced a bug that caused 'readable' to be nextTicked
on EOF of a ReadableStream. This breaks the dicer module on CITGM.
That change was partially reverted to still fix the bug in #25810 and
not break dicer.

See: #26059
Fixes: #25810

PR-URL: #26643
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>
  • Loading branch information
mcollina committed Mar 16, 2019
1 parent c5e619b commit 269103a
Show file tree
Hide file tree
Showing 7 changed files with 28 additions and 33 deletions.
22 changes: 17 additions & 5 deletions lib/_stream_readable.js
Original file line number Diff line number Diff line change
Expand Up @@ -512,12 +512,24 @@ function onEofChunk(stream, state) {
}
}
state.ended = true;
state.needReadable = false;

// We are not protecting if emittedReadable = true,
// so 'readable' gets scheduled anyway.
state.emittedReadable = true;
process.nextTick(emitReadable_, stream);
if (state.sync) {
// If we are sync, wait until next tick to emit the data.
// Otherwise we risk emitting data in the flow()
// the readable code triggers during a read() call
emitReadable(stream);
} else {
// Emit 'readable' now to make sure it gets picked up.
state.needReadable = false;
state.emittedReadable = true;
// We have to emit readable now that we are EOF. Modules
// in the ecosystem (e.g. dicer) rely on this event being sync.
if (state.ended) {
emitReadable_(stream);
} else {
process.nextTick(emitReadable_, stream);
}
}
}

// Don't emit readable right away in sync mode, because this can trigger
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ const assert = require('assert');
break;
assert.strictEqual(chunk.toString(), 'content');
}
}, 2));
}));
}

{
Expand All @@ -78,7 +78,7 @@ const assert = require('assert');
break;
assert.strictEqual(chunk.toString(), 'content');
}
}, 2));
}));
}

{
Expand All @@ -94,7 +94,7 @@ const assert = require('assert');
break;
assert.strictEqual(chunk.toString(), 'content');
}
}, 2));
}));

t.push('content');
t.push(null);
Expand Down
13 changes: 1 addition & 12 deletions test/parallel/test-stream-readable-emittedReadable.js
Original file line number Diff line number Diff line change
Expand Up @@ -43,23 +43,12 @@ const noRead = new Readable({
read: () => {}
});

noRead.once('readable', common.mustCall(() => {
noRead.on('readable', common.mustCall(() => {
// emittedReadable should be true when the readable event is emitted
assert.strictEqual(noRead._readableState.emittedReadable, true);
noRead.read(0);
// emittedReadable is not reset during read(0)
assert.strictEqual(noRead._readableState.emittedReadable, true);

noRead.on('readable', common.mustCall(() => {
// The second 'readable' is emitted because we are ending

// emittedReadable should be true when the readable event is emitted
assert.strictEqual(noRead._readableState.emittedReadable, false);
noRead.read(0);
// emittedReadable is not reset during read(0)
assert.strictEqual(noRead._readableState.emittedReadable, false);

}));
}));

noRead.push('foo');
Expand Down
2 changes: 1 addition & 1 deletion test/parallel/test-stream-readable-needReadable.js
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ readable.on('readable', common.mustCall(() => {
// When the readable event fires, needReadable is reset.
assert.strictEqual(readable._readableState.needReadable, false);
readable.read();
}, 2));
}));

// If a readable listener is attached, then a readable event is needed.
assert.strictEqual(readable._readableState.needReadable, true);
Expand Down
2 changes: 1 addition & 1 deletion test/parallel/test-stream-readable-reading-readingMore.js
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ const Readable = require('stream').Readable;
assert.strictEqual(state.reading, false);
}

const expectedReadingMore = [true, false, false];
const expectedReadingMore = [true, true, false];
readable.on('readable', common.mustCall(() => {
// There is only one readingMore scheduled from on('data'),
// after which everything is governed by the .read() call
Expand Down
2 changes: 1 addition & 1 deletion test/parallel/test-stream2-httpclient-response-end.js
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ const server = http.createServer(function(req, res) {
while ((chunk = res.read()) !== null) {
data += chunk;
}
}, 2));
}));
res.on('end', common.mustCall(function() {
console.log('end event');
assert.strictEqual(msg, data);
Expand Down
14 changes: 4 additions & 10 deletions test/parallel/test-stream2-transform.js
Original file line number Diff line number Diff line change
Expand Up @@ -321,16 +321,10 @@ const Transform = require('_stream_transform');

pt.end();

// The next readable is emitted on the next tick.
assert.strictEqual(emits, 0);

process.on('nextTick', function() {
assert.strictEqual(emits, 1);
assert.strictEqual(pt.read(5).toString(), 'l');
assert.strictEqual(pt.read(5), null);

assert.strictEqual(emits, 1);
});
assert.strictEqual(emits, 1);
assert.strictEqual(pt.read(5).toString(), 'l');
assert.strictEqual(pt.read(5), null);
assert.strictEqual(emits, 1);
}

{
Expand Down

0 comments on commit 269103a

Please sign in to comment.