Throttle number of observables

throttleAsync.ts

import { Observable, Subject, Subscription } from 'rxjs/Rx';

export type FuncAsync<T> = () => Observable<T>;


type QueuedExecution = () => Subscription;


export class ThrottleAsync<T> {
    private queue: QueuedExecution[];

    constructor (private max: number) {
    }

    execute(fnAsync: FuncAsync<T>): Observable<T> {
        if (this.queue.length >= this.max ) {
            const proxy = new Subject<T>();
            this.queue.push(() => fnAsync().subscribe(proxy));
            return proxy;
        }

        return new Observable(observer => {
            const sub = fnAsync().subscribe(
                value => observer.next(value),
                (error: any) => {
                    observer.error(error);
                    if (this.queue.length > 0) {
                        const nextFn = this.queue.shift();
                        nextFn();
                    }
                },
                () => {
                    observer.complete();
                    if (this.queue.length > 0) {
                        const nextFn = this.queue.shift();
                        nextFn();
                    }
                }
            );

            return () => sub.unsubscribe();
        });
    }
};