fix: MuxAsyncIterator throws muxed errors (#6295)

Fixes #5260

2 files changed, 44 insertions, 8 deletions

diff --git a/std/async/mux_async_iterator.ts b/std/async/mux_async_iterator.ts
index b32689a29..ba6f22775 100644
--- a/std/async/mux_async_iterator.ts
+++ b/std/async/mux_async_iterator.ts
@@ -7,14 +7,15 @@ interface TaggedYieldedValue<T> {
 }
 
 /** The MuxAsyncIterator class multiplexes multiple async iterators into a
- * single stream. It currently makes a few assumptions:
- * - The iterators do not throw.
+ * single stream. It currently makes an assumption:
  * - The final result (the value returned and not yielded from the iterator)
  *   does not matter; if there is any, it is discarded.
  */
 export class MuxAsyncIterator<T> implements AsyncIterable<T> {
   private iteratorCount = 0;
   private yields: Array<TaggedYieldedValue<T>> = [];
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  private throws: any[] = [];
   private signal: Deferred<void> = deferred();
 
   add(iterator: AsyncIterableIterator<T>): void {
@@ -25,11 +26,15 @@ export class MuxAsyncIterator<T> implements AsyncIterable<T> {
   private async callIteratorNext(
     iterator: AsyncIterableIterator<T>
   ): Promise<void> {
-    const { value, done } = await iterator.next();
-    if (done) {
-      --this.iteratorCount;
-    } else {
-      this.yields.push({ iterator, value });
+    try {
+      const { value, done } = await iterator.next();
+      if (done) {
+        --this.iteratorCount;
+      } else {
+        this.yields.push({ iterator, value });
+      }
+    } catch (e) {
+      this.throws.push(e);
     }
     this.signal.resolve();
   }
@@ -46,6 +51,12 @@ export class MuxAsyncIterator<T> implements AsyncIterable<T> {
         this.callIteratorNext(iterator);
       }
 
+      if (this.throws.length) {
+        for (const e of this.throws) {
+          throw e;
+        }
+        this.throws.length = 0;
+      }
       // Clear the `yields` list and reset the `signal` promise.
       this.yields.length = 0;
       this.signal = deferred();
diff --git a/std/async/mux_async_iterator_test.ts b/std/async/mux_async_iterator_test.ts
index 7017a4eba..5ca28903b 100644
--- a/std/async/mux_async_iterator_test.ts
+++ b/std/async/mux_async_iterator_test.ts
@@ -1,5 +1,5 @@
 // Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
-import { assertEquals } from "../testing/asserts.ts";
+import { assertEquals, assertThrowsAsync } from "../testing/asserts.ts";
 import { MuxAsyncIterator } from "./mux_async_iterator.ts";
 
 // eslint-disable-next-line require-await
@@ -16,6 +16,12 @@ async function* gen456(): AsyncIterableIterator<number> {
   yield 6;
 }
 
+// eslint-disable-next-line require-await
+async function* genThrows(): AsyncIterableIterator<number> {
+  yield 7;
+  throw new Error("something went wrong");
+}
+
 Deno.test("[async] MuxAsyncIterator", async function (): Promise<void> {
   const mux = new MuxAsyncIterator<number>();
   mux.add(gen123());
@@ -26,3 +32,22 @@ Deno.test("[async] MuxAsyncIterator", async function (): Promise<void> {
   }
   assertEquals(results.size, 6);
 });
+
+Deno.test({
+  name: "[async] MuxAsyncIterator throws",
+  async fn() {
+    const mux = new MuxAsyncIterator<number>();
+    mux.add(gen123());
+    mux.add(genThrows());
+    const results = new Set();
+    await assertThrowsAsync(
+      async () => {
+        for await (const value of mux) {
+          results.add(value);
+        }
+      },
+      Error,
+      "something went wrong"
+    );
+  },
+});