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,8 @@ 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, 'readableHighWaterMark',
+                                        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,8 @@ 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, 'writableHighWaterMark',
+                                        isDuplex);
 
   // if _final has been called
   this.finalCalled = false;

lib/internal/streams/state.js

@@ -0,0 +1,26 @@
+'use strict';
+
+const errors = require('internal/errors');
+
+function getHighWaterMark(state, options, duplexKey, isDuplex) {
+  let hwm = options.highWaterMark;
+  if (hwm != null) {
+    if (typeof hwm !== 'number' || !(hwm >= 0))
+      throw new errors.TypeError('ERR_INVALID_OPT_VALUE', 'highWaterMark', hwm);
+    return Math.floor(hwm);
+  } else if (isDuplex) {
+    hwm = options[duplexKey];
+    if (hwm != null) {
+      if (typeof hwm !== 'number' || !(hwm >= 0))
+        throw new errors.TypeError('ERR_INVALID_OPT_VALUE', duplexKey, hwm);
+      return Math.floor(hwm);
+    }
+  }
+
+  // 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-stream-transform-split-highwatermark.js

@@ -1,5 +1,5 @@
 'use strict';
-require('../common');
+const common = require('../common');
 const assert = require('assert');
 
 const { Transform, Readable, Writable } = require('stream');
@@ -54,14 +54,33 @@ testTransform(0, 0, {
   writableHighWaterMark: 777,
 });
 
-// test undefined, null, NaN
-[undefined, null, NaN].forEach((v) => {
+// test undefined, null
+[undefined, null].forEach((v) => {
   testTransform(DEFAULT, DEFAULT, { readableHighWaterMark: v });
   testTransform(DEFAULT, DEFAULT, { writableHighWaterMark: v });
   testTransform(666, DEFAULT, { highWaterMark: v, readableHighWaterMark: 666 });
   testTransform(DEFAULT, 777, { highWaterMark: v, writableHighWaterMark: 777 });
 });
 
+// test NaN
+{
+  common.expectsError(() => {
+    new Transform({ readableHighWaterMark: NaN });
+  }, {
+    type: TypeError,
+    code: 'ERR_INVALID_OPT_VALUE',
+    message: 'The value "NaN" is invalid for option "readableHighWaterMark"'
+  });
+
+  common.expectsError(() => {
+    new Transform({ writableHighWaterMark: NaN });
+  }, {
+    type: TypeError,
+    code: 'ERR_INVALID_OPT_VALUE',
+    message: 'The value "NaN" is invalid for option "writableHighWaterMark"'
+  });
+}
+
 // test non Duplex streams ignore the options
 {
   const r = new Readable({ readableHighWaterMark: 666 });

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, 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"`
+    });
+  }
+}