From a295bb0d4255993103b6afe9ffdd2bd4e4c65c95 Mon Sep 17 00:00:00 2001
From: Ryan Dahl <ry@tinyclouds.org>
Date: Mon, 20 May 2019 09:17:26 -0400
Subject: Clean up HTTP async iterator code (denoland/deno_std#411)

Original: https://github.com/denoland/deno_std/commit/68faf32f721d2a95c7b1c75661713c8118c077c7
---
 util/async.ts      | 85 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
 util/async_test.ts | 34 ++++++++++++++++++++++
 util/test.ts       |  1 +
 3 files changed, 120 insertions(+)
 create mode 100644 util/async.ts
 create mode 100644 util/async_test.ts

(limited to 'util')

diff --git a/util/async.ts b/util/async.ts
new file mode 100644
index 000000000..f9f2477d0
--- /dev/null
+++ b/util/async.ts
@@ -0,0 +1,85 @@
+// Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
+
+// TODO(ry) It'd be better to make Deferred a class that inherits from
+// Promise, rather than an interface. This is possible in ES2016, however
+// typescript produces broken code when targeting ES5 code.
+// See https://github.com/Microsoft/TypeScript/issues/15202
+// At the time of writing, the github issue is closed but the problem remains.
+export interface Deferred<T> extends Promise<T> {
+  resolve: (value?: T | PromiseLike<T>) => void;
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  reject: (reason?: any) => void;
+}
+
+/** Creates a Promise with the `reject` and `resolve` functions
+ * placed as methods on the promise object itself. It allows you to do:
+ *
+ *     const p = deferred<number>();
+ *     // ...
+ *     p.resolve(42);
+ */
+export function deferred<T>(): Deferred<T> {
+  let methods;
+  const promise = new Promise<T>(
+    (resolve, reject): void => {
+      methods = { resolve, reject };
+    }
+  );
+  return Object.assign(promise, methods) as Deferred<T>;
+}
+
+interface TaggedYieldedValue<T> {
+  iterator: AsyncIterableIterator<T>;
+  value: T;
+}
+
+/** The MuxAsyncIterator class multiplexes multiple async iterators into a
+ * single stream. It currently makes a few assumptions:
+ * - The iterators do not throw.
+ * - 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>> = [];
+  private signal: Deferred<void> = deferred();
+
+  add(iterator: AsyncIterableIterator<T>): void {
+    ++this.iteratorCount;
+    this.callIteratorNext(iterator);
+  }
+
+  private async callIteratorNext(
+    iterator: AsyncIterableIterator<T>
+  ): Promise<void> {
+    const { value, done } = await iterator.next();
+    if (done) {
+      --this.iteratorCount;
+    } else {
+      this.yields.push({ iterator, value });
+    }
+    this.signal.resolve();
+  }
+
+  async *iterate(): AsyncIterableIterator<T> {
+    while (this.iteratorCount > 0) {
+      // Sleep until any of the wrapped iterators yields.
+      await this.signal;
+
+      // Note that while we're looping over `yields`, new items may be added.
+      for (let i = 0; i < this.yields.length; i++) {
+        const { iterator, value } = this.yields[i];
+        yield value;
+        this.callIteratorNext(iterator);
+      }
+
+      // Clear the `yields` list and reset the `signal` promise.
+      this.yields.length = 0;
+      this.signal = deferred();
+    }
+  }
+
+  [Symbol.asyncIterator](): AsyncIterableIterator<T> {
+    return this.iterate();
+  }
+}
diff --git a/util/async_test.ts b/util/async_test.ts
new file mode 100644
index 000000000..c704002d4
--- /dev/null
+++ b/util/async_test.ts
@@ -0,0 +1,34 @@
+// Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
+import { test, runIfMain } from "../testing/mod.ts";
+import { assertEquals } from "../testing/asserts.ts";
+import { MuxAsyncIterator, deferred } from "./async.ts";
+
+test(async function asyncDeferred(): Promise<void> {
+  const d = deferred<number>();
+  d.resolve(12);
+});
+
+async function* gen123(): AsyncIterableIterator<number> {
+  yield 1;
+  yield 2;
+  yield 3;
+}
+
+async function* gen456(): AsyncIterableIterator<number> {
+  yield 4;
+  yield 5;
+  yield 6;
+}
+
+test(async function asyncMuxAsyncIterator(): Promise<void> {
+  const mux = new MuxAsyncIterator<number>();
+  mux.add(gen123());
+  mux.add(gen456());
+  const results = new Set();
+  for await (const value of mux) {
+    results.add(value);
+  }
+  assertEquals(results.size, 6);
+});
+
+runIfMain(import.meta);
diff --git a/util/test.ts b/util/test.ts
index a617c10ab..ede984904 100644
--- a/util/test.ts
+++ b/util/test.ts
@@ -1 +1,2 @@
+import "./async_test.ts";
 import "./deep_assign_test.ts";
-- 
cgit v1.2.3