stream: use readableObjectMode public api for js stream

PR-URL: https://github.com/nodejs/node/pull/27655
Refs: https://github.com/nodejs/node/issues/445
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Yongsheng Zhang <zyszys98@gmail.com>

doc/api/stream.md

@@ -502,6 +502,13 @@ This property contains the number of bytes (or objects) in the queue
 ready to be written. The value provides introspection data regarding
 the status of the `highWaterMark`.
 
+##### writable.writableObjectMode
+<!-- YAML
+added: REPLACEME
+-->
+
+Getter for the property `objectMode` of a given `Writable` stream.
+
 ##### writable.write(chunk[, encoding][, callback])
 <!-- YAML
 added: v0.9.4
@@ -1089,6 +1096,13 @@ This property contains the number of bytes (or objects) in the queue
 ready to be read. The value provides introspection data regarding
 the status of the `highWaterMark`.
 
+##### readable.readableObjectMode
+<!-- YAML
+added: REPLACEME
+-->
+
+Getter for the property `objectMode` of a given `Readable` stream.
+
 ##### readable.resume()
 <!-- YAML
 added: v0.9.4

lib/_stream_readable.js

@@ -1081,6 +1081,13 @@ Object.defineProperty(Readable.prototype, 'readableLength', {
   }
 });
 
+Object.defineProperty(Readable.prototype, 'readableObjectMode', {
+  enumerable: false,
+  get() {
+    return this._readableState ? this._readableState.objectMode : false;
+  }
+});
+
 // Pluck off n bytes from an array of buffers.
 // Length is the combined lengths of all the buffers in the list.
 // This function is designed to be inlinable, so please take care when making

lib/_stream_writable.js

@@ -707,6 +707,13 @@ Object.defineProperty(Writable.prototype, 'destroyed', {
   }
 });
 
+Object.defineProperty(Writable.prototype, 'writableObjectMode', {
+  enumerable: false,
+  get() {
+    return this._writableState ? this._writableState.objectMode : false;
+  }
+});
+
 Writable.prototype.destroy = destroyImpl.destroy;
 Writable.prototype._undestroy = destroyImpl.undestroy;
 Writable.prototype._destroy = function(err, cb) {

lib/internal/js_stream_socket.js

@@ -50,7 +50,7 @@ class JSStreamSocket extends Socket {
     stream.on('error', (err) => this.emit('error', err));
     const ondata = (chunk) => {
       if (typeof chunk === 'string' ||
-          stream._readableState.objectMode === true) {
+          stream.readableObjectMode === true) {
         // Make sure that no further `data` events will happen.
         stream.pause();
         stream.removeListener('data', ondata);

test/parallel/test-stream2-basic.js

@@ -23,6 +23,7 @@
 
 const common = require('../common');
 const R = require('_stream_readable');
+const W = require('_stream_writable');
 const assert = require('assert');
 
 const EE = require('events').EventEmitter;
@@ -420,3 +421,15 @@ class TestWriter extends EE {
   const r2 = r.setEncoding('utf8').pause().resume().pause();
   assert.strictEqual(r, r2);
 }
+
+{
+  // Verify readableObjectMode property
+  const r = new R({ objectMode: true });
+  assert.strictEqual(r.readableObjectMode, true);
+}
+
+{
+  // Verify writableObjectMode property
+  const w = new W({ objectMode: true });
+  assert.strictEqual(w.writableObjectMode, true);
+}