PromiseRateLimiter

export class PromiseRateLimiter<T = any> {

    private callbacks: (() => Promise<T>)[] = [];
    private callbackResults: T[] = [];

    private nextIndex: number = 0;
    private endPromise: Promise<any> = null;
    private launched: boolean = false;
    private aborted: boolean = false;
    private resolveCallback: (value?: (PromiseLike<T[]> | T[])) => void;
    private resolvedPromises: number = 0;
    private rejectCallback: (reason?: any) => void;
    private startedPromises: number = 0;

    public constructor(private maxConcurrentPromises: number) {}

    public addCallback(callback: () => Promise<T>) {
        if (this.launched) {
            throw new Error("Cannot add callback after launching rate limiter");
        }
        this.callbacks.push(callback);
    }

    public launch(): Promise<T[]> {
        this.launched = true;

        this.endPromise = new Promise<T[]>((resolve, reject) => {
            this.resolveCallback = resolve;
            this.rejectCallback = reject;

            if (this.callbacks.length === 0) {
                setTimeout(() => this.aborted ? reject([]) : resolve([]), 0);
            }
        });

        let min = this.maxConcurrentPromises;
        let length = this.callbacks.length;
        if (min > length) {
            min = length;
        }

        for (let i = 0; i < min; ++i) {
            this.startNext();
        }

        return this.endPromise;
    }

    public abort() {
        this.aborted = true;
        this.rejectCallback(this.callbackResults);
    }

    private startNext() {
        if (this.aborted) {
            return;
        }
        ++this.startedPromises;
        let index = this.nextIndex;
        let callback = this.callbacks[this.nextIndex++];
        let promise = callback();
        promise.then(result => {
            if (this.aborted) {
                return;
            }
            this.callbackResults[index] = result;
            ++this.resolvedPromises;
            if (this.resolvedPromises === this.callbacks.length) {
                this.resolveCallback(this.callbackResults);
            } else if (this.startedPromises < this.callbacks.length) {
                this.startNext();
            }
        });
    }

    public getEndPromise(): Promise<T[]> {
        return this.endPromise;
    }
}