stream: split objectMode for Duplex

This commit introduces `readableObjectMode` and
`writableObjectMode` options for Duplex streams.
This can be used mostly to make parsers and
serializers with Transform streams.

Also the docs section about stream state objects
is removed, because it is not relevant anymore.
The example from the section is remade to show
new options.

fixes #6284

Signed-off-by: Timothy J Fontaine <tjfontaine@gmail.com>

doc/api/stream.markdown

@@ -1082,6 +1082,12 @@ Writable.  It is thus up to the user to implement both the lowlevel
   * `allowHalfOpen` {Boolean} Default=true.  If set to `false`, then
     the stream will automatically end the readable side when the
     writable side ends and vice versa.
+  * `readableObjectMode` {Boolean} Default=false. Sets `objectMode`
+    for readable side of the stream. Has no effect if `objectMode`
+    is `true`.
+  * `writableObjectMode` {Boolean} Default=false. Sets `objectMode`
+    for writable side of the stream. Has no effect if `objectMode`
+    is `true`.
 
 In classes that extend the Duplex class, make sure to call the
 constructor so that the buffering settings can be properly
@@ -1420,17 +1426,10 @@ used by userland streaming libraries.
 You should set `objectMode` in your stream child class constructor on
 the options object.  Setting `objectMode` mid-stream is not safe.
 
-### State Objects
-
-[Readable][] streams have a member object called `_readableState`.
-[Writable][] streams have a member object called `_writableState`.
-[Duplex][] streams have both.
-
-**These objects should generally not be modified in child classes.**
-However, if you have a Duplex or Transform stream that should be in
-`objectMode` on the readable side, and not in `objectMode` on the
-writable side, then you may do this in the constructor by setting the
-flag explicitly on the appropriate state object.
+For Duplex streams `objectMode` can be set exclusively for readable or
+writable side with `readableObjectMode` and `writableObjectMode`
+respectivly. These options can be used to implement parsers and
+serializers with Transform streams.
 
 ```javascript
 var util = require('util');
@@ -1439,13 +1438,12 @@ var Transform = require('stream').Transform;
 util.inherits(JSONParseStream, Transform);
 
 // Gets \n-delimited JSON string data, and emits the parsed objects
-function JSONParseStream(options) {
+function JSONParseStream() {
   if (!(this instanceof JSONParseStream))
-    return new JSONParseStream(options);
+    return new JSONParseStream();
+
+  Transform.call(this, { readableObjectMode : true });
 
-  Transform.call(this, options);
-  this._writableState.objectMode = false;
-  this._readableState.objectMode = true;
   this._buffer = '';
   this._decoder = new StringDecoder('utf8');
 }
@@ -1487,11 +1485,6 @@ JSONParseStream.prototype._flush = function(cb) {
 };
 ```
 
-The state objects contain other useful information for debugging the
-state of streams in your programs.  It is safe to look at them, but
-beyond setting option flags in the constructor, it is **not** safe to
-modify them.
-
 
 [EventEmitter]: events.html#events_class_events_eventemitter
 [Object mode]: #stream_object_mode

lib/_stream_readable.js

@@ -68,6 +68,9 @@ function ReadableState(options, stream) {
   // 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

@@ -51,6 +51,9 @@ function WritableState(options, stream) {
   // contains buffers or objects.
   this.objectMode = !!options.objectMode;
 
+  if (stream instanceof Stream.Duplex)
+    this.objectMode = this.objectMode || !!options.writableObjectMode;
+
   // cast to ints.
   this.highWaterMark = ~~this.highWaterMark;
 

test/simple/test-stream-duplex.js

@@ -0,0 +1,52 @@
+// Copyright Joyent, Inc. and other Node contributors.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the
+// "Software"), to deal in the Software without restriction, including
+// without limitation the rights to use, copy, modify, merge, publish,
+// distribute, sublicense, and/or sell copies of the Software, and to permit
+// persons to whom the Software is furnished to do so, subject to the
+// following conditions:
+//
+// The above copyright notice and this permission notice shall be included
+// in all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
+// USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+var common = require('../common');
+var assert = require('assert');
+
+var Duplex = require('stream').Transform;
+
+var stream = new Duplex({ objectMode: true });
+
+assert(stream._readableState.objectMode);
+assert(stream._writableState.objectMode);
+
+var written;
+var read;
+
+stream._write = function (obj, _, cb) {
+  written = obj;
+  cb();
+};
+
+stream._read = function () {};
+
+stream.on('data', function (obj) {
+  read = obj;
+});
+
+stream.push({ val: 1 });
+stream.end({ val: 2 });
+
+process.on('exit', function () {
+  assert(read.val === 1);
+  assert(written.val === 2);
+});

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

@@ -0,0 +1,68 @@
+// Copyright Joyent, Inc. and other Node contributors.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the
+// "Software"), to deal in the Software without restriction, including
+// without limitation the rights to use, copy, modify, merge, publish,
+// distribute, sublicense, and/or sell copies of the Software, and to permit
+// persons to whom the Software is furnished to do so, subject to the
+// following conditions:
+//
+// The above copyright notice and this permission notice shall be included
+// in all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
+// USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+var common = require('../common');
+var assert = require('assert');
+
+var Transform = require('stream').Transform;
+
+var parser = new Transform({ readableObjectMode : true });
+
+assert(parser._readableState.objectMode);
+assert(!parser._writableState.objectMode);
+
+parser._transform = function (chunk, enc, callback) {
+  callback(null, { val : chunk[0] });
+};
+
+var parsed;
+
+parser.on('data', function (obj) {
+  parsed = obj;
+});
+
+parser.end(new Buffer([42]));
+
+process.on('exit', function () {
+  assert(parsed.val === 42);
+});
+
+
+var serializer = new Transform({ writableObjectMode : true });
+
+assert(!serializer._readableState.objectMode);
+assert(serializer._writableState.objectMode);
+
+serializer._transform = function (obj, _, callback) {
+  callback(null, new Buffer([obj.val]));
+}
+
+var serialized;
+
+serializer.on('data', function (chunk) {
+  serialized = chunk;
+});
+
+serializer.write({ val : 42 });
+
+process.on('exit', function () {
+  assert(serialized[0] === 42);
+});