@@ -32,6 +32,8 @@ function isWritableFinished(stream) {
3232 return wState . finished || ( wState . ended && wState . length === 0 ) ;
3333}
3434
35+ function nop ( ) { }
36+
3537function eos ( stream , opts , callback ) {
3638 if ( arguments . length === 2 ) {
3739 callback = opts ;
@@ -52,20 +54,23 @@ function eos(stream, opts, callback) {
5254 let writable = opts . writable ||
5355 ( opts . writable !== false && isWritable ( stream ) ) ;
5456
57+ const wState = stream . _writableState ;
58+ const rState = stream . _readableState ;
59+
5560 const onlegacyfinish = ( ) => {
5661 if ( ! stream . writable ) onfinish ( ) ;
5762 } ;
5863
5964 let writableFinished = stream . writableFinished ||
60- ( stream . _writableState && stream . _writableState . finished ) ;
65+ ( rState && rState . finished ) ;
6166 const onfinish = ( ) => {
6267 writable = false ;
6368 writableFinished = true ;
6469 if ( ! readable ) callback . call ( stream ) ;
6570 } ;
6671
6772 let readableEnded = stream . readableEnded ||
68- ( stream . _readableState && stream . _readableState . endEmitted ) ;
73+ ( rState && rState . endEmitted ) ;
6974 const onend = ( ) => {
7075 readable = false ;
7176 readableEnded = true ;
@@ -79,7 +84,7 @@ function eos(stream, opts, callback) {
7984 const onclose = ( ) => {
8085 let err ;
8186 if ( readable && ! readableEnded ) {
82- if ( ! stream . _readableState || ! stream . _readableState . ended )
87+ if ( ! rState || ! rState . ended )
8388 err = new ERR_STREAM_PREMATURE_CLOSE ( ) ;
8489 return callback . call ( stream , err ) ;
8590 }
@@ -99,7 +104,7 @@ function eos(stream, opts, callback) {
99104 stream . on ( 'abort' , onclose ) ;
100105 if ( stream . req ) onrequest ( ) ;
101106 else stream . on ( 'request' , onrequest ) ;
102- } else if ( writable && ! stream . _writableState ) { // legacy streams
107+ } else if ( writable && ! wState ) { // legacy streams
103108 stream . on ( 'end' , onlegacyfinish ) ;
104109 stream . on ( 'close' , onlegacyfinish ) ;
105110 }
@@ -114,7 +119,24 @@ function eos(stream, opts, callback) {
114119 if ( opts . error !== false ) stream . on ( 'error' , onerror ) ;
115120 stream . on ( 'close' , onclose ) ;
116121
122+ const closed = ( wState && wState . closed ) || ( rState && rState . closed ) ||
123+ ( wState && wState . errorEmitted ) || ( rState && rState . errorEmitted ) ||
124+ ( wState && wState . finished ) || ( rState && rState . endEmitted ) ||
125+ ( rState && stream . req && stream . aborted ) ;
126+
127+ if ( closed ) {
128+ // TODO(ronag): Re-throw error if errorEmitted?
129+ // TODO(ronag): Throw premature close as if finished was called?
130+ // before being closed? i.e. if closed but not errored, ended or finished.
131+ // TODO(ronag): Throw some kind of error? Does it make sense
132+ // to call finished() on a "finished" stream?
133+ process . nextTick ( ( ) => {
134+ callback ( ) ;
135+ } ) ;
136+ }
137+
117138 return function ( ) {
139+ callback = nop ;
118140 stream . removeListener ( 'aborted' , onclose ) ;
119141 stream . removeListener ( 'complete' , onfinish ) ;
120142 stream . removeListener ( 'abort' , onclose ) ;
0 commit comments