stream: add closed property

fs streams already have a closed property. This PR standardized
it and brings it inline with rest of streams.

doc/api/stream.md

@@ -406,6 +406,16 @@ added: v8.0.0
 
 Is `true` after [`writable.destroy()`][writable-destroy] has been called.
 
+##### `writable.closed`
+<!-- YAML
+added: REPLACEME
+-->
+
+* {boolean}
+
+Is `true` after `callback` in [`writable._destroy(err, callback)`][writable-_destroy]
+has been called.
+
 ##### `writable.end([chunk[, encoding]][, callback])`
 <!-- YAML
 added: v0.9.4
@@ -983,6 +993,16 @@ added: v8.0.0
 
 Is `true` after [`readable.destroy()`][readable-destroy] has been called.
 
+##### `readable.closed`
+<!-- YAML
+added: REPLACEME
+-->
+
+* {boolean}
+
+Is `true` after `callback` in [`readable._destroy(err, callback)`][readable-_destroy]
+has been called.
+
 ##### `readable.isPaused()`
 <!-- YAML
 added: v0.11.14

lib/_stream_duplex.js

@@ -104,5 +104,23 @@ ObjectDefineProperties(Duplex.prototype, {
         this._writableState.destroyed = value;
       }
     }
+  },
+
+  closed: {
+    get() {
+      if (this._readableState === undefined ||
+        this._writableState === undefined) {
+        return false;
+      }
+      return this._readableState.closed && this._writableState.closed;
+    },
+    set(value) {
+      // Backward compatibility, the user is explicitly
+      // managing closed.
+      if (this._readableState && this._writableState) {
+        this._readableState.closed = value;
+        this._writableState.closed = value;
+      }
+    }
   }
 });

lib/_stream_readable.js

@@ -1167,6 +1167,18 @@ ObjectDefineProperties(Readable.prototype, {
     }
   },
 
+  closed: {
+    get() {
+      return this._readableState ? this._readableState.closed : false;
+    },
+    set(value) {
+      // Backward compatibility, the user is explicitly managing closed.
+      if (this._readableState) {
+        this._readableState.closed = value;
+      }
+    }
+  },
+
   readableEnded: {
     enumerable: false,
     get() {

lib/_stream_writable.js

@@ -735,6 +735,18 @@ ObjectDefineProperties(Writable.prototype, {
     }
   },
 
+  closed: {
+    get() {
+      return this._writableState ? this._writableState.closed : false;
+    },
+    set(value) {
+      // Backward compatibility, the user is explicitly managing closed.
+      if (this._writableState) {
+        this._writableState.closed = value;
+      }
+    }
+  },
+
   writable: {
     get() {
       const w = this._writableState;

lib/internal/fs/streams.js

@@ -88,12 +88,9 @@ function _construct(callback) {
 
 function close(stream, err, cb) {
   if (!stream.fd) {
-    // TODO(ronag)
-    // stream.closed = true;
     cb(err);
   } else {
     stream[kFs].close(stream.fd, (er) => {
-      stream.closed = true;
       cb(er || err);
     });
     stream.fd = null;
@@ -143,7 +140,6 @@ function ReadStream(path, options) {
   this.end = options.end;
   this.pos = undefined;
   this.bytesRead = 0;
-  this.closed = false;
   this[kIsPerformingIO] = false;
 
   if (this.start !== undefined) {
@@ -327,7 +323,6 @@ function WriteStream(path, options) {
   this.start = options.start;
   this.pos = undefined;
   this.bytesWritten = 0;
-  this.closed = false;
   this[kIsPerformingIO] = false;
 
   if (this.start !== undefined) {

test/parallel/test-fs-read-stream-inherit.js

@@ -199,7 +199,7 @@ const rangeFile = fixtures.path('x.txt');
   file.on('error', common.mustCall());
 
   process.on('exit', function() {
-    assert(!file.closed);
+    assert(file.closed);
     assert(file.destroyed);
   });
 }

test/parallel/test-fs-read-stream.js

@@ -271,7 +271,7 @@ if (!common.isWindows) {
   file.on('error', common.mustCall());
 
   process.on('exit', function() {
-    assert(!file.closed);
+    assert(file.closed);
     assert(file.destroyed);
   });
 }

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

@@ -6,14 +6,21 @@ const assert = require('assert');
 
 {
   const read = new Readable({
-    read() {}
+    read() {},
+    destroy(err, cb) {
+      assert.strictEqual(read.closed, false);
+      cb();
+      assert.strictEqual(read.closed, true);
+    }
   });
   read.resume();
 
   read.on('close', common.mustCall());
 
+  assert.strictEqual(read.closed, false);
   read.destroy();
   assert.strictEqual(read.destroyed, true);
+  assert.strictEqual(read.closed, true);
 }
 
 {
@@ -101,7 +108,9 @@ const assert = require('assert');
     assert.strictEqual(err, null);
     process.nextTick(() => {
       this.push(null);
+      assert.strictEqual(read.closed, false);
       cb();
+      assert.strictEqual(read.closed, true);
     });
   });
 
@@ -112,9 +121,11 @@ const assert = require('assert');
 
   read.destroy();
 
+  assert.strictEqual(read.closed, false);
   read.removeListener('end', fail);
   read.on('end', common.mustNotCall());
   assert.strictEqual(read.destroyed, true);
+  assert.strictEqual(read.closed, false);
 }
 
 {

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

@@ -6,14 +6,21 @@ const assert = require('assert');
 
 {
   const write = new Writable({
-    write(chunk, enc, cb) { cb(); }
+    write(chunk, enc, cb) { cb(); },
+    destroy(err, cb) {
+      assert.strictEqual(write.closed, false);
+      cb();
+      assert.strictEqual(write.closed, true);
+    }
   });
 
   write.on('finish', common.mustNotCall());
   write.on('close', common.mustCall());
 
+  assert.strictEqual(write.closed, false);
   write.destroy();
   assert.strictEqual(write.destroyed, true);
+  assert.strictEqual(write.closed, true);
 }
 
 {
@@ -113,7 +120,9 @@ const assert = require('assert');
     assert.strictEqual(err, null);
     process.nextTick(() => {
       this.end();
+      assert.strictEqual(write.closed, false);
       cb();
+      assert.strictEqual(write.closed, true);
     });
   });
 
@@ -127,6 +136,7 @@ const assert = require('assert');
   write.removeListener('finish', fail);
   write.on('finish', common.mustCall());
   assert.strictEqual(write.destroyed, true);
+  assert.strictEqual(write.closed, false);
 }
 
 {