stream: inline needMoreData function

Inline the needMoreData function since it has only one call place.
Update the related comment.
Add a test for the edge case where HWM=0 and state.length=0.
Add a test for ReadableStream.read(n) method's edge case where
n, HWM and state.length are all zero.
This proves that there is no easy way to simplify the check at
https://github.com/nodejs/node/blob/master/lib/_stream_readable.js#L440

Fixes: nodejs#19893
Refs: nodejs#19896

lib/_stream_readable.js

@@ -270,7 +270,11 @@ function readableAddChunk(stream, chunk, encoding, addToFront, skipChunkCheck) {
     }
   }
 
-  return needMoreData(state);
+  // We can push more data if we are below the highWaterMark.
+  // Also, if we have no data yet, we can stand some more bytes.
+  // This is to work around cases where hwm=0, such as the repl.
+  return !state.ended &&
+    (state.length < state.highWaterMark || state.length === 0);
 }
 
 function addChunk(stream, state, chunk, addToFront) {
@@ -304,19 +308,6 @@ function chunkInvalid(state, chunk) {
 }
 
 
-// We can push more data if we are below the highWaterMark.
-// Also, if we have no data yet, we can stand some
-// more bytes.  This is to work around cases where hwm=0,
-// such as the repl.  Also, if the push() triggered a
-// readable event, and the user called read(largeNumber) such that
-// needReadable was set, then we ought to push more, so that another
-// 'readable' event will be triggered.
-function needMoreData(state) {
-  return !state.ended &&
-         (state.length < state.highWaterMark ||
-          state.length === 0);
-}
-
 Readable.prototype.isPaused = function() {
   return this._readableState.flowing === false;
 };

test/parallel/test-streams-highwatermark.js

@@ -1,31 +1,58 @@
 'use strict';
 const common = require('../common');
 
-// This test ensures that the stream implementation correctly handles values
-// for highWaterMark which exceed the range of signed 32 bit integers and
-// rejects invalid values.
-
 const assert = require('assert');
 const stream = require('stream');
 
-// This number exceeds the range of 32 bit integer arithmetic but should still
-// be handled correctly.
-const ovfl = Number.MAX_SAFE_INTEGER;
-
-const readable = stream.Readable({ highWaterMark: ovfl });
-assert.strictEqual(readable._readableState.highWaterMark, ovfl);
-
-const writable = stream.Writable({ highWaterMark: ovfl });
-assert.strictEqual(writable._writableState.highWaterMark, ovfl);
-
-for (const invalidHwm of [true, false, '5', {}, -5, NaN]) {
-  for (const type of [stream.Readable, stream.Writable]) {
-    common.expectsError(() => {
-      type({ highWaterMark: invalidHwm });
-    }, {
-      type: TypeError,
-      code: 'ERR_INVALID_OPT_VALUE',
-      message: `The value "${invalidHwm}" is invalid for option "highWaterMark"`
-    });
+{
+  // This test ensures that the stream implementation correctly handles values
+  // for highWaterMark which exceed the range of signed 32 bit integers and
+  // rejects invalid values.
+
+  // This number exceeds the range of 32 bit integer arithmetic but should still
+  // be handled correctly.
+  const ovfl = Number.MAX_SAFE_INTEGER;
+
+  const readable = stream.Readable({ highWaterMark: ovfl });
+  assert.strictEqual(readable._readableState.highWaterMark, ovfl);
+
+  const writable = stream.Writable({ highWaterMark: ovfl });
+  assert.strictEqual(writable._writableState.highWaterMark, ovfl);
+
+  for (const invalidHwm of [true, false, '5', {}, -5, NaN]) {
+    for (const type of [stream.Readable, stream.Writable]) {
+      common.expectsError(() => {
+        type({ highWaterMark: invalidHwm });
+      }, {
+        type: TypeError,
+        code: 'ERR_INVALID_OPT_VALUE',
+        message:
+          `The value "${invalidHwm}" is invalid for option "highWaterMark"`
+      });
+    }
+  }
+}
+
+{
+  // This test ensures that the push method's implementation
+  // correctly handles the edge case where the highWaterMark and
+  // the state.length are both zero
+
+  const readable = stream.Readable({ highWaterMark: 0 });
+
+  for (let i = 0; i < 3; i++) {
+    const needMoreData = readable.push();
+    assert.strictEqual(needMoreData, true);
   }
 }
+
+{
+  // This test ensures that the read(n) method's implementation
+  // correctly handles the edge case where the highWaterMark, state.length
+  // and n are all zero
+
+  const readable = stream.Readable({ highWaterMark: 0 });
+
+  readable._read = common.mustCall();
+  readable.read(0);
+}