RxJS Mastery – #55 exhaustAll

The RxJS exhaustAll operator flattens higher-order Observable. While doing so, it drops values as long as an inner Observable has not completed yet.

RxJS exhaustAll drops some values

In the example below we would like to output a, b, c, d in an inner Observables twice. So, a normal result Observable would look like aabbccdd. Let’s see what happens when we apply exhaustAll.

it('drops some values', () => {
    createTestScheduler().run((helpers) => {
        const { expectObservable } = helpers;

        const indexedValues = ['a', 'b', 'c', 'd'];

        const source$ = interval(100).pipe(
            take(4),
            map((outerV) => interval(100).pipe(
                map(_ => indexedValues[outerV]),
                take(2)),
            )
        );
        const result$ = source$.pipe(exhaustAll());

        expectObservable(result$).toBe(
            '200ms a 99ms a 199ms d 99ms (d|)',
            { a: 'a', d: 'd' }
        );
    });
});

Thanks to exhaustAll, the values b and c are not part of the result Observable. This is because the inner Observables for b and c run concurrently with other inner Observables. The 4th inner Observable to output d is running alone and therefore its values make it into the final output.

Would we replace exhaustAll with mergeAll the output would emit the values concurrently:

expectObservable(result$).toBe(
    '200ms a 99ms (ab) 96ms (bc) 96ms (cd) 96ms (d|)',
    { a: 'a', b: 'b', c: 'c', d: 'd' }
);

Note: marble groups take some milliseconds away due to the parentheses. That’s why 96ms is used as time between.

Exercise for the RxJS exhaustAll operator 💪

Use exhaustAll to not overload a service. In the code below you find a callService method that simulates a service cal taking 99ms. The source$ Observable is basically the input from a user interface triggering service calls with various increases of a value. Make sure that only one service request runs at a time.

const callService = (increase): Observable<string> => cold('99ms a|', { a: 'increased by ' + increase });

const source$ = cold('5 50ms 2 30ms 3 50ms 8|').pipe(
    map(v => callService(v)),
);
const result$ = source$ // your part;

expectObservable(result$).toBe(
    '99ms a 132ms d|',
    { a: 'increased by 5', d: 'increased by 8' }
);

This post is part of the RxJS mastery series. As always the code examples can be found on GitHub.