stream: make readable & writable computed

This makes readable and writable automatically computed based
on the stream state.

Effectivly deprecating/discouraging manual management of this.

Makes the properties  more consistent and easier to reason about.

Fixes: nodejs#29377

doc/api/stream.md

@@ -495,7 +495,8 @@ added: v11.4.0
 
 * {boolean}
 
-Is `true` if it is safe to call [`writable.write()`][stream-write].
+Is `true` if it is safe to call [`writable.write()`][stream-write]. Which means
+the stream has not been destroyed, errored or ended.
 
 ##### `writable.writableEnded`
 <!-- YAML
@@ -1134,7 +1135,8 @@ added: v11.4.0
 
 * {boolean}
 
-Is `true` if it is safe to call [`readable.read()`][stream-read].
+Is `true` if it is safe to call [`readable.read()`][stream-read]. Which means
+the stream has not been destroyed or emitted `'error'` or `'end'`.
 
 ##### `readable.readableEncoding`
 <!-- YAML

lib/_http_incoming.js

@@ -63,8 +63,6 @@ function IncomingMessage(socket) {
   this.trailers = {};
   this.rawTrailers = [];
 
-  this.readable = true;
-
   this.aborted = false;
 
   this.upgrade = null;

lib/_stream_duplex.js

@@ -28,6 +28,7 @@
 
 const {
   ObjectDefineProperties,
+  ObjectGetOwnPropertyDescriptor,
   ObjectKeys,
   ObjectSetPrototypeOf,
 } = primordials;
@@ -71,6 +72,7 @@ function Duplex(options) {
 }
 
 ObjectDefineProperties(Duplex.prototype, {
+  writable: ObjectGetOwnPropertyDescriptor(Writable.prototype, 'writable'),
 
   destroyed: {
     get() {

lib/_stream_readable.js

@@ -23,6 +23,7 @@
 
 const {
   ArrayIsArray,
+  Boolean,
   NumberIsInteger,
   NumberIsNaN,
   ObjectDefineProperties,
@@ -181,9 +182,6 @@ function Readable(options) {
 
   this._readableState = new ReadableState(options, this, isDuplex);
 
-  // legacy
-  this.readable = true;
-
   if (options) {
     if (typeof options.read === 'function')
       this._read = options.read;
@@ -1057,6 +1055,20 @@ Readable.prototype[SymbolAsyncIterator] = function() {
 // because otherwise some prototype manipulation in
 // userland will fail
 ObjectDefineProperties(Readable.prototype, {
+  readable: {
+    get() {
+      const r = this._readableState;
+      if (!r) return false;
+      if (r.readable !== undefined) return r.readable && !r.endEmitted;
+      return Boolean(!r.destroyed && !r.errorEmitted && !r.endEmitted);
+    },
+    set(val) {
+      // Backwards compat.
+      if (this._readableState) {
+        this._readableState.readable = !!val;
+      }
+    }
+  },
 
   readableHighWaterMark: {
     enumerable: false,
@@ -1198,7 +1210,6 @@ function endReadableNT(state, stream) {
   // Check that we didn't get one last unshift.
   if (!state.errorEmitted && !state.endEmitted && state.length === 0) {
     state.endEmitted = true;
-    stream.readable = false;
     stream.emit('end');
 
     if (state.autoDestroy) {

lib/_stream_writable.js

@@ -27,6 +27,7 @@
 
 const {
   Array,
+  Boolean,
   FunctionPrototype,
   ObjectDefineProperty,
   ObjectSetPrototypeOf,
@@ -241,9 +242,6 @@ function Writable(options) {
 
   this._writableState = new WritableState(options, this, isDuplex);
 
-  // legacy.
-  this.writable = true;
-
   if (options) {
     if (typeof options.write === 'function')
       this._write = options.write;
@@ -784,6 +782,21 @@ function onFinished(stream, state, cb) {
   stream.prependListener('error', onerror);
 }
 
+ObjectDefineProperty(Writable.prototype, 'writable', {
+  get() {
+    const w = this._writableState;
+    if (!w) return false;
+    if (w.writable !== undefined) return w.writable;
+    return Boolean(!w.destroyed && !w.errored && !w.ending);
+  },
+  set(val) {
+    // Backwards compat.
+    if (this._writableState) {
+      this._writableState.writable = !!val;
+    }
+  }
+});
+
 ObjectDefineProperty(Writable.prototype, 'destroyed', {
   // Making it explicit this property is not enumerable
   // because otherwise some prototype manipulation in

test/parallel/test-http2-socket-proxy.js

@@ -85,10 +85,14 @@ server.on('stream', common.mustCall(function(stream, headers) {
 
   stream.respond();
 
-  socket.writable = 0;
-  socket.readable = 0;
-  assert.strictEqual(socket.writable, 0);
-  assert.strictEqual(socket.readable, 0);
+  socket.writable = true;
+  socket.readable = true;
+  assert.strictEqual(socket.writable, true);
+  assert.strictEqual(socket.readable, true);
+  socket.writable = false;
+  socket.readable = false;
+  assert.strictEqual(socket.writable, false);
+  assert.strictEqual(socket.readable, false);
 
   stream.end();
 

test/parallel/test-stream-readable-readable.js

@@ -0,0 +1,46 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+
+const { Readable } = require('stream');
+
+{
+  const r = new Readable({
+    read() {}
+  });
+  assert.strictEqual(r.readable, true);
+  r.destroy();
+  assert.strictEqual(r.readable, false);
+}
+
+{
+  const mustNotCall = common.mustNotCall();
+  const r = new Readable({
+    read() {}
+  });
+  assert.strictEqual(r.readable, true);
+  r.on('end', mustNotCall);
+  r.resume();
+  r.push(null);
+  assert.strictEqual(r.readable, true);
+  r.off('end', mustNotCall);
+  r.on('end', common.mustCall(() => {
+    assert.strictEqual(r.readable, false);
+  }));
+}
+
+{
+  const r = new Readable({
+    read: common.mustCall(() => {
+      process.nextTick(() => {
+        r.destroy(new Error());
+        assert.strictEqual(r.readable, false);
+      });
+    })
+  });
+  r.resume();
+  r.on('error', common.mustCall());
+  r.on('error', common.mustCall(() => {
+    assert.strictEqual(r.readable, false);
+  }));
+}

test/parallel/test-stream-writable-writable.js

@@ -0,0 +1,49 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+
+const { Writable } = require('stream');
+
+{
+  const w = new Writable({
+    write() {}
+  });
+  assert.strictEqual(w.writable, true);
+  w.destroy();
+  assert.strictEqual(w.writable, false);
+}
+
+{
+  const w = new Writable({
+    write: common.mustCall((chunk, encoding, callback) => {
+      callback(new Error());
+    })
+  });
+  assert.strictEqual(w.writable, true);
+  w.write('asd');
+  assert.strictEqual(w.writable, false);
+  w.on('error', common.mustCall());
+  w.destroy();
+}
+
+{
+  const w = new Writable({
+    write: common.mustCall((chunk, encoding, callback) => {
+      process.nextTick(() => {
+        callback(new Error());
+        assert.strictEqual(w.writable, false);
+      });
+    })
+  });
+  w.write('asd');
+  w.on('error', common.mustCall());
+}
+
+{
+  const w = new Writable({
+    write: common.mustNotCall()
+  });
+  assert.strictEqual(w.writable, true);
+  w.end();
+  assert.strictEqual(w.writable, false);
+}