47 lines
1.7 KiB
TypeScript
47 lines
1.7 KiB
TypeScript
/**
|
||
* 会话内事件总线（A1.3）。
|
||
*
|
||
* 职责：给事件分配会话内单调递增的 seq、向订阅者广播、支持从某个 seq 之后回放（断点续传）。
|
||
* A3.7 会话落库后，回放改为读存储，这里只保留实时广播；内存缓冲仅作为 A1 的过渡实现。
|
||
*/
|
||
import { PROTOCOL_VERSION, type PlatformEvent } from '@aps/contracts';
|
||
|
||
type DistributiveOmit<T, K extends PropertyKey> = T extends unknown ? Omit<T, K> : never;
|
||
|
||
/** 发布事件的入参：v 与 seq 由总线分配，ts 可由调用方指定 */
|
||
export type EventInput = DistributiveOmit<PlatformEvent, 'v' | 'seq' | 'ts'> & { ts?: number };
|
||
|
||
const REPLAY_BUFFER = 500;
|
||
|
||
export class EventBus {
|
||
private seq = 0;
|
||
private readonly buffer: PlatformEvent[] = [];
|
||
private readonly listeners = new Set<(event: PlatformEvent) => void>();
|
||
|
||
publish(input: EventInput): PlatformEvent {
|
||
this.seq += 1;
|
||
const event = { ...input, v: PROTOCOL_VERSION, seq: this.seq, ts: input.ts ?? Date.now() } as PlatformEvent;
|
||
this.buffer.push(event);
|
||
if (this.buffer.length > REPLAY_BUFFER) this.buffer.shift();
|
||
for (const listener of this.listeners) listener(event);
|
||
return event;
|
||
}
|
||
|
||
/** 从 since（不含）之后回放 */
|
||
replay(since = 0): PlatformEvent[] {
|
||
return this.buffer.filter((event) => event.seq > since);
|
||
}
|
||
|
||
/** 先回放再订阅：避免回放与订阅之间漏事件 */
|
||
subscribe(since: number, listener: (event: PlatformEvent) => void): () => void {
|
||
for (const event of this.replay(since)) listener(event);
|
||
this.listeners.add(listener);
|
||
return () => {
|
||
this.listeners.delete(listener);
|
||
};
|
||
}
|
||
|
||
get lastSeq(): number {
|
||
return this.seq;
|
||
}
|
||
} |