stream: check type and range of highWaterMark

lib/_stream_readable.js

@@ -31,6 +31,7 @@ const util = require('util');
 const debug = util.debuglog('stream');
 const BufferList = require('internal/streams/BufferList');
 const destroyImpl = require('internal/streams/destroy');
+const { getHighWaterMark } = require('internal/streams/state');
 const errors = require('internal/errors');
 var StringDecoder;
 
@@ -75,19 +76,7 @@ function ReadableState(options, stream) {
 
   // the point at which it stops calling _read() to fill the buffer
   // Note: 0 is a valid value, means "don't call _read preemptively ever"
-  var hwm = options.highWaterMark;
-  var readableHwm = options.readableHighWaterMark;
-  var defaultHwm = this.objectMode ? 16 : 16 * 1024;
-
-  if (hwm || hwm === 0)
-    this.highWaterMark = hwm;
-  else if (isDuplex && (readableHwm || readableHwm === 0))
-    this.highWaterMark = readableHwm;
-  else
-    this.highWaterMark = defaultHwm;
-
-  // cast to ints.
-  this.highWaterMark = Math.floor(this.highWaterMark);
+  this.highWaterMark = getHighWaterMark(this, options, isDuplex);
 
   // A linked list is used to store data chunks instead of an array because the
   // linked list can remove elements from the beginning faster than

lib/_stream_writable.js

@@ -33,6 +33,7 @@ const internalUtil = require('internal/util');
 const Stream = require('stream');
 const { Buffer } = require('buffer');
 const destroyImpl = require('internal/streams/destroy');
+const { getHighWaterMark } = require('internal/streams/state');
 const errors = require('internal/errors');
 
 util.inherits(Writable, Stream);
@@ -59,19 +60,7 @@ function WritableState(options, stream) {
   // the point at which write() starts returning false
   // Note: 0 is a valid value, means that we always return false if
   // the entire buffer is not flushed immediately on write()
-  var hwm = options.highWaterMark;
-  var writableHwm = options.writableHighWaterMark;
-  var defaultHwm = this.objectMode ? 16 : 16 * 1024;
-
-  if (hwm || hwm === 0)
-    this.highWaterMark = hwm;
-  else if (isDuplex && (writableHwm || writableHwm === 0))
-    this.highWaterMark = writableHwm;
-  else
-    this.highWaterMark = defaultHwm;
-
-  // cast to ints.
-  this.highWaterMark = Math.floor(this.highWaterMark);
+  this.highWaterMark = getHighWaterMark(this, options, isDuplex);
 
   // if _final has been called
   this.finalCalled = false;

lib/internal/streams/state.js

@@ -0,0 +1,35 @@
+'use strict';
+
+const stream = require('stream');
+const errors = require('internal/errors');
+
+function getHighWaterMarkFromProperty(hwm, name) {
+  if (typeof hwm !== 'number' || hwm < 0)
+    throw new errors.TypeError('ERR_INVALID_OPT_VALUE', name, hwm);
+  return Math.floor(hwm);
+}
+
+function getHighWaterMark(state, options, isDuplex) {
+  if (options.highWaterMark != null && !Number.isNaN(options.highWaterMark)) {
+    return getHighWaterMarkFromProperty(options.highWaterMark, 'highWaterMark');
+  } else if (isDuplex) {
+    if (state instanceof stream.Readable.ReadableState) {
+      if (options.readableHighWaterMark != null &&
+          !Number.isNaN(options.readableHighWaterMark)) {
+        return getHighWaterMarkFromProperty(options.readableHighWaterMark,
+                                            'readableHighWaterMark');
+      }
+    } else if (options.writableHighWaterMark != null &&
+               !Number.isNaN(options.writableHighWaterMark)) {
+      return getHighWaterMarkFromProperty(options.writableHighWaterMark,
+                                          'writableHighWaterMark');
+    }
+  }
+
+  // Default value
+  return state.objectMode ? 16 : 16 * 1024;
+}
+
+module.exports = {
+  getHighWaterMark
+};

node.gyp

@@ -141,6 +141,7 @@
       'lib/internal/streams/BufferList.js',
       'lib/internal/streams/legacy.js',
       'lib/internal/streams/destroy.js',
+      'lib/internal/streams/state.js',
       'lib/internal/wrap_js_stream.js',
       'deps/v8/tools/splaytree.js',
       'deps/v8/tools/codemap.js',

test/parallel/test-streams-highwatermark.js

@@ -1,8 +1,9 @@
 'use strict';
-require('../common');
+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.
+// for highWaterMark which exceed the range of signed 32 bit integers and
+// rejects invalid values.
 
 const assert = require('assert');
 const stream = require('stream');
@@ -16,3 +17,15 @@ 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]) {
+  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"`
+    });
+  }
+}