streams2: Add high water mark for Readable

Also, organize the numeric settings a bit on the ReadableState class

lib/_stream_readable.js

@@ -32,13 +32,34 @@ util.inherits(Readable, Stream);
 function ReadableState(options, stream) {
   options = options || {};
 
-  this.bufferSize = options.bufferSize || 16 * 1024;
-  assert(typeof this.bufferSize === 'number');
   // cast to an int
   this.bufferSize = ~~this.bufferSize;
 
+  // the argument passed to this._read(n,cb)
+  this.bufferSize = options.hasOwnProperty('bufferSize') ?
+      options.bufferSize : 16 * 1024;
+
+  // the point at which it stops calling _read() to fill the buffer
+  this.highWaterMark = options.hasOwnProperty('highWaterMark') ?
+      options.highWaterMark : 16 * 1024;
+
+  // the minimum number of bytes to buffer before emitting 'readable'
+  // default to pushing everything out as fast as possible.
   this.lowWaterMark = options.hasOwnProperty('lowWaterMark') ?
       options.lowWaterMark : 1024;
+
+  // cast to ints.
+  assert(typeof this.bufferSize === 'number');
+  assert(typeof this.lowWaterMark === 'number');
+  assert(typeof this.highWaterMark === 'number');
+  this.bufferSize = ~~this.bufferSize;
+  this.lowWaterMark = ~~this.lowWaterMark;
+  this.highWaterMark = ~~this.highWaterMark;
+  assert(this.bufferSize >= 0);
+  assert(this.lowWaterMark >= 0);
+  assert(this.highWaterMark >= this.lowWaterMark,
+         this.highWaterMark + '>=' + this.lowWaterMark);
+
   this.buffer = [];
   this.length = 0;
   this.pipes = [];
@@ -136,9 +157,16 @@ Readable.prototype.read = function(n) {
 
   // if we need a readable event, then we need to do some reading.
   var doRead = state.needReadable;
-  // if we currently have less than the lowWaterMark, then also read some
-  if (state.length - n <= state.lowWaterMark)
+
+  // if we currently have less than the highWaterMark, then also read some
+  if (state.length - n <= state.highWaterMark)
     doRead = true;
+
+  // if we currently have *nothing*, then always try to get *something*
+  // no matter what the high water mark says.
+  if (state.length === 0)
+    doRead = true;
+
   // however, if we've ended, then there's no point, and if we're already
   // reading, then it's unnecessary.
   if (state.ended || state.reading)

lib/_stream_transform.js

@@ -130,7 +130,7 @@ Transform.prototype._write = function(chunk, cb) {
 
   // if we weren't waiting for it, but nothing is queued up, then
   // still kick off a transform, just so it's there when the user asks.
-  var doRead = rs.needReadable || rs.length <= rs.lowWaterMark;
+  var doRead = rs.needReadable || rs.length <= rs.highWaterMark;
   if (doRead && !rs.reading) {
     var ret = this.read(0);
     if (ret !== null)