From ceaa3ebfcf60aee8b0d1fd7486f239d01ebd7c60 Mon Sep 17 00:00:00 2001 From: chenhang Date: Fri, 26 Jun 2026 11:24:48 +0800 Subject: [PATCH] refactor(message-queue): improve message handling and add options for joiner --- src/app/subscription/subscription-screen.tsx | 2 - src/core/net/__tests__/message-queue.test.ts | 120 +++++++++++++++++++ src/core/net/message-queue.ts | 103 ++++++++++++---- 3 files changed, 198 insertions(+), 27 deletions(-) create mode 100644 src/core/net/__tests__/message-queue.test.ts diff --git a/src/app/subscription/subscription-screen.tsx b/src/app/subscription/subscription-screen.tsx index 17c27b12..3b1b2ab7 100644 --- a/src/app/subscription/subscription-screen.tsx +++ b/src/app/subscription/subscription-screen.tsx @@ -2,8 +2,6 @@ /** * 订阅页主屏幕 * - * 原始 Dart: lib/ui/subscription/... - * * 布局（顶到底）： * 1. 顶部返回箭头 * 2. 用户信息行 diff --git a/src/core/net/__tests__/message-queue.test.ts b/src/core/net/__tests__/message-queue.test.ts new file mode 100644 index 00000000..b90bfb65 --- /dev/null +++ b/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((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(); + }); +}); diff --git a/src/core/net/message-queue.ts b/src/core/net/message-queue.ts index f8c929f3..1c8df807 100644 --- a/src/core/net/message-queue.ts +++ b/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; +export interface MessageQueueOptions { + intervalMs?: number; + joiner?: string; +} + const log = new Logger("MessageQueue"); export class MessageQueue { - private queue: string[] = []; + private pending: string[] = []; private timer: ReturnType | 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 { - 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 { + 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 { + 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 {