streams: set default hwm properly for Duplex

Default highWaterMark is now set properly when using stream Duplex's
writableObjectMode and readableObjectMode options.

Added condition to the already existing split objectMode test to ensure
the highWaterMark is being set to the correct default value on both the
ReadableState and WritableState for readableObjectMode and
writableObjectMode.

Signed-off-by: Fedor Indutny <fedor@indutny.com>

lib/_stream_readable.js

@@ -33,10 +33,17 @@ util.inherits(Readable, Stream);
 function ReadableState(options, stream) {
   options = options || {};
 
+  // object stream flag. Used to make read(n) ignore n and to
+  // make all the buffer merging and length checks go away
+  this.objectMode = !!options.objectMode;
+
+  if (stream instanceof Stream.Duplex)
+    this.objectMode = this.objectMode || !!options.readableObjectMode;
+
   // 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 defaultHwm = options.objectMode ? 16 : 16 * 1024;
+  var defaultHwm = this.objectMode ? 16 : 16 * 1024;
   this.highWaterMark = (hwm || hwm === 0) ? hwm : defaultHwm;
 
   // cast to ints.
@@ -63,14 +70,6 @@ function ReadableState(options, stream) {
   this.emittedReadable = false;
   this.readableListening = false;
 
-
-  // object stream flag. Used to make read(n) ignore n and to
-  // make all the buffer merging and length checks go away
-  this.objectMode = !!options.objectMode;
-
-  if (stream instanceof Stream.Duplex)
-    this.objectMode = this.objectMode || !!options.readableObjectMode;
-
   // Crypto is kind of old and crusty.  Historically, its default string
   // encoding is 'binary' so we have to make this configurable.
   // Everything else in the universe uses 'utf8', though.

lib/_stream_writable.js

@@ -40,13 +40,6 @@ function WriteReq(chunk, encoding, cb) {
 function WritableState(options, stream) {
   options = options || {};
 
-  // 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 defaultHwm = options.objectMode ? 16 : 16 * 1024;
-  this.highWaterMark = (hwm || hwm === 0) ? hwm : defaultHwm;
-
   // object stream flag to indicate whether or not this stream
   // contains buffers or objects.
   this.objectMode = !!options.objectMode;
@@ -54,6 +47,13 @@ function WritableState(options, stream) {
   if (stream instanceof Stream.Duplex)
     this.objectMode = this.objectMode || !!options.writableObjectMode;
 
+  // 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 defaultHwm = this.objectMode ? 16 : 16 * 1024;
+  this.highWaterMark = (hwm || hwm === 0) ? hwm : defaultHwm;
+
   // cast to ints.
   this.highWaterMark = ~~this.highWaterMark;
 

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

@@ -28,6 +28,8 @@ var parser = new Transform({ readableObjectMode : true });
 
 assert(parser._readableState.objectMode);
 assert(!parser._writableState.objectMode);
+assert(parser._readableState.highWaterMark === 16);
+assert(parser._writableState.highWaterMark === (16 * 1024));
 
 parser._transform = function (chunk, enc, callback) {
   callback(null, { val : chunk[0] });
@@ -50,10 +52,12 @@ var serializer = new Transform({ writableObjectMode : true });
 
 assert(!serializer._readableState.objectMode);
 assert(serializer._writableState.objectMode);
+assert(serializer._readableState.highWaterMark === (16 * 1024));
+assert(serializer._writableState.highWaterMark === 16);
 
 serializer._transform = function (obj, _, callback) {
   callback(null, new Buffer([obj.val]));
-}
+};
 
 var serialized;