47 lines
1.2 KiB
TypeScript
47 lines
1.2 KiB
TypeScript
/*
|
|
* SPDX-FileCopyrightText: syuilo and misskey-project
|
|
* SPDX-License-Identifier: AGPL-3.0-only
|
|
*/
|
|
|
|
import type { TimeService, TimerHandle } from '@/global/TimeService.js';
|
|
|
|
type Job<V> = {
|
|
value: V;
|
|
timer: TimerHandle;
|
|
};
|
|
|
|
// TODO: redis使えるようにする
|
|
export class CollapsedQueue<K, V> {
|
|
private jobs: Map<K, Job<V>> = new Map();
|
|
|
|
constructor(
|
|
protected readonly timeService: TimeService,
|
|
private timeout: number,
|
|
private collapse: (oldValue: V, newValue: V) => V,
|
|
private perform: (key: K, value: V) => Promise<void>,
|
|
) {}
|
|
|
|
enqueue(key: K, value: V) {
|
|
if (this.jobs.has(key)) {
|
|
const old = this.jobs.get(key)!;
|
|
const merged = this.collapse(old.value, value);
|
|
this.jobs.set(key, { ...old, value: merged });
|
|
} else {
|
|
const timer = this.timeService.startTimer(() => {
|
|
const job = this.jobs.get(key)!;
|
|
this.jobs.delete(key);
|
|
this.perform(key, job.value);
|
|
}, this.timeout);
|
|
this.jobs.set(key, { value, timer });
|
|
}
|
|
}
|
|
|
|
async performAllNow() {
|
|
const entries = [...this.jobs.entries()];
|
|
this.jobs.clear();
|
|
for (const [_key, job] of entries) {
|
|
this.timeService.stopTimer(job.timer);
|
|
}
|
|
await Promise.allSettled(entries.map(([key, job]) => this.perform(key, job.value)));
|
|
}
|
|
}
|