stream: close iterator in Readable.from

Call iterator.return() if not all of its values are consumed. Fixes: #32842 PR-URL: #32844 Reviewed-By: Robert Nagy <[email protected]> Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Zeyu Yang <[email protected]>

Author: vadzim

lib/internal/streams/from.js

@@ -34,21 +34,51 @@ function from(Readable, iterable, opts){
 objectMode: true,
  ...opts
 });
+
 // Reading boolean to protect against _read
 // being called before last iteration completion.
 letreading=false;
+
+// needToClose boolean if iterator needs to be explicitly closed
+letneedToClose=false;
+
 readable._read=function(){
 if(!reading){
 reading=true;
 next();
 }
 };
+
+readable._destroy=function(error,cb){
+if(needToClose){
+needToClose=false;
+close().then(
+()=>process.nextTick(cb,error),
+(e)=>process.nextTick(cb,error||e),
+);
+}else{
+cb(error);
+}
+};
+
+asyncfunctionclose(){
+if(typeofiterator.return==='function'){
+const{ value }=awaititerator.return();
+awaitvalue;
+}
+}
+
 asyncfunctionnext(){
 try{
+needToClose=false;
 const{ value, done }=awaititerator.next();
+needToClose=!done;
+constresolved=awaitvalue;
 if(done){
 readable.push(null);
-}elseif(readable.push(awaitvalue)){
+}elseif(readable.destroyed){
+awaitclose();
+}elseif(readable.push(resolved)){
 next();
 }else{
 reading=false;

test/parallel/test-readable-from-iterator-closing.js

@@ -0,0 +1,198 @@
+'use strict';
+
+const{ mustCall, mustNotCall }=require('../common');
+const{ Readable }=require('stream');
+const{ strictEqual }=require('assert');
+
+asyncfunctionasyncSupport(){
+constfinallyMustCall=mustCall();
+constbodyMustCall=mustCall();
+
+asyncfunction*infiniteGenerate(){
+try{
+while(true)yield'a';
+}finally{
+finallyMustCall();
+}
+}
+
+conststream=Readable.from(infiniteGenerate());
+
+forawait(constchunkofstream){
+bodyMustCall();
+strictEqual(chunk,'a');
+break;
+}
+}
+
+asyncfunctionsyncSupport(){
+constfinallyMustCall=mustCall();
+constbodyMustCall=mustCall();
+
+function*infiniteGenerate(){
+try{
+while(true)yield'a';
+}finally{
+finallyMustCall();
+}
+}
+
+conststream=Readable.from(infiniteGenerate());
+
+forawait(constchunkofstream){
+bodyMustCall();
+strictEqual(chunk,'a');
+break;
+}
+}
+
+asyncfunctionsyncPromiseSupport(){
+constreturnMustBeAwaited=mustCall();
+constbodyMustCall=mustCall();
+
+function*infiniteGenerate(){
+try{
+while(true)yieldPromise.resolve('a');
+}finally{
+// eslint-disable-next-line no-unsafe-finally
+return{then(cb){
+returnMustBeAwaited();
+cb();
+}};
+}
+}
+
+conststream=Readable.from(infiniteGenerate());
+
+forawait(constchunkofstream){
+bodyMustCall();
+strictEqual(chunk,'a');
+break;
+}
+}
+
+asyncfunctionsyncRejectedSupport(){
+constreturnMustBeAwaited=mustCall();
+constbodyMustNotCall=mustNotCall();
+constcatchMustCall=mustCall();
+constsecondNextMustNotCall=mustNotCall();
+
+function*generate(){
+try{
+yieldPromise.reject('a');
+secondNextMustNotCall();
+}finally{
+// eslint-disable-next-line no-unsafe-finally
+return{then(cb){
+returnMustBeAwaited();
+cb();
+}};
+}
+}
+
+conststream=Readable.from(generate());
+
+try{
+forawait(constchunkofstream){
+bodyMustNotCall(chunk);
+}
+}catch{
+catchMustCall();
+}
+}
+
+asyncfunctionnoReturnAfterThrow(){
+constreturnMustNotCall=mustNotCall();
+constbodyMustNotCall=mustNotCall();
+constcatchMustCall=mustCall();
+constnextMustCall=mustCall();
+
+conststream=Readable.from({
+[Symbol.asyncIterator](){returnthis;},
+asyncnext(){
+nextMustCall();
+thrownewError('a');
+},
+asyncreturn(){
+returnMustNotCall();
+return{done: true};
+},
+});
+
+try{
+forawait(constchunkofstream){
+bodyMustNotCall(chunk);
+}
+}catch{
+catchMustCall();
+}
+}
+
+asyncfunctioncloseStreamWhileNextIsPending(){
+constfinallyMustCall=mustCall();
+constdataMustCall=mustCall();
+
+letresolveDestroy;
+constdestroyed=
+newPromise((resolve)=>{resolveDestroy=mustCall(resolve);});
+letresolveYielded;
+constyielded=
+newPromise((resolve)=>{resolveYielded=mustCall(resolve);});
+
+asyncfunction*infiniteGenerate(){
+try{
+while(true){
+yield'a';
+resolveYielded();
+awaitdestroyed;
+}
+}finally{
+finallyMustCall();
+}
+}
+
+conststream=Readable.from(infiniteGenerate());
+
+stream.on('data',(data)=>{
+dataMustCall();
+strictEqual(data,'a');
+});
+
+yielded.then(()=>{
+stream.destroy();
+resolveDestroy();
+});
+}
+
+asyncfunctioncloseAfterNullYielded(){
+constfinallyMustCall=mustCall();
+constdataMustCall=mustCall(3);
+
+function*infiniteGenerate(){
+try{
+yield'a';
+yield'a';
+yield'a';
+while(true)yieldnull;
+}finally{
+finallyMustCall();
+}
+}
+
+conststream=Readable.from(infiniteGenerate());
+
+stream.on('data',(chunk)=>{
+dataMustCall();
+strictEqual(chunk,'a');
+});
+}
+
+Promise.all([
+asyncSupport(),
+syncSupport(),
+syncPromiseSupport(),
+syncRejectedSupport(),
+noReturnAfterThrow(),
+closeStreamWhileNextIsPending(),
+closeAfterNullYielded(),
+]).then(mustCall());