stream: normalize async iterator stream destroy

ronag · codebytere · commit 5a95fa4aeb71 · 2020-02-17T11:21:32.000-08:00
PR-URL: #31316 Reviewed-By: Anna Henningsen <anna@addaleax.net> Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Rich Trott <rtrott@gmail.com> Reviewed-By: Minwoo Jung <nodecorelab@gmail.com>
diff --git a/lib/internal/streams/async_iterator.js b/lib/internal/streams/async_iterator.js
@@ -22,6 +22,15 @@ const kStream = Symbol('stream');
 
 let Readable;
 
+function destroy(stream, err){
+ // request.destroy just do .end - .abort is what we want
+ if (typeof stream.abort === 'function') return stream.abort();
+ if (stream.req &&
+ typeof stream.req.abort === 'function') return stream.req.abort();
+ if (typeof stream.destroy === 'function') return stream.destroy(err);
+ if (typeof stream.close === 'function') return stream.close();
+}
+
 function createIterResult(value, done){
 return{value, done };
 }
@@ -141,7 +150,7 @@ const ReadableStreamAsyncIteratorPrototype = ObjectSetPrototypeOf({
 resolve(createIterResult(undefined, true));
 }
 });
- stream.destroy();
+ destroy(stream);
 });
 },
 }, AsyncIteratorPrototype);
@@ -156,11 +165,7 @@ const createReadableStreamAsyncIterator = (stream) =>{
 
 const src = stream;
 stream = new Readable({objectMode: true }).wrap(src);
- finished(stream, (err) =>{
- if (typeof src.destroy === 'function'){
- src.destroy(err);
- }
- });
+ finished(stream, (err) => destroy(src, err));
 }
 
 const iterator = ObjectCreate(ReadableStreamAsyncIteratorPrototype,{
diff --git a/test/parallel/test-stream-readable-async-iterators.js b/test/parallel/test-stream-readable-async-iterators.js
@@ -56,6 +56,25 @@ async function tests(){
 }));
 }
 
+{
+ // Non standard stream cleanup
+
+ const readable = new Readable({autoDestroy: false, read(){} });
+ readable.push('asd');
+ readable.push('asd');
+ readable.destroy = null;
+ readable.close = common.mustCall(() =>{
+ readable.emit('close');
+ });
+
+ await (async () =>{
+ for await (const d of readable){
+ d;
+ return;
+ }
+ })();
+ }
+
{
 const readable = new Readable({objectMode: true, read(){} });
 readable.push(0);

-Original file line number
+Diff line change
 letReadable;
 +functiondestroy(stream,err){
 +// request.destroy just do .end - .abort is what we want
 +if(typeofstream.abort==='function')returnstream.abort();
 +if(stream.req&&
 +typeofstream.req.abort==='function')returnstream.req.abort();
 +if(typeofstream.destroy==='function')returnstream.destroy(err);
 +if(typeofstream.close==='function')returnstream.close();
 +}
++
 functioncreateIterResult(value,done){
 return{ value, done };
+}
 resolve(createIterResult(undefined,true));
+}
 });
 -stream.destroy();
 +destroy(stream);
 });
 },
 },AsyncIteratorPrototype);
 constsrc=stream;
 stream=newReadable({objectMode: true}).wrap(src);
 -finished(stream,(err)=>{
 -if(typeofsrc.destroy==='function'){
 -src.destroy(err);
 -}
 -});
 +finished(stream,(err)=>destroy(src,err));
+}
 constiterator=ObjectCreate(ReadableStreamAsyncIteratorPrototype,{