79 lines
1.9 KiB
TypeScript
79 lines
1.9 KiB
TypeScript
/**
|
|
* 消息发送队列（限流）
|
|
*
|
|
* 原始 Dart: lib/core/net/message_queue.dart
|
|
*
|
|
* 业务行为：
|
|
* - 入队消息按 `intervalMs` 间隔逐条发送（默认 1000ms）
|
|
* - 消费者可声明每条消息是否成功消费；不成功则重新入队
|
|
* - 队列可在任意时刻清空（dispose）
|
|
*/
|
|
import { Logger } from "@/utils";
|
|
|
|
export type MessageConsumer = (content: string) => boolean | Promise<boolean>;
|
|
|
|
const log = new Logger("MessageQueue");
|
|
|
|
export class MessageQueue {
|
|
private queue: string[] = [];
|
|
private timer: ReturnType<typeof setTimeout> | null = null;
|
|
private disposed = false;
|
|
private consumer: MessageConsumer | null = null;
|
|
|
|
constructor(private readonly intervalMs = 1000) {}
|
|
|
|
/** 注册消费者（消费者必须能处理失败/重试）。 */
|
|
setConsumer(consumer: MessageConsumer): void {
|
|
this.consumer = consumer;
|
|
}
|
|
|
|
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);
|
|
return;
|
|
}
|
|
let ok = false;
|
|
try {
|
|
ok = await this.consumer(next);
|
|
} catch (e) {
|
|
log.error({ err: e }, "consumer error");
|
|
ok = false;
|
|
}
|
|
if (!ok) {
|
|
// 消费失败，放回队首等待下次
|
|
this.queue.unshift(next);
|
|
}
|
|
}
|
|
|
|
private scheduleNext(): void {
|
|
if (this.disposed) return;
|
|
const next = this.queue.shift();
|
|
if (next === undefined) {
|
|
this.timer = null;
|
|
return;
|
|
}
|
|
this.timer = setTimeout(() => {
|
|
void this.attemptSend(next).finally(() => this.scheduleNext());
|
|
}, this.intervalMs);
|
|
}
|
|
|
|
clear(): void {
|
|
this.queue = [];
|
|
}
|
|
|
|
dispose(): void {
|
|
this.disposed = true;
|
|
this.clear();
|
|
if (this.timer) {
|
|
clearTimeout(this.timer);
|
|
this.timer = null;
|
|
}
|
|
}
|
|
}
|