stream: add errored and closed props

PR-URL: #40696 Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Benjamin Gruenbaum <[email protected]> Reviewed-By: James M Snell <[email protected]>

Author: ronag

doc/api/stream.md

@@ -456,6 +456,16 @@ further errors except from `_destroy()` may be emitted as `'error'`.
 Implementors should not override this method,
 but instead implement [`writable._destroy()`][writable-_destroy].
 
+##### `writable.closed`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+*{boolean}
+
+Is `true` after `'close'` has been emitted.
+
 ##### `writable.destroyed`
 
 <!-- YAML
@@ -611,6 +621,17 @@ added:
 Number of times [`writable.uncork()`][stream-uncork] needs to be
 called in order to fully uncork the stream.
 
+##### `writable.writableErrored`
+
+<!-- YAML
+added:
+ REPLACEME
+-->
+
+*{Error}
+
+Returns error if the stream has been destroyed with an error.
+
 ##### `writable.writableFinished`
 
 <!-- YAML
@@ -1080,14 +1101,24 @@ further errors except from `_destroy()` may be emitted as `'error'`.
 Implementors should not override this method, but instead implement
 [`readable._destroy()`][readable-_destroy].
 
-##### `readable.destroyed`
+##### `readable.closed`
 
 <!-- YAML
 added: v8.0.0
 -->
 
 *{boolean}
 
+Is `true` after `'close'` has been emitted.
+
+##### `readable.destroyed`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+*{boolean}
+
 Is `true` after [`readable.destroy()`][readable-destroy] has been called.
 
 ##### `readable.isPaused()`
@@ -1346,6 +1377,17 @@ added: v12.9.0
 
 Becomes `true` when [`'end'`][] event is emitted.
 
+##### `readable.readableErrored`
+
+<!-- YAML
+added:
+ REPLACEME
+-->
+
+*{Error}
+
+Returns error if the stream has been destroyed with an error.
+
 ##### `readable.readableFlowing`
 
 <!-- YAML

lib/internal/fs/streams.js

@@ -107,12 +107,9 @@ const FileHandleOperations = (handle) =>{
 
 functionclose(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;
@@ -186,7 +183,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){
@@ -358,10 +354,8 @@ function WriteStream(path, options){
 this.start=options.start;
 this.pos=undefined;
 this.bytesWritten=0;
-this.closed=false;
 this[kIsPerformingIO]=false;
 
-
 if(this.start!==undefined){
 validateInteger(this.start,'start',0);
 

lib/internal/streams/end-of-stream.js

@@ -22,9 +22,11 @@ const{
  isReadable,
  isReadableNodeStream,
  isReadableFinished,
+ isReadableErrored,
  isWritable,
  isWritableNodeStream,
  isWritableFinished,
+ isWritableErrored,
  isNodeStream,
 willEmitClose: _willEmitClose,
 }=require('internal/streams/utils');
@@ -110,7 +112,7 @@ function eos(stream, options, callback){
 constonclose=()=>{
 closed=true;
 
-consterrored=wState?.errored||rState?.errored;
+consterrored=isWritableErrored(stream)||isReadableErrored(stream);
 
 if(errored&&typeoferrored!=='boolean'){
 returncallback.call(stream,errored);

lib/internal/streams/readable.js

@@ -1239,13 +1239,23 @@ ObjectDefineProperties(Readable.prototype,{
 }
 },
 
+readableErrored: {
+enumerable: false,
+get(){
+returnthis._readableState ? this._readableState.errored : null;
+}
+},
+
+closed: {
+get(){
+returnthis._readableState ? this._readableState.closed : false;
+}
+},
+
 destroyed: {
 enumerable: false,
 get(){
-if(this._readableState===undefined){
-returnfalse;
-}
-returnthis._readableState.destroyed;
+returnthis._readableState ? this._readableState.destroyed : false;
 },
 set(value){
 // We ignore the value if the stream

lib/internal/streams/utils.js

@@ -144,11 +144,39 @@ function isFinished(stream, opts){
 returntrue;
 }
 
+functionisWritableErrored(stream){
+if(!isNodeStream(stream)){
+returnnull;
+}
+
+if(stream.writableErrored){
+returnstream.writableErrored;
+}
+
+returnstream._writableState?.errored??null;
+}
+
+functionisReadableErrored(stream){
+if(!isNodeStream(stream)){
+returnnull;
+}
+
+if(stream.readableErrored){
+returnstream.readableErrored;
+}
+
+returnstream._readableState?.errored??null;
+}
+
 functionisClosed(stream){
 if(!isNodeStream(stream)){
 returnnull;
 }
 
+if(typeofstream.closed==='boolean'){
+returnstream.closed;
+}
+
 constwState=stream._writableState;
 constrState=stream._readableState;
 
@@ -226,11 +254,13 @@ module.exports ={
  isReadableNodeStream,
  isReadableEnded,
  isReadableFinished,
+ isReadableErrored,
  isNodeStream,
  isWritable,
  isWritableNodeStream,
  isWritableEnded,
  isWritableFinished,
+ isWritableErrored,
  isServerRequest,
  isServerResponse,
  willEmitClose,

lib/internal/streams/writable.js

@@ -768,6 +768,12 @@ function finish(stream, state){
 
 ObjectDefineProperties(Writable.prototype,{
 
+closed: {
+get(){
+returnthis._writableState ? this._writableState.closed : false;
+}
+},
+
 destroyed: {
 get(){
 returnthis._writableState ? this._writableState.destroyed : false;
@@ -846,7 +852,14 @@ ObjectDefineProperties(Writable.prototype,{
 get(){
 returnthis._writableState&&this._writableState.length;
 }
-}
+},
+
+writableErrored: {
+enumerable: false,
+get(){
+returnthis._writableState ? this._writableState.errored : null;
+}
+},
 });
 
 constdestroy=destroyImpl.destroy;

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-finished.js

@@ -612,8 +612,10 @@ testClosed((opts) => new Writable({write(){}, ...opts }));
 constw=newWritable();
 const_err=newError();
 w.destroy(_err);
+assert.strictEqual(w.writableErrored,_err);
 finished(w,common.mustCall((err)=>{
 assert.strictEqual(_err,err);
+assert.strictEqual(w.closed,true);
 finished(w,common.mustCall((err)=>{
 assert.strictEqual(_err,err);
 }));
@@ -623,7 +625,9 @@ testClosed((opts) => new Writable({write(){}, ...opts }));
 {
 constw=newWritable();
 w.destroy();
+assert.strictEqual(w.writableErrored,null);
 finished(w,common.mustCall((err)=>{
+assert.strictEqual(w.closed,true);
 assert.strictEqual(err.code,'ERR_STREAM_PREMATURE_CLOSE');
 finished(w,common.mustCall((err)=>{
 assert.strictEqual(err.code,'ERR_STREAM_PREMATURE_CLOSE');

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

@@ -13,6 +13,7 @@ const assert = require('assert');
 read.on('close',common.mustCall());
 
 read.destroy();
+assert.strictEqual(read.readableErrored,null);
 assert.strictEqual(read.destroyed,true);
 }
 
@@ -31,6 +32,7 @@ const assert = require('assert');
 }));
 
 read.destroy(expected);
+assert.strictEqual(read.readableErrored,expected);
 assert.strictEqual(read.destroyed,true);
 }