File size: 2,653 Bytes
1f21206 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | /**
* 流式消息缓冲
*
* 将 content_delta 累积后按时间窗口或字符数批量 flush。
* 用于 Telegram editMessage / 飞书流式卡片更新。
*/
export type FlushCallback = (text: string, isComplete: boolean) => void | Promise<void>
const DEFAULT_INTERVAL_MS = 500
const DEFAULT_CHAR_THRESHOLD = 200
export class MessageBuffer {
private buffer = ''
private timer: ReturnType<typeof setTimeout> | null = null
private flushing = false
private pendingComplete = false
private activeFlush: Promise<void> | null = null
constructor(
private onFlush: FlushCallback,
private intervalMs = DEFAULT_INTERVAL_MS,
private charThreshold = DEFAULT_CHAR_THRESHOLD,
) {}
/** Append text delta. Triggers flush if threshold reached. */
append(text: string): void {
this.buffer += text
if (this.buffer.length >= this.charThreshold) {
this.scheduleFlush()
} else if (!this.timer) {
this.timer = setTimeout(() => this.flush(false), this.intervalMs)
}
}
/** Immediately flush all remaining content (called on message_complete). */
async complete(): Promise<void> {
if (this.timer) {
clearTimeout(this.timer)
this.timer = null
}
if (this.flushing) {
// A flush is in-flight; mark pending so it fires after current flush finishes
this.pendingComplete = true
await this.activeFlush
return
}
await this.flush(true)
}
/** Reset the buffer for a new message. */
reset(): void {
this.buffer = ''
this.pendingComplete = false
if (this.timer) {
clearTimeout(this.timer)
this.timer = null
}
}
private scheduleFlush(): void {
if (this.timer) {
clearTimeout(this.timer)
this.timer = null
}
queueMicrotask(() => this.flush(false))
}
private async flush(isComplete: boolean): Promise<void> {
if (this.timer) {
clearTimeout(this.timer)
this.timer = null
}
if (this.flushing) {
await this.activeFlush
return
}
if (this.buffer.length === 0) return
this.flushing = true
const text = this.buffer
this.buffer = ''
this.activeFlush = (async () => {
try {
await this.onFlush(text, isComplete)
} catch (err) {
console.error('[MessageBuffer] Flush error:', err)
} finally {
this.flushing = false
this.activeFlush = null
// If complete() was called while we were flushing, do the final flush now.
if (this.pendingComplete) {
this.pendingComplete = false
await this.flush(true)
}
}
})()
await this.activeFlush
}
}
|