stream: async iterator destroy compat

lib/internal/streams/async_iterator.js

@@ -110,17 +110,26 @@ const ReadableStreamAsyncIteratorPrototype = Object.setPrototypeOf({
   },
 
   return() {
-    // destroy(err, cb) is a private API.
-    // We can guarantee we have that here, because we control the
-    // Readable class this is attached to.
     return new Promise((resolve, reject) => {
-      this[kStream].destroy(null, (err) => {
-        if (err) {
+      const stream = this[kStream];
+
+      // TODO(ronag): Remove this check once finished() handles
+      // already ended and/or destroyed streams.
+      const ended = stream.destroyed || stream.readableEnded ||
+        (stream._readableState && stream._readableState.endEmitted);
+      if (ended) {
+        resolve(createIterResult(undefined, true));
+        return;
+      }
+
+      finished(stream, (err) => {
+        if (err && err.code !== 'ERR_STREAM_PREMATURE_CLOSE') {
           reject(err);
-          return;
+        } else {
+          resolve(createIterResult(undefined, true));
         }
-        resolve(createIterResult(undefined, true));
       });
+      stream.destroy();
     });
   },
 }, AsyncIteratorPrototype);

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

@@ -486,5 +486,46 @@ async function tests() {
   }
 }
 
+{
+  // AsyncIterator return should end even when destroy
+  // does not implement the callback API.
+
+  const r = new Readable({
+    objectMode: true,
+    read() {
+    }
+  });
+
+  const originalDestroy = r.destroy;
+  r.destroy = (err) => {
+    originalDestroy.call(r, err);
+  };
+  const it = r[Symbol.asyncIterator]();
+  const p = it.return();
+  r.push(null);
+  p.then(common.mustCall());
+}
+
+
+{
+  // AsyncIterator return should not error with
+  // premature close.
+
+  const r = new Readable({
+    objectMode: true,
+    read() {
+    }
+  });
+
+  const originalDestroy = r.destroy;
+  r.destroy = (err) => {
+    originalDestroy.call(r, err);
+  };
+  const it = r[Symbol.asyncIterator]();
+  const p = it.return();
+  r.emit('close');
+  p.then(common.mustCall()).catch(common.mustNotCall());
+}
+
 // To avoid missing some tests if a promise does not resolve
 tests().then(common.mustCall());