streams2: Allow 0 as a lowWaterMark value

lib/_stream_readable.js

@@ -36,7 +36,8 @@ function ReadableState(options, stream) {
   // cast to an int
   this.bufferSize = ~~this.bufferSize;
 
-  this.lowWaterMark = options.lowWaterMark || 1024;
+  this.lowWaterMark = options.hasOwnProperty('lowWaterMark') ?
+      options.lowWaterMark : 1024;
   this.buffer = [];
   this.length = 0;
   this.pipes = [];
@@ -94,7 +95,7 @@ Readable.prototype.read = function(n) {
   // but then it won't ever cause _read to be called, so in that case,
   // we just return what we have, and let the programmer deal with it.
   if (n > state.length) {
-    if (!state.ended && state.length < state.lowWaterMark) {
+    if (!state.ended && state.length <= state.lowWaterMark) {
       state.needReadable = true;
       n = 0;
     } else
@@ -114,7 +115,7 @@ Readable.prototype.read = function(n) {
   state.length -= n;
 
   if (!state.ended &&
-      state.length < state.lowWaterMark &&
+      state.length <= state.lowWaterMark &&
       !state.reading) {
     state.reading = true;
     // call internal read method
@@ -145,7 +146,7 @@ Readable.prototype.read = function(n) {
       // that it's time to read more data.  Otherwise, that'll
       // probably kick off another stream.read(), which can trigger
       // another _read(n,cb) before this one returns!
-      if (state.length < state.lowWaterMark) {
+      if (state.length <= state.lowWaterMark) {
         state.reading = true;
         this._read(state.bufferSize, onread.bind(this));
         return;
@@ -398,7 +399,7 @@ Readable.prototype.wrap = function(stream) {
     var ret = fromList(n, state.buffer, state.length, !!state.decoder);
     state.length -= n;
 
-    if (state.length < state.lowWaterMark && paused) {
+    if (state.length <= state.lowWaterMark && paused) {
       stream.resume();
       paused = false;
     }

lib/_stream_writable.js

@@ -33,7 +33,10 @@ util.inherits(Writable, Stream);
 function WritableState(options) {
   options = options || {};
   this.highWaterMark = options.highWaterMark || 16 * 1024;
-  this.lowWaterMark = options.lowWaterMark || 1024;
+  this.highWaterMark = options.hasOwnProperty('highWaterMark') ?
+      options.highWaterMark : 16 * 1024;
+  this.lowWaterMark = options.hasOwnProperty('lowWaterMark') ?
+      options.lowWaterMark : 1024;
   this.needDrain = false;
   this.ended = false;
   this.ending = false;
@@ -103,7 +106,7 @@ Writable.prototype.write = function(chunk, encoding) {
       this._write(chunk, writecb.bind(this));
     }
 
-    if (state.length < state.lowWaterMark && state.needDrain) {
+    if (state.length <= state.lowWaterMark && state.needDrain) {
       // Must force callback to be called on nextTick, so that we don't
       // emit 'drain' before the write() consumer gets the 'false' return
       // value, and has a chance to attach a 'drain' listener.