stream: make async iterator .next() always resolve

See: nodejs/readable-stream#387
PR-URL: #24668
Reviewed-By: Gus Caplan <me@gus.host>
Reviewed-By: Franziska Hinkelmann <franziska.hinkelmann@gmail.com>
Reviewed-By: Joyee Cheung <joyeec9h3@gmail.com>
Reviewed-By: Ruben Bridgewater <ruben@bridgewater.de>

Author: mcollina

lib/internal/streams/async_iterator.js

@@ -39,6 +39,11 @@ function onReadable(iter) {
 functionwrapForNext(lastPromise,iter){
 return(resolve,reject)=>{
 lastPromise.then(()=>{
+if(iter[kEnded]){
+resolve(createIterResult(undefined,true));
+return;
+}
+
 iter[kHandlePromise](resolve,reject);
 },reject);
 };
@@ -61,7 +66,7 @@ const ReadableStreamAsyncIteratorPrototype = Object.setPrototypeOf({
 }
 
 if(this[kEnded]){
-returnPromise.resolve(createIterResult(null,true));
+returnPromise.resolve(createIterResult(undefined,true));
 }
 
 if(this[kStream].destroyed){
@@ -74,7 +79,7 @@ const ReadableStreamAsyncIteratorPrototype = Object.setPrototypeOf({
 if(this[kError]){
 reject(this[kError]);
 }else{
-resolve(createIterResult(null,true));
+resolve(createIterResult(undefined,true));
 }
 });
 });
@@ -115,7 +120,7 @@ const ReadableStreamAsyncIteratorPrototype = Object.setPrototypeOf({
 reject(err);
 return;
 }
-resolve(createIterResult(null,true));
+resolve(createIterResult(undefined,true));
 });
 });
 },
@@ -131,7 +136,6 @@ const createReadableStreamAsyncIterator = (stream) => {
 value: stream._readableState.endEmitted,
 writable: true
 },
-[kLastPromise]: {value: null,writable: true},
 // the function passed to new Promise
 // is cached so we avoid allocating a new
 // closure at every run
@@ -151,6 +155,7 @@ const createReadableStreamAsyncIterator = (stream) => {
 writable: true,
 },
 });
+iterator[kLastPromise]=null;
 
 finished(stream,(err)=>{
 if(err&&err.code!=='ERR_STREAM_PREMATURE_CLOSE'){
@@ -172,7 +177,7 @@ const createReadableStreamAsyncIterator = (stream) => {
 iterator[kLastPromise]=null;
 iterator[kLastResolve]=null;
 iterator[kLastReject]=null;
-resolve(createIterResult(null,true));
+resolve(createIterResult(undefined,true));
 }
 iterator[kEnded]=true;
 });

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

@@ -393,6 +393,71 @@ async function tests() {
 r.destroy(null);
 }
 })();
+
+await(async()=>{
+console.log('all next promises must be resolved on end');
+constr=newReadable({
+objectMode: true,
+read(){
+}
+});
+
+constb=r[Symbol.asyncIterator]();
+constc=b.next();
+constd=b.next();
+r.push(null);
+assert.deepStrictEqual(awaitc,{done: true,value: undefined});
+assert.deepStrictEqual(awaitd,{done: true,value: undefined});
+})();
+
+await(async()=>{
+console.log('all next promises must be resolved on destroy');
+constr=newReadable({
+objectMode: true,
+read(){
+}
+});
+
+constb=r[Symbol.asyncIterator]();
+constc=b.next();
+constd=b.next();
+r.destroy();
+assert.deepStrictEqual(awaitc,{done: true,value: undefined});
+assert.deepStrictEqual(awaitd,{done: true,value: undefined});
+})();
+
+await(async()=>{
+console.log('all next promises must be resolved on destroy with error');
+constr=newReadable({
+objectMode: true,
+read(){
+}
+});
+
+constb=r[Symbol.asyncIterator]();
+constc=b.next();
+constd=b.next();
+consterr=newError('kaboom');
+r.destroy(err);
+
+awaitPromise.all([(async()=>{
+lete;
+try{
+awaitc;
+}catch(_e){
+e=_e;
+}
+assert.strictEqual(e,err);
+})(),(async()=>{
+lete;
+try{
+awaitd;
+}catch(_e){
+e=_e;
+}
+assert.strictEqual(e,err);
+})()]);
+})();
 }
 
 // to avoid missing some tests if a promise does not resolve