Proposal: new operator exhaustMap with buffer

[Etienne-M]

This is an improvement proposal based on a use-case I am often facing and can't easily address with the current operators.

Description
It's the same thing as exhaustMap, except that it replays the last missed event on completion of the inner observable. Or concatMap, but with a buffer size of 1.

Context
One of the basic use of RxJS that I have in my projects is the following: there is some state in my application that I am observing and for which I want to perform some side effects. In other words, my app is a big state machine and depending on its state, I want to do some actions.

How do you handle backpressure in those cases? mergeMap does not make sense, switchMap would trigger way more operations than necessary, concatMap would exacerbate the problem. The one that makes the most sense is exhaustMap, but there is a problem: you end up being late on the state machine. What I am suggesting is an alternate operator that would replay the last event upon completion of the inner observable. Translated for this use-case, it would wait for the async operation to finish, then make sure to handle the most up-to-date state.

Implementation
I have everything already coded, I would love to contribute it back since I think it might be a use case other people have.

[kwonoj]

Moved to the discussions with specific category.

[raymonddavis]

Does https://rxjs-dev.firebaseapp.com/api/operators/takeLast solve your problem if it is applied on the inner observable?

[Etienne-M]

No, not really. The inner observable is left untouched. It's the backpressure behavior that changes

[raymonddavis]

I get what you mean now. Sorry I misread.

[backbone87]

this should be similar to the removed inspectTime, but instead of a fixed timeout you want the inner observable to take the decision when to resume.

so something like this:

function inspectMap<T, R>(project: (value: T, index: number) => ObservableInput<R>): OperatorFunction<T, R> {
  return (source$) => {
    const subject$ = new BehaviorSubject(true);

    return source$.pipe(
      audit(() => subject$.pipe(filter(Boolean))),
      tap(() => void subject$.next(false)),
      map(project),
      map(from),
      mergeMap(tap({ complete: () => void subject$.next(true) })),
    );
  };
}

[Etienne-M]

Wow, this is much simpler [shorter] than what I had written 😅

[Etienne-M]

I don't know how the authors of the library feel in those cases. Looking at the code you've shared, it's not at all obvious what's happening, but it is technically feasible in a very short number of lines.