feat(kv) queue implementation (#19459)

Extend the unstable `Deno.Kv` API to support queues.

1 files changed, 56 insertions, 0 deletions

diff --git a/cli/tests/unit/kv_queue_undelivered_test.ts b/cli/tests/unit/kv_queue_undelivered_test.ts
new file mode 100644
index 000000000..c2c02905c
--- /dev/null
+++ b/cli/tests/unit/kv_queue_undelivered_test.ts
@@ -0,0 +1,56 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+import { assertEquals } from "./test_util.ts";
+
+const sleep = (time: number) => new Promise((r) => setTimeout(r, time));
+
+let isCI: boolean;
+try {
+  isCI = Deno.env.get("CI") !== undefined;
+} catch {
+  isCI = true;
+}
+
+function queueTest(name: string, fn: (db: Deno.Kv) => Promise<void>) {
+  Deno.test({
+    name,
+    // https://github.com/denoland/deno/issues/18363
+    ignore: Deno.build.os === "darwin" && isCI,
+    async fn() {
+      const db: Deno.Kv = await Deno.openKv(
+        ":memory:",
+      );
+      await fn(db);
+    },
+  });
+}
+
+async function collect<T>(
+  iter: Deno.KvListIterator<T>,
+): Promise<Deno.KvEntry<T>[]> {
+  const entries: Deno.KvEntry<T>[] = [];
+  for await (const entry of iter) {
+    entries.push(entry);
+  }
+  return entries;
+}
+
+queueTest("queue with undelivered", async (db) => {
+  const listener = db.listenQueue((_msg) => {
+    throw new TypeError("dequeue error");
+  });
+  try {
+    await db.enqueue("test", {
+      keysIfUndelivered: [["queue_failed", "a"], ["queue_failed", "b"]],
+    });
+    await sleep(100000);
+    const undelivered = await collect(db.list({ prefix: ["queue_failed"] }));
+    assertEquals(undelivered.length, 2);
+    assertEquals(undelivered[0].key, ["queue_failed", "a"]);
+    assertEquals(undelivered[0].value, "test");
+    assertEquals(undelivered[1].key, ["queue_failed", "b"]);
+    assertEquals(undelivered[1].value, "test");
+  } finally {
+    db.close();
+    await listener;
+  }
+});