104 lines
3.4 KiB
TypeScript
104 lines
3.4 KiB
TypeScript
import { randomUUID } from 'node:crypto';
|
|
|
|
export interface PiProjectWriteLease {
|
|
id: string;
|
|
projectId: string;
|
|
holderId: string;
|
|
release(): void;
|
|
}
|
|
|
|
interface WaitingLease {
|
|
projectId: string;
|
|
holderId: string;
|
|
signal?: AbortSignal;
|
|
resolve(lease: PiProjectWriteLease): void;
|
|
reject(error: Error): void;
|
|
onAbort?: () => void;
|
|
}
|
|
|
|
export class PiProjectWriteLeaseCoordinator {
|
|
private readonly active = new Map<string, PiProjectWriteLease>();
|
|
private readonly waiting = new Map<string, WaitingLease[]>();
|
|
|
|
get activeCount(): number { return this.active.size; }
|
|
|
|
waitingCount(projectId?: string): number {
|
|
if (projectId) return this.waiting.get(projectId)?.length ?? 0;
|
|
return [...this.waiting.values()].reduce((total, queue) => total + queue.length, 0);
|
|
}
|
|
|
|
acquire(projectId: string, holderId: string, signal?: AbortSignal): Promise<PiProjectWriteLease> {
|
|
if (!projectId.trim() || !holderId.trim()) throw new Error('Project and lease holder are required');
|
|
if (signal?.aborted) return Promise.reject(new Error('Project write lease wait cancelled'));
|
|
if (!this.active.has(projectId)) return Promise.resolve(this.issue(projectId, holderId));
|
|
|
|
return new Promise<PiProjectWriteLease>((resolve, reject) => {
|
|
const waiter: WaitingLease = { projectId, holderId, resolve, reject, ...(signal ? { signal } : {}) };
|
|
if (signal) {
|
|
waiter.onAbort = () => {
|
|
this.removeWaiter(waiter);
|
|
reject(new Error('Project write lease wait cancelled'));
|
|
};
|
|
signal.addEventListener('abort', waiter.onAbort, { once: true });
|
|
}
|
|
const queue = this.waiting.get(projectId) ?? [];
|
|
queue.push(waiter);
|
|
this.waiting.set(projectId, queue);
|
|
});
|
|
}
|
|
|
|
cancelProject(projectId: string): void {
|
|
for (const waiter of this.waiting.get(projectId) ?? []) {
|
|
this.detachAbort(waiter);
|
|
waiter.reject(new Error('Project write lease wait cancelled'));
|
|
}
|
|
this.waiting.delete(projectId);
|
|
this.active.get(projectId)?.release();
|
|
}
|
|
|
|
private issue(projectId: string, holderId: string): PiProjectWriteLease {
|
|
let released = false;
|
|
const lease: PiProjectWriteLease = {
|
|
id: randomUUID(),
|
|
projectId,
|
|
holderId,
|
|
release: () => {
|
|
if (released) return;
|
|
released = true;
|
|
if (this.active.get(projectId) === lease) this.active.delete(projectId);
|
|
this.advance(projectId);
|
|
},
|
|
};
|
|
this.active.set(projectId, lease);
|
|
return lease;
|
|
}
|
|
|
|
private advance(projectId: string): void {
|
|
const queue = this.waiting.get(projectId);
|
|
while (!this.active.has(projectId) && queue?.length) {
|
|
const waiter = queue.shift() as WaitingLease;
|
|
this.detachAbort(waiter);
|
|
if (waiter.signal?.aborted) {
|
|
waiter.reject(new Error('Project write lease wait cancelled'));
|
|
continue;
|
|
}
|
|
waiter.resolve(this.issue(projectId, waiter.holderId));
|
|
}
|
|
if (!queue?.length) this.waiting.delete(projectId);
|
|
}
|
|
|
|
private removeWaiter(waiter: WaitingLease): void {
|
|
const queue = this.waiting.get(waiter.projectId);
|
|
const index = queue?.indexOf(waiter) ?? -1;
|
|
if (index >= 0) queue?.splice(index, 1);
|
|
if (!queue?.length) this.waiting.delete(waiter.projectId);
|
|
this.detachAbort(waiter);
|
|
}
|
|
|
|
private detachAbort(waiter: WaitingLease): void {
|
|
if (waiter.signal && waiter.onAbort) {
|
|
waiter.signal.removeEventListener('abort', waiter.onAbort);
|
|
}
|
|
}
|
|
}
|