bufferConcatMap.ts

export function bufferConcatMap<T, R>(project: (val: T[]) => Observable<R>) {
    return (sourceObservable: Observable<T>) => {
        // this function will be called each time this Observable is subscribed to.

        // initialize the state - these variables will hold the state for every 
        // subscripition of the returned Observable
        let bufferedNotifications = [] as T[];
        let processing = false;

        // buld the Observable returned by this operator
        return sourceObservable.pipe(
            tap((val) => {
                // every value notified by the source is stored in the buffer
                bufferedNotifications.push(val);
            }),
            concatMap(() => {
                // if processing or if there are no items in the buffer just complete 
                // without notifying anything
                if (processing || bufferedNotifications.length === 0) {
                    return EMPTY;
                }
                // create a copy of the buffer to be passed to the project function 
                // so that the bufferedNotifications array can be safely reset
                const _buffer = [...bufferedNotifications];
                // update the state: processing starts and the bufferedNotifications 
                // needs to be reset
                processing = true;
                bufferedNotifications = [];
                // return the result of the project function invoked 
                // with the buffer of values stored
                return project(_buffer).pipe(
                    tap({
                        // when the Observable returned by the project function completes
                        // it means that there is no processing on the fly 
                        complete: () => {
                            processing = false;
                        },
                    }),
                );
            }),
        );
    };
}