stream: remove useage of _readableState/_writableState.highWaterMark

Replace with .readableHighWaterMark or .writableHighWaterMark getters. Ref: #445.

Author: calvinmetcalf

doc/api/stream.md

@@ -437,6 +437,14 @@ process.nextTick(() => {
 
 See also: [`writable.cork()`][].
 
+##### writable.writableHighWaterMark
+<!-- YAML
+added: REPLACEME
+-->
+
+Return the value of `highWaterMark` passed when constructing this
+`Writable`.
+
 ##### writable.write(chunk[, encoding][, callback])
 <!-- YAML
 added: v0.9.4
@@ -879,6 +887,14 @@ to prevent memory leaks.
 never closed until the Node.js process exits, regardless of the specified
 options.
 
+##### readable.readableHighWaterMark
+<!-- YAML
+added: REPLACEME
+-->
+
+Return the value of `highWaterMark` passed when constructing this
+`Readable`.
+
 ##### readable.read([size])
 <!-- YAML
 added: v0.9.4

lib/_http_server.js

@@ -373,13 +373,13 @@ function connectionListener(socket) {
 function updateOutgoingData(socket, state, delta) {
  state.outgoingData += delta;
  if (socket._paused &&
- state.outgoingData < socket._writableState.highWaterMark) {
+ state.outgoingData < socket.writeHWM) {
  return socketOnDrain(socket, state);
  }
 }
 
 function socketOnDrain(socket, state) {
- var needPause = state.outgoingData > socket._writableState.highWaterMark;
+ var needPause = state.outgoingData > socket.writeHWM;
 
  // If we previously paused, then start reading again.
  if (socket._paused && !needPause) {
@@ -569,7 +569,7 @@ function parserOnIncoming(server, socket, state, req, keepAlive) {
  // pipelined requests that may never be resolved.
  if (!socket._paused) {
  var ws = socket._writableState;
- if (ws.needDrain || state.outgoingData >= ws.highWaterMark) {
+ if (ws.needDrain || state.outgoingData >= socket.writableHighWaterMark) {
  socket._paused = true;
  // We also need to pause the parser, but don't do that until after
  // the call to execute, because we may still be processing the last

lib/_stream_duplex.js

@@ -34,11 +34,14 @@ const Writable = require('_stream_writable');
 
 util.inherits(Duplex, Readable);
 
-var keys = Object.keys(Writable.prototype);
-for (var v = 0; v < keys.length; v++) {
- var method = keys[v];
- if (!Duplex.prototype[method])
- Duplex.prototype[method] = Writable.prototype[method];
+{
+ // avoid scope creep, the keys array can then be collected
+ const keys = Object.keys(Writable.prototype);
+ for (var v = 0; v < keys.length; v++) {
+ const method = keys[v];
+ if (!Duplex.prototype[method])
+ Duplex.prototype[method] = Writable.prototype[method];
+ }
 }
 
 function Duplex(options) {
@@ -61,6 +64,16 @@ function Duplex(options) {
  this.once('end', onend);
 }
 
+Object.defineProperty(Duplex.prototype, 'writableHighWaterMark', {
+ // making it explicit this property is not enumerable
+ // because otherwise some prototype manipulation in
+ // userland will fail
+ enumerable: false,
+ get: function() {
+ return this._writableState.highWaterMark;
+ }
+});
+
 // the no-half-open enforcer
 function onend() {
  // if we allow half-open state, or if the writable side ended,

lib/_stream_readable.js

@@ -916,6 +916,15 @@ Readable.prototype.wrap = function(stream) {
  return self;
 };
 
+Object.defineProperty(Readable.prototype, 'readableHighWaterMark', {
+ // making it explicit this property is not enumerable
+ // because otherwise some prototype manipulation in
+ // userland will fail
+ enumerable: false,
+ get: function() {
+ return this._readableState.highWaterMark;
+ }
+});
 
 // exposed for testing purposes only.
 Readable._fromList = fromList;

lib/_stream_writable.js

@@ -334,6 +334,16 @@ function decodeChunk(state, chunk, encoding) {
  return chunk;
 }
 
+Object.defineProperty(Writable.prototype, 'writableHighWaterMark', {
+ // making it explicit this property is not enumerable
+ // because otherwise some prototype manipulation in
+ // userland will fail
+ enumerable: false,
+ get: function() {
+ return this._writableState.highWaterMark;
+ }
+});
+
 // if we're already writing something, then just put this
 // in the queue, and wait our turn. Otherwise, call _write
 // If we return false, then we need a drain event, so set that flag.

lib/fs.js

@@ -2072,7 +2072,7 @@ ReadStream.prototype._read = function(n) {
 
  if (!pool || pool.length - pool.used < kMinPoolSpace) {
  // discard the old pool.
- allocNewPool(this._readableState.highWaterMark);
+ allocNewPool(this.readableHighWaterMark);
  }
 
  // Grab another reference to the pool in the case that while we're

test/parallel/test-http-pipeline-regr-3508.js

@@ -30,7 +30,7 @@ const server = http.createServer(function(req, res) {
  res.end(chunk);
  }
  size += res.outputSize;
- if (size <= req.socket._writableState.highWaterMark) {
+ if (size <= req.socket.writableHighWaterMark) {
  more();
  return;
  }

test/parallel/test-stream-big-packet.js

@@ -51,7 +51,7 @@ s1.pipe(s3);
 s2.pipe(s3, { end: false });
 
 // We must write a buffer larger than highWaterMark
-const big = Buffer.alloc(s1._writableState.highWaterMark + 1, 'x');
+const big = Buffer.alloc(s1.writableHighWaterMark + 1, 'x');
 
 // Since big is larger than highWaterMark, it will be buffered internally.
 assert(!s1.write(big));

test/parallel/test-stream-readable-flow-recursion.js

@@ -68,7 +68,7 @@ flow(stream, 5000, function() {
 process.on('exit', function(code) {
  assert.strictEqual(reads, 2);
  // we pushed up the high water mark
- assert.strictEqual(stream._readableState.highWaterMark, 8192);
+ assert.strictEqual(stream.readableHighWaterMark, 8192);
  // length is 0 right now, because we pulled it all out.
  assert.strictEqual(stream._readableState.length, 0);
  assert(!code);

test/parallel/test-stream-transform-split-objectmode.js

@@ -29,8 +29,12 @@ const parser = new Transform({ readableObjectMode: true });
 
 assert(parser._readableState.objectMode);
 assert(!parser._writableState.objectMode);
-assert.strictEqual(parser._readableState.highWaterMark, 16);
-assert.strictEqual(parser._writableState.highWaterMark, 16 * 1024);
+assert.strictEqual(parser.readableHighWaterMark, 16);
+assert.strictEqual(parser.writableHighWaterMark, 16 * 1024);
+assert.strictEqual(parser.readableHighWaterMark,
+ parser._readableState.highWaterMark);
+assert.strictEqual(parser.writableHighWaterMark,
+ parser._writableState.highWaterMark);
 
 parser._transform = function(chunk, enc, callback) {
  callback(null, { val: chunk[0] });
@@ -53,8 +57,12 @@ const serializer = new Transform({ writableObjectMode: true });
 
 assert(!serializer._readableState.objectMode);
 assert(serializer._writableState.objectMode);
-assert.strictEqual(serializer._readableState.highWaterMark, 16 * 1024);
-assert.strictEqual(serializer._writableState.highWaterMark, 16);
+assert.strictEqual(serializer.readableHighWaterMark, 16 * 1024);
+assert.strictEqual(serializer.writableHighWaterMark, 16);
+assert.strictEqual(parser.readableHighWaterMark,
+ parser._readableState.highWaterMark);
+assert.strictEqual(parser.writableHighWaterMark,
+ parser._writableState.highWaterMark);
 
 serializer._transform = function(obj, _, callback) {
  callback(null, Buffer.from([obj.val]));