35 lines
1.7 KiB
TypeScript
35 lines
1.7 KiB
TypeScript
import type { Cursor, Event, Task, Timestamp } from "../../protocol/model";
|
|
import type { ActivityAdapter, ActivityScope, ActivitySubscription, ResumeGap, StreamPhase } from "./types";
|
|
|
|
export class MemoryActivityAdapter implements ActivityAdapter {
|
|
readonly #events: readonly Event[];
|
|
readonly #tasks = new Map<string, Task>();
|
|
|
|
constructor(events: readonly Event[] = [], tasks: readonly Task[] = []) {
|
|
this.#events = events;
|
|
tasks.forEach((task) => this.#tasks.set(task.task_id, task));
|
|
}
|
|
|
|
subscribe(scope: ActivityScope, cursor: Cursor | undefined, onEvent: (event: Event) => void, onPhase: (phase: StreamPhase, gap?: ResumeGap) => void): ActivitySubscription {
|
|
let closed = false;
|
|
const scoped = this.#events.filter((event) => event.context.server_id === scope.server_id && event.context.project_id === scope.project_id);
|
|
const cursorIndex = cursor ? scoped.findIndex((event) => event.cursor === cursor) : -1;
|
|
const gap = cursor && cursorIndex < 0 ? { requested_cursor: cursor, resumed_cursor: scoped[0]?.cursor, message: "事件游标已失效，期间事件可能缺失。" } : undefined;
|
|
queueMicrotask(() => {
|
|
if (closed) return;
|
|
onPhase("streaming", gap);
|
|
scoped.slice(cursorIndex + 1).forEach((event) => { if (!closed) onEvent(event); });
|
|
});
|
|
return { close: () => { closed = true; } };
|
|
}
|
|
|
|
async cancelTask(task: Task): Promise<Task> {
|
|
if (!task.cancellable || ["SUCCEEDED", "FAILED", "CANCELLED"].includes(task.state)) {
|
|
throw new Error("该任务当前不可取消");
|
|
}
|
|
const cancelled = { ...task, state: "CANCELLED" as const, cancellable: false, updated_at: new Date().toISOString() as Timestamp };
|
|
this.#tasks.set(task.task_id, cancelled);
|
|
return cancelled;
|
|
}
|
|
}
|