Refactor how it polls

Author: dahlia

.vscode/settings.json

@@ -1,6 +1,9 @@
 {
+  "cSpell.enabled": false,
   "deno.enable": true,
-  "deno.unstable": true,
+  "deno.unstable": [
+    "temporal"
+  ],
   "files.eol": "\n",
   "files.insertFinalNewline": true,
   "files.trimFinalNewlines": true,
@@ -18,8 +21,5 @@
     "editor.codeActionsOnSave": {
       "source.sortImports": "always"
     }
-  },
-  "cSpell.words": [
-    "fedify"
-  ]
+  }
 }

README.md

@@ -66,6 +66,10 @@ Changelog
 
 To be released.
 
+ -  Polling is now more efficient.
+ -  Renamed `RedisMessageQueueOptions.loopInterval` option to `pollInterval`
+    option.
+
 ### Version 0.2.0
 
 Released on September 26, 2024.

deno.json

@@ -11,9 +11,10 @@
   "imports": {
     "@deno/dnt": "jsr:@deno/dnt@^0.41.3",
     "@fedify/fedify": "jsr:@fedify/fedify@^1.0.0",
+    "@logtape/logtape": "jsr:@logtape/logtape@^0.6.3",
     "@std/assert": "jsr:@std/assert@^0.226.0",
     "@std/async": "jsr:@std/async@^0.224.2",
-    "ioredis": "npm:ioredis@^5.4.0"
+    "ioredis": "npm:ioredis@^5.4.1"
   },
   "unstable": [
     "temporal"

deno.lock

@@ -12,6 +12,9 @@ await Deno.writeTextFile(
       "@fedify/fedify": metadata.imports["@fedify/fedify"]
         .replace(/^jsr:/, "npm:")
         .replace(/\+.+$/, ""),
+      "@logtape/logtape": metadata.imports["@logtape/logtape"]
+        .replace(/^jsr:/, "npm:")
+        .replace(/\+.+$/, ""),
     },
   }),
 );

dnt.ts

@@ -1,4 +1,5 @@
-import { assertEquals, assertGreater } from "@std/assert";
+import { assertEquals } from "@std/assert/assert-equals";
+import { assertGreater } from "@std/assert/assert-greater";
 import { delay } from "@std/async/delay";
 import { Redis } from "ioredis";
 import { RedisMessageQueue } from "./mq.ts";
@@ -8,24 +9,24 @@ Deno.test("RedisMessageQueue", async (t) => {
   const queueKey = `fedify_test_queue_${crypto.randomUUID()}`;
   const lockKey = `fedify_test_lock_${crypto.randomUUID()}`;
   const mq = new RedisMessageQueue(() => new Redis(), {
-    loopInterval: { seconds: 1 },
+    pollInterval: { seconds: 1 },
     channelKey,
     queueKey,
     lockKey,
   });
   const mq2 = new RedisMessageQueue(() => new Redis(), {
-    loopInterval: { seconds: 1 },
+    pollInterval: { seconds: 1 },
     channelKey,
     queueKey,
     lockKey,
   });
 
-  const messages: string[] = [];
+  const messages: (string | number)[] = [];
   const controller = new AbortController();
-  const listening = mq.listen((message: string) => {
+  const listening = mq.listen((message: string | number) => {
     messages.push(message);
   }, controller);
-  const listening2 = mq2.listen((message: string) => {
+  const listening2 = mq2.listen((message: string | number) => {
     messages.push(message);
   }, controller);
 
@@ -55,6 +56,18 @@ Deno.test("RedisMessageQueue", async (t) => {
     assertGreater(Date.now() - started, 3_000);
   });
 
+  await t.step("enqueue() [bulk]", async () => {
+    for (let i = 0; i < 1_000; i++) await mq.enqueue(i);
+  });
+
+  await waitFor(() => messages.length > 1_001, 30_000);
+
+  await t.step("listen() [bulk]", () => {
+    const numbers: Set<number> = new Set();
+    for (let i = 0; i < 1_000; i++) numbers.add(i);
+    assertEquals(new Set(messages.slice(2)), numbers);
+  });
+
   controller.abort();
   await listening;
   await listening2;