stream: don't destroy on async iterator success

Destroying on async iterator completion ignores autoDestroy. PR-URL: #35122 Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Anna Henningsen <[email protected]> Reviewed-By: Benjamin Gruenbaum <[email protected]> Reviewed-By: James M Snell <[email protected]>

Author: ronag

lib/internal/streams/readable.js

@@ -1066,7 +1066,7 @@ Readable.prototype[SymbolAsyncIterator] = function(){
 objectMode: true,
 destroy(err,callback){
 destroyImpl.destroyer(src,err);
-callback();
+callback(err);
 }
 }).wrap(src);
 }
@@ -1088,24 +1088,39 @@ async function* createAsyncIterator(stream){
 }
 }
 
+conststate=stream._readableState;
+
+leterror=state.errored;
+leterrorEmitted=state.errorEmitted;
+letendEmitted=state.endEmitted;
+letcloseEmitted=state.closeEmitted;
+
 stream
 .on('readable',next)
-.on('error',next)
-.on('end',next)
-.on('close',next);
+.on('error',function(err){
+error=err;
+errorEmitted=true;
+next.call(this);
+})
+.on('end',function(){
+endEmitted=true;
+next.call(this);
+})
+.on('close',function(){
+closeEmitted=true;
+next.call(this);
+});
 
 try{
-conststate=stream._readableState;
 while(true){
 constchunk=stream.read();
 if(chunk!==null){
 yieldchunk;
-}elseif(state.errored){
-throwstate.errored;
-}elseif(state.ended){
+}elseif(errorEmitted){
+throwerror;
+}elseif(endEmitted){
 break;
-}elseif(state.closed){
-// TODO(ronag): ERR_PREMATURE_CLOSE?
+}elseif(closeEmitted){
 break;
 }else{
 awaitnewPromise(next);
@@ -1115,7 +1130,10 @@ async function* createAsyncIterator(stream){
 destroyImpl.destroyer(stream,err);
 throwerr;
 }finally{
-destroyImpl.destroyer(stream,null);
+if(state.autoDestroy||!endEmitted){
+// TODO(ronag): ERR_PREMATURE_CLOSE?
+destroyImpl.destroyer(stream,null);
+}
 }
 }
 

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

@@ -9,6 +9,7 @@ const{
  pipeline
 }=require('stream');
 constassert=require('assert');
+consthttp=require('http');
 
 asyncfunctiontests(){
 {
@@ -44,9 +45,11 @@ async function tests(){
 constiter=Readable.prototype[Symbol.asyncIterator].call(stream);
 awaititer.next();
 awaititer.next();
-awaititer.next().catch(common.mustCall((err)=>{
-assert.strictEqual(err.message,'asd');
-}));
+awaititer.next()
+.then(common.mustNotCall())
+.catch(common.mustCall((err)=>{
+assert.strictEqual(err.message,'asd');
+}));
 }
 
 {
@@ -581,6 +584,61 @@ async function tests(){
 assert.strictEqual(err,_err);
 }));
 }
+
+{
+// Don't destroy if no auto destroy.
+// https://github.com/nodejs/node/issues/35116
+
+constr=newReadable({
+autoDestroy: false,
+read(){
+this.push('asd');
+this.push(null);
+}
+});
+
+forawait(constchunkofr){
+chunk;
+}
+assert.strictEqual(r.destroyed,false);
+}
+
+{
+// Destroy if no auto destroy and premature break.
+// https://github.com/nodejs/node/pull/35122/files#r485678318
+
+constr=newReadable({
+autoDestroy: false,
+read(){
+this.push('asd');
+}
+});
+
+forawait(constchunkofr){
+chunk;
+break;
+}
+assert.strictEqual(r.destroyed,true);
+}
+
+{
+// Don't destroy before 'end'.
+
+constr=newReadable({
+read(){
+this.push('asd');
+this.push(null);
+}
+}).on('end',()=>{
+assert.strictEqual(r.destroyed,false);
+});
+
+forawait(constchunkofr){
+chunk;
+}
+
+assert.strictEqual(r.destroyed,true);
+}
 }
 
 {
@@ -643,5 +701,78 @@ async function tests(){
 });
 }
 
+{
+let_req;
+constserver=http.createServer((request,response)=>{
+response.statusCode=404;
+response.write('never ends');
+});
+
+server.listen(()=>{
+_req=http.request(`http://localhost:${server.address().port}`)
+.on('response',common.mustCall(async(res)=>{
+setTimeout(()=>{
+_req.destroy(newError('something happened'));
+},100);
+
+res.on('error',common.mustCall());
+
+let_err;
+try{
+forawait(constchunkofres){
+chunk;
+}
+}catch(err){
+_err=err;
+}
+
+assert.strictEqual(_err.code,'ECONNRESET');
+server.close();
+}))
+.on('error',common.mustCall())
+.end();
+});
+}
+
+{
+asyncfunctiongetParsedBody(request){
+letbody='';
+
+forawait(constdataofrequest){
+body+=data;
+}
+
+try{
+returnJSON.parse(body);
+}catch{
+return{};
+}
+}
+
+conststr=JSON.stringify({asd: true});
+constserver=http.createServer(async(request,response)=>{
+constbody=awaitgetParsedBody(request);
+response.statusCode=200;
+assert.strictEqual(JSON.stringify(body),str);
+response.end(JSON.stringify(body));
+}).listen(()=>{
+http
+.request({
+method: 'POST',
+hostname: 'localhost',
+port: server.address().port,
+})
+.end(str)
+.on('response',async(res)=>{
+letbody='';
+forawait(constchunkofres){
+body+=chunk;
+}
+assert.strictEqual(body,str);
+server.close();
+});
+});
+}
+
 // To avoid missing some tests if a promise does not resolve
 tests().then(common.mustCall());