62 lines
2.7 KiB
TypeScript
62 lines
2.7 KiB
TypeScript
/** Shared, bounded queue; unused requests are removed or aborted. */
|
|
export function createResourceLoader<T>(load: (key: string, signal: AbortSignal) => Promise<T>, concurrency = 6, cacheSize = 16) {
|
|
type Subscriber = { resolve: (value: T) => void; reject: (error: unknown) => void; cleanup: () => void };
|
|
type Job = { key: string; controller: AbortController; subscribers: Set<Subscriber> };
|
|
const cache = new Map<string, T>();
|
|
const jobs = new Map<string, Job>();
|
|
const queue: Job[] = [];
|
|
let active = 0;
|
|
function peek(key: string) {
|
|
const value = cache.get(key);
|
|
if (value !== undefined) { cache.delete(key); cache.set(key, value); }
|
|
return value;
|
|
}
|
|
function pump() {
|
|
while (active < concurrency && queue.length) {
|
|
const job = queue.shift()!;
|
|
if (!job.subscribers.size) continue;
|
|
active++;
|
|
Promise.resolve().then(() => {
|
|
if (job.controller.signal.aborted) throw new DOMException("Aborted", "AbortError");
|
|
return load(job.key, job.controller.signal);
|
|
}).then((value) => {
|
|
if (!job.controller.signal.aborted) {
|
|
cache.set(job.key, value);
|
|
while (cache.size > cacheSize) cache.delete(cache.keys().next().value!);
|
|
for (const subscriber of job.subscribers) { subscriber.cleanup(); subscriber.resolve(value); }
|
|
}
|
|
}, (error) => {
|
|
for (const subscriber of job.subscribers) { subscriber.cleanup(); subscriber.reject(error); }
|
|
}).finally(() => {
|
|
if (jobs.get(job.key) === job) jobs.delete(job.key);
|
|
active--; pump();
|
|
});
|
|
}
|
|
}
|
|
function request(key: string, signal?: AbortSignal): Promise<T> {
|
|
if (signal?.aborted) return Promise.reject(new DOMException("Aborted", "AbortError"));
|
|
const cached = peek(key);
|
|
if (cached !== undefined) return Promise.resolve(cached);
|
|
let job = jobs.get(key);
|
|
if (!job) { job = { key, controller: new AbortController(), subscribers: new Set() }; jobs.set(key, job); queue.push(job); }
|
|
const current = job;
|
|
return new Promise<T>((resolve, reject) => {
|
|
const abort = () => {
|
|
current.subscribers.delete(subscriber); subscriber.cleanup();
|
|
reject(new DOMException("Aborted", "AbortError"));
|
|
if (!current.subscribers.size) {
|
|
if (jobs.get(key) === current) jobs.delete(key);
|
|
const index = queue.indexOf(current);
|
|
if (index >= 0) queue.splice(index, 1);
|
|
current.controller.abort();
|
|
}
|
|
};
|
|
const subscriber: Subscriber = { resolve, reject, cleanup: () => signal?.removeEventListener("abort", abort) };
|
|
current.subscribers.add(subscriber);
|
|
signal?.addEventListener("abort", abort, { once: true });
|
|
pump();
|
|
});
|
|
}
|
|
return { request, peek };
|
|
}
|