stream: stricter isReadableNodeStream

Fixes: #40938
PR-URL: #40941
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: ronag

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

@@ -116,7 +116,7 @@ function eos(stream, options, callback) {
 returncallback.call(stream,errored);
 }
 
-if(readable&&!readableFinished){
+if(readable&&!readableFinished&&isReadableNodeStream(stream,true)){
 if(!isReadableFinished(stream,false))
 returncallback.call(stream,
 newERR_STREAM_PREMATURE_CLOSE());

lib/internal/streams/utils.js

@@ -9,11 +9,15 @@ const {
 constkDestroyed=Symbol('kDestroyed');
 constkIsDisturbed=Symbol('kIsDisturbed');
 
-functionisReadableNodeStream(obj){
+functionisReadableNodeStream(obj,strict=false){
 return!!(
 obj&&
 typeofobj.pipe==='function'&&
 typeofobj.on==='function'&&
+(
+!strict||
+(typeofobj.pause==='function'&&typeofobj.resume==='function')
+)&&
 (!obj._writableState||obj._readableState?.readable!==false)&&// Duplex
 (!obj._writableState||obj._readableState)// Writable has .pipe.
 );

test/parallel/test-stream-finished.js

@@ -639,3 +639,20 @@ testClosed((opts) => new Writable({ write() {}, ...opts }));
 consts=newStream();
 finished(s,common.mustNotCall());
 }
+
+{
+constserver=http.createServer(common.mustCall(function(req,res){
+fs.createReadStream(__filename).pipe(res);
+finished(res,common.mustCall(function(err){
+assert.strictEqual(err,undefined);
+}));
+})).listen(0,function(){
+http.request(
+{method: 'GET',port: this.address().port},
+common.mustCall(function(res){
+res.resume();
+server.close();
+})
+).end();
+});
+}