refactor(message-queue): improve message handling and add options for joiner

src/core/net/__tests__/message-queue.test.ts

@@ -0,0 +1,120 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+import { MessageQueue } from "@/core/net/message-queue";
+
+describe("MessageQueue", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  it("sends the first message immediately", async () => {
+    const consumer = vi.fn(() => true);
+    const queue = new MessageQueue();
+    queue.setConsumer(consumer);
+
+    queue.enqueue("hello");
+    await Promise.resolve();
+
+    expect(consumer).toHaveBeenCalledTimes(1);
+    expect(consumer).toHaveBeenCalledWith("hello");
+
+    queue.dispose();
+  });
+
+  it("merges pending messages after the interval", async () => {
+    const consumer = vi.fn(() => true);
+    const queue = new MessageQueue({ intervalMs: 1000 });
+    queue.setConsumer(consumer);
+
+    queue.enqueue("first");
+    await Promise.resolve();
+    queue.enqueue("second");
+    queue.enqueue("third");
+
+    expect(consumer).toHaveBeenCalledTimes(1);
+
+    await vi.advanceTimersByTimeAsync(999);
+    expect(consumer).toHaveBeenCalledTimes(1);
+
+    await vi.advanceTimersByTimeAsync(1);
+    expect(consumer).toHaveBeenCalledTimes(2);
+    expect(consumer).toHaveBeenLastCalledWith("second\nthird");
+
+    queue.dispose();
+  });
+
+  it("does not send a pending batch while a previous send is still running", async () => {
+    let resolveFirstSend: (ok: boolean) => void = () => undefined;
+    const consumer = vi.fn(
+      () =>
+        new Promise<boolean>((resolve) => {
+          resolveFirstSend = resolve;
+        }),
+    );
+    const queue = new MessageQueue({ intervalMs: 1000 });
+    queue.setConsumer(consumer);
+
+    queue.enqueue("first");
+    await Promise.resolve();
+    queue.enqueue("second");
+    queue.enqueue("third");
+
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(consumer).toHaveBeenCalledTimes(1);
+
+    resolveFirstSend(true);
+    await Promise.resolve();
+    await vi.advanceTimersByTimeAsync(1000);
+
+    expect(consumer).toHaveBeenCalledTimes(2);
+    expect(consumer).toHaveBeenLastCalledWith("second\nthird");
+
+    queue.dispose();
+  });
+
+  it("requeues a failed batch and merges it with newer pending messages", async () => {
+    const consumer = vi
+      .fn((content: string) => content.length > 0)
+      .mockReturnValueOnce(true)
+      .mockReturnValueOnce(false)
+      .mockReturnValue(true);
+    const queue = new MessageQueue({ intervalMs: 1000 });
+    queue.setConsumer(consumer);
+
+    queue.enqueue("first");
+    await Promise.resolve();
+    queue.enqueue("second");
+    queue.enqueue("third");
+
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(consumer).toHaveBeenLastCalledWith("second\nthird");
+
+    queue.enqueue("fourth");
+    await vi.advanceTimersByTimeAsync(1000);
+
+    expect(consumer).toHaveBeenLastCalledWith("second\nthird\nfourth");
+
+    queue.dispose();
+  });
+
+  it("supports a custom joiner", async () => {
+    const consumer = vi.fn(() => true);
+    const queue = new MessageQueue({ intervalMs: 1000, joiner: " " });
+    queue.setConsumer(consumer);
+
+    queue.enqueue("one");
+    await Promise.resolve();
+    queue.enqueue("two");
+    queue.enqueue("three");
+
+    await vi.advanceTimersByTimeAsync(1000);
+
+    expect(consumer).toHaveBeenLastCalledWith("two three");
+
+    queue.dispose();
+  });
+});

src/core/net/message-queue.ts

@@ -1,26 +1,42 @@
 /**
  * 消息发送队列（限流）
  *
- * 原始 Dart: lib/core/net/message_queue.dart
- *
  * 业务行为：
- *   - 入队消息按 `intervalMs` 间隔逐条发送（默认 1000ms）
- *   - 消费者可声明每条消息是否成功消费；不成功则重新入队
+ *   - 队列空闲时，第一条消息立即发送
+ *   - 后续消息在冷却窗口内暂存，并按 `intervalMs` 合并成一条发送
+ *   - 消费者可声明每批消息是否成功消费；不成功则重新入队
  *   - 队列可在任意时刻清空（dispose）
  */
 import { Logger } from "@/utils";
 
 export type MessageConsumer = (content: string) => boolean | Promise<boolean>;
 
+export interface MessageQueueOptions {
+  intervalMs?: number;
+  joiner?: string;
+}
+
 const log = new Logger("MessageQueue");
 
 export class MessageQueue {
-  private queue: string[] = [];
+  private pending: string[] = [];
   private timer: ReturnType<typeof setTimeout> | null = null;
+  private sending = false;
   private disposed = false;
   private consumer: MessageConsumer | null = null;
+  private readonly intervalMs: number;
+  private readonly joiner: string;
 
-  constructor(private readonly intervalMs = 1000) {}
+  constructor(options: MessageQueueOptions | number = 1000) {
+    if (typeof options === "number") {
+      this.intervalMs = options;
+      this.joiner = "\n";
+      return;
+    }
+
+    this.intervalMs = options.intervalMs ?? 1000;
+    this.joiner = options.joiner ?? "\n";
+  }
 
   /** 注册消费者（消费者必须能处理失败/重试）。 */
   setConsumer(consumer: MessageConsumer): void {
@@ -29,42 +45,79 @@ export class MessageQueue {
 
   enqueue(content: string): void {
     if (this.disposed) return;
-    this.queue.push(content);
-    if (!this.timer) this.scheduleNext();
-  }
-
-  private async attemptSend(next: string): Promise<void> {
-    if (!this.consumer) {
-      this.queue.unshift(next);
+    if (!this.sending && !this.timer && this.pending.length === 0) {
+      log.debug("send immediate", { contentLength: content.length });
+      void this.sendBatch(content);
       return;
     }
+
+    this.pending.push(content);
+    log.debug("enqueue pending", {
+      pendingCount: this.pending.length,
+      contentLength: content.length,
+    });
+  }
+
+  private async sendBatch(content: string): Promise<void> {
+    if (this.disposed) return;
+    this.sending = true;
+    this.scheduleFlush();
+    if (!this.consumer) {
+      this.pending.unshift(content);
+      this.sending = false;
+      this.scheduleFlush();
+      return;
+    }
+
     let ok = false;
     try {
-      ok = await this.consumer(next);
+      ok = await this.consumer(content);
     } catch (e) {
       log.error({ err: e }, "consumer error");
       ok = false;
     }
+
+    this.sending = false;
     if (!ok) {
-      // 消费失败，放回队首等待下次
-      this.queue.unshift(next);
+      // 消费失败，整批放回队首等待下次合并发送。
+      this.pending.unshift(content);
+      log.debug("retry batch", {
+        pendingCount: this.pending.length,
+        contentLength: content.length,
+      });
     }
+
+    this.scheduleFlush();
   }
 
-  private scheduleNext(): void {
-    if (this.disposed) return;
-    const next = this.queue.shift();
-    if (next === undefined) {
-      this.timer = null;
-      return;
-    }
+  private scheduleFlush(): void {
+    if (this.disposed || this.timer) return;
     this.timer = setTimeout(() => {
-      void this.attemptSend(next).finally(() => this.scheduleNext());
+      this.timer = null;
+      void this.flushPending();
     }, this.intervalMs);
   }
 
+  private async flushPending(): Promise<void> {
+    if (this.disposed) return;
+    if (this.sending) {
+      this.scheduleFlush();
+      return;
+    }
+
+    const batch = this.pending.splice(0, this.pending.length);
+    if (batch.length === 0) return;
+
+    const content = batch.join(this.joiner);
+    log.debug("flush batch", {
+      batchCount: batch.length,
+      contentLength: content.length,
+    });
+    await this.sendBatch(content);
+  }
+
   clear(): void {
-    this.queue = [];
+    this.pending = [];
   }
 
   dispose(): void {