counted_subject.ts

import {Subject} from 'rxjs';

interface CountedSubscription {
    unsubscribe: () => void
}

export type CountParams = {
    newCount: number,
    prevCount: number
};

export class CountedSubject<T = any> {

    private count: number = 0;
    private sub: Subject<T> = new Subject<T>();

    public readonly onCount: Subject<CountParams> = new Subject<CountParams>();

    public subscribe(next?: (value: T) => void, error?: (error: any) => void, complete?: () => void): CountedSubscription {
        const s = this.sub.subscribe(next, error, complete);
        this.addCount(1);
        return {
            unsubscribe: () => {
                s.unsubscribe();
                this.addCount(-1);
            }
        };
    }

    public next(value?: T) {
        this.sub.next(value);
    }

    private addCount(add: number) {
        const prevCount = this.count;
        const newCount = prevCount + add;
        this.count = newCount;
        this.onCount.next({newCount, prevCount});
    }

}