✨ implemented

Author: takker99

README.md

@@ -1,2 +1,3 @@
 # async-lib
+
 便利な非同期関数置き場

delay.ts

@@ -0,0 +1,2 @@
+export const delay = (milliseconds: number): Promise<void> =>
+  new Promise((resolve) => setTimeout(resolve, milliseconds));

deps_test.ts

@@ -0,0 +1 @@
+export * from "https://deno.land/std@0.129.0/testing/asserts.ts";

mod.ts

@@ -0,0 +1,4 @@
+export * from "./promisify.ts";
+export * from "./pool.ts";
+export * from "./sort.ts";
+export * from "./types.ts";

pool.test.ts

@@ -0,0 +1,53 @@
+import { delay } from "./delay.ts";
+import { pool } from "./pool.ts";
+import { assert, assertEquals, assertStringIncludes } from "./deps_test.ts";
+
+Deno.test("pool()", async (t) => {
+  await t.step("order", async () => {
+    const start = new Date();
+    const results = pool(
+      2,
+      [1, 2, 3],
+      (i, id) => delay(1000).then(() => [i, id]),
+    );
+    const list = await Promise.all(results);
+
+    const diff = new Date().getTime() - start.getTime();
+    assert(diff >= 2000);
+    assert(diff < 3000);
+
+    assertEquals(
+      list.flatMap((result) => result.success ? [result.value] : []),
+      [
+        [1, 0],
+        [2, 1],
+        [3, 0],
+      ],
+    );
+  });
+
+  await t.step("error", async () => {
+    async function mapNumber(n: number): Promise<number> {
+      if (n <= 2) throw new Error(`Bad number: ${n}`);
+      await delay(100);
+      return n;
+    }
+    const mappedNumbers: number[] = [];
+    const errors: unknown[] = [];
+    for (const promise of pool(3, [1, 2, 3, 4], mapNumber)) {
+      const m = await promise;
+      if (m.success) {
+        mappedNumbers.push(m.value);
+      } else {
+        errors.push(m.reason);
+      }
+    }
+
+    assertEquals(errors.length, 2);
+    assert(errors[0] instanceof Error);
+    assert(errors[1] instanceof Error);
+    assertStringIncludes(errors[0].stack ?? "", "Error: Bad number: 1");
+    assertStringIncludes(errors[1].stack ?? "", "Error: Bad number: 2");
+    assertEquals(mappedNumbers, [3, 4]);
+  });
+});

pool.ts

@@ -0,0 +1,37 @@
+import type { Result } from "./types.ts";
+
+export function* pool<T, U, E = unknown>(
+  threshold: number,
+  args: Iterable<T>,
+  func: (value: T, id: number) => Promise<U>,
+): Generator<Promise<Result<U, E>>, void, unknown> {
+  let running = 0;
+  const waitings = [] as ((value: number | PromiseLike<number>) => void)[];
+  const waitForReady = async () => {
+    running++;
+    if (running <= threshold) return running - 1;
+    return await new Promise<number>(
+      (resolve) => waitings.push(resolve),
+    );
+  };
+  for (const arg of args) {
+    yield (async () => {
+      const id = await waitForReady();
+      try {
+        return {
+          success: true,
+          value: await func(arg, id),
+        };
+      } catch (e: unknown) {
+        return {
+          success: false,
+          reason: e as E,
+        };
+      } finally {
+        running--;
+        waitings.shift()?.(id);
+      }
+    })();
+  }
+  // 実行しきれない可能性はある？
+}

promisify.test.ts

@@ -0,0 +1,94 @@
+import { promisify } from "./promisify.ts";
+import { assert } from "./deps_test.ts";
+
+Deno.test("promisify()", async (t) => {
+  await t.step("keep all results", async () => {
+    const [get, resolve] = promisify<number>();
+
+    const pending = get();
+    resolve(0);
+    resolve(1);
+    assert(await pending === 0);
+    assert(await get() === 1);
+    resolve(2);
+    assert(await get() === 2);
+    resolve(3);
+    resolve(4);
+    assert(await get() === 3);
+    assert(await get() === 4);
+    resolve(5);
+    resolve(6);
+    resolve(7);
+    assert(await get() === 5);
+    resolve(8);
+    assert(await get() === 6);
+    assert(await get() === 7);
+    assert(await get() === 8);
+  });
+
+  await t.step("keep nothing", async () => {
+    const [get, resolve] = promisify<number>({ maxQueued: 0 });
+
+    let pending = get();
+    resolve(1);
+    assert(await pending === 1);
+    resolve(2);
+    resolve(3);
+    pending = get();
+    resolve(4);
+    assert(await pending === 4);
+    pending = get();
+    resolve(5);
+    resolve(6);
+    assert(await pending === 5);
+  });
+
+  await t.step("keep the latest result", async () => {
+    const [get, resolve] = promisify<number>({ maxQueued: 1 });
+
+    let pending = get();
+    resolve(1);
+    assert(await pending === 1);
+    resolve(2);
+    resolve(3);
+    assert(await get() === 3);
+    resolve(4);
+    resolve(5);
+    resolve(6);
+    assert(await get() === 6);
+    pending = get();
+    resolve(7);
+    resolve(8);
+    resolve(9);
+    assert(await pending === 7);
+    assert(await get() === 9);
+  });
+
+  await t.step("keep the latest and second latest results", async () => {
+    const [get, resolve] = promisify<number>({ maxQueued: 2 });
+
+    let pending = get();
+    resolve(1);
+    assert(await pending === 1);
+    resolve(2);
+    resolve(3);
+    assert(await get() === 2);
+    assert(await get() === 3);
+    resolve(4);
+    resolve(5);
+    resolve(6);
+    assert(await get() === 5);
+    resolve(7);
+    resolve(8);
+    assert(await get() === 7);
+    assert(await get() === 8);
+    pending = get();
+    resolve(9);
+    resolve(10);
+    resolve(11);
+    resolve(12);
+    assert(await pending === 9);
+    assert(await get() === 11);
+    assert(await get() === 12);
+  });
+});

promisify.ts

@@ -0,0 +1,90 @@
+import type { Result } from "./types.ts";
+
+export interface PromisifyOptions {
+  /** callbackから次の値を取り出すまでに実行されたcallbackの結果を最新でいくつまで保持するかを表す値
+   *
+   * - 最後に実行された値だけを残したいときは`1`を指定する
+   * - 一つも残したくないときは`0`を指定する
+   * - 全部残したいときはundefinedを指定する
+   *   - これが既定の動作
+   *
+   * @default undefined
+   */
+  maxQueued?: number | undefined;
+}
+/** callbackをPromiseに変換するやつ
+ *
+ * @return 左から順に、Promiseを返すやつ、正常値を受け取るcallback、異常値を受け取るcallback
+ */
+export function promisify<T, E = unknown>(
+  options?: PromisifyOptions,
+): readonly [
+  () => Promise<T>,
+  (value: T) => void,
+  (reason: E) => void,
+] {
+  const maxQueued = options?.maxQueued === undefined
+    ? undefined
+    : Math.max(0, options.maxQueued);
+  if (maxQueued === 0) return promisifyWithoutQueue();
+
+  const queue = [] as Result<T, E>[];
+  let _resolve: ((value: T) => void) | undefined;
+  let _reject: ((value: E) => void) | undefined;
+
+  /** queueから一つ取り出す。空なら_resolveをセットする */
+  const waitForSettled = async () => {
+    if (maxQueued !== undefined) queue.splice(0, queue.length - maxQueued);
+    const value = queue.shift();
+    if (value) {
+      if (value.success) return value.value;
+      throw value.reason;
+    }
+
+    return await new Promise<T>(
+      (res, rej) => {
+        _resolve = res;
+        _reject = rej;
+      },
+    );
+  };
+  const resolve = (value: T) => {
+    if (!_resolve) {
+      queue.push({ success: true, value });
+      return;
+    }
+    _resolve(value);
+    _resolve = _reject = undefined;
+  };
+  const reject = (value: E) => {
+    if (!_reject) {
+      queue.push({ success: false, reason: value });
+      return;
+    }
+    _reject(value);
+    _resolve = _reject = undefined;
+  };
+
+  return [waitForSettled, resolve, reject] as const;
+}
+
+function promisifyWithoutQueue<T, E = unknown>(): readonly [
+  () => Promise<T>,
+  (value: T) => void,
+  (reason: E) => void,
+] {
+  let _resolve: ((value: T) => void) | undefined;
+  let _reject: ((value: E) => void) | undefined;
+
+  const waitForSettled = () =>
+    new Promise<T>(
+      (res, rej) => {
+        _resolve = res;
+        _reject = rej;
+      },
+    );
+  const resolve = (value: T) => _resolve?.(value);
+  const reject = (value: E) => _reject?.(value);
+
+  return [waitForSettled, resolve, reject] as const;
+}

sort.test.ts

@@ -0,0 +1,82 @@
+import { delay } from "./delay.ts";
+import type { Result } from "./types.ts";
+import { sort, sortSettled } from "./sort.ts";
+import { assertEquals } from "./deps_test.ts";
+
+Deno.test("sort()", async (t) => {
+  await t.step("return in order of settled", async () => {
+    const results: number[] = [];
+
+    for await (
+      const waited of sort(
+        [1000, 2000, 500, 1500].map((n, index) => delay(n).then(() => index)),
+      )
+    ) {
+      results.push(waited);
+    }
+
+    assertEquals(results, [2, 0, 3, 1]);
+  });
+
+  await t.step("ignore errors", async () => {
+    const results: number[] = [];
+
+    for await (
+      const waited of sort(
+        [1000, 2000, 1500, 500].map(async (n, index) => {
+          if (index < 2) throw Error(`Error: ${index}`);
+          await delay(n);
+          return index;
+        }),
+      )
+    ) {
+      results.push(waited);
+    }
+
+    assertEquals(results, [3, 2]);
+  });
+});
+
+Deno.test("sortSettled()", async (t) => {
+  await t.step("return in order of settled", async () => {
+    const results: Result<number>[] = [];
+
+    for await (
+      const waited of sortSettled(
+        [1000, 2000, 500, 1500].map((n, index) => delay(n).then(() => index)),
+      )
+    ) {
+      results.push(waited);
+    }
+
+    assertEquals(results, [
+      { success: true, value: 2 },
+      { success: true, value: 0 },
+      { success: true, value: 3 },
+      { success: true, value: 1 },
+    ]);
+  });
+
+  await t.step("catch errors", async () => {
+    const results: Result<number>[] = [];
+
+    for await (
+      const waited of sortSettled(
+        [1000, 2000, 1500, 500].map(async (n, index) => {
+          if (index < 2) throw `Error: ${index}`;
+          await delay(n);
+          return index;
+        }),
+      )
+    ) {
+      results.push(waited);
+    }
+
+    assertEquals(results, [
+      { success: false, reason: "Error: 0" },
+      { success: false, reason: "Error: 1" },
+      { success: true, value: 3 },
+      { success: true, value: 2 },
+    ]);
+  });
+});