stream: make _read() be called indefinitely if the user wants so

Fixes: #26097

PR-URL: #26135
Reviewed-By: Luigi Pinca <[email protected]>
Reviewed-By: Anna Henningsen <[email protected]>
Reviewed-By: James M Snell <[email protected]>
Reviewed-By: Ruben Bridgewater <[email protected]>

lib/_stream_readable.js

@@ -494,6 +494,7 @@ Readable.prototype.read = function(n) {
 };
 
 function onEofChunk(stream, state) {
+  debug('onEofChunk');
   if (state.ended) return;
   if (state.decoder) {
     var chunk = state.decoder.end();
@@ -524,6 +525,7 @@ function onEofChunk(stream, state) {
 // a nextTick recursion warning, but that's not so bad.
 function emitReadable(stream) {
   var state = stream._readableState;
+  debug('emitReadable', state.needReadable, state.emittedReadable);
   state.needReadable = false;
   if (!state.emittedReadable) {
     debug('emitReadable', state.flowing);
@@ -537,6 +539,7 @@ function emitReadable_(stream) {
   debug('emitReadable_', state.destroyed, state.length, state.ended);
   if (!state.destroyed && (state.length || state.ended)) {
     stream.emit('readable');
+    state.emittedReadable = false;
   }
 
   // The stream needs another readable event if

test/parallel/test-stream-readable-infinite-read.js

@@ -0,0 +1,32 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const { Readable } = require('stream');
+
+const buf = Buffer.alloc(8192);
+
+const readable = new Readable({
+  read: common.mustCall(function() {
+    this.push(buf);
+  }, 31)
+});
+
+let i = 0;
+
+readable.on('readable', common.mustCall(function() {
+  if (i++ === 10) {
+    // We will just terminate now.
+    process.removeAllListeners('readable');
+    return;
+  }
+
+  const data = readable.read();
+  // TODO(mcollina): there is something odd in the highWaterMark logic
+  // investigate.
+  if (i === 1) {
+    assert.strictEqual(data.length, 8192 * 2);
+  } else {
+    assert.strictEqual(data.length, 8192 * 3);
+  }
+}, 11));