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;
|
|||
|
|
}
|
|||
|
|
}
|