stream: always defer 'readable' with nextTick

Emit 'readable' always in the next tick, resulting in a single
call to _read() per microtick. This removes the need for the
user to implement buffering if they wanted to call this.push()
multiple times in an asynchronous fashion, as this.push() triggers
this._read() call.

PR-URL: https://github.com/nodejs/node/pull/17979
Fixes: https://github.com/nodejs/node/issues/3203
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>

1 files changed, 12 insertions, 11 deletions

diff --git a/test/parallel/test-stream-readable-needReadable.js b/test/parallel/test-stream-readable-needReadable.js
index be397dc5dc..7058e123f0 100644
--- a/test/parallel/test-stream-readable-needReadable.js
+++ b/test/parallel/test-stream-readable-needReadable.js
@@ -38,7 +38,7 @@ asyncReadable.on('readable', common.mustCall(() => {
     // then we need to notify the reader on future changes.
     assert.strictEqual(asyncReadable._readableState.needReadable, true);
   }
-}, 3));
+}, 2));
 
 process.nextTick(common.mustCall(() => {
   asyncReadable.push('foooo');
@@ -46,8 +46,9 @@ process.nextTick(common.mustCall(() => {
 process.nextTick(common.mustCall(() => {
   asyncReadable.push('bar');
 }));
-process.nextTick(common.mustCall(() => {
+setImmediate(common.mustCall(() => {
   asyncReadable.push(null);
+  assert.strictEqual(asyncReadable._readableState.needReadable, false);
 }));
 
 const flowing = new Readable({
@@ -84,13 +85,13 @@ slowProducer.on('readable', common.mustCall(() => {
 
 process.nextTick(common.mustCall(() => {
   slowProducer.push('foo');
-}));
-process.nextTick(common.mustCall(() => {
-  slowProducer.push('foo');
-}));
-process.nextTick(common.mustCall(() => {
-  slowProducer.push('foo');
-}));
-process.nextTick(common.mustCall(() => {
-  slowProducer.push(null);
+  process.nextTick(common.mustCall(() => {
+    slowProducer.push('foo');
+    process.nextTick(common.mustCall(() => {
+      slowProducer.push('foo');
+      process.nextTick(common.mustCall(() => {
+        slowProducer.push(null);
+      }));
+    }));
+  }));
 }));