RFC: Change behavior of ofType in Effects to be more consistent when combining observables

[brandonroberts]

There are two predominant things you do when listening to an action after using the ofType operator:

actions.pipe(
  ofType('SOME_ACTION')
  someMap(action => doSomething(action))
)

and

actions.pipe(
  ofType('SOME_ACTION'),
  withLatestFrom(store.select(someThing)),
  someMap(([action, latest]) => doSomething(action, latest))
)

These have two different behaviors as far as effects are concerned. The first one doesn't do anything until the action is dispatched, which is what it should do.

The second one immediately subscribes to the withLatestFrom whether the action is dispatched yet or not. If that state used by the selector is not initialized yet, you could get an error you're not expecting. In order to make it "lazy" you end up having to nest it within a flattening observable operator.

actions.pipe(
  ofType('SOME_ACTION'),
  someMap(action =>
    of(action).pipe(
      withLatestFrom(store.select(someThing)),
      someMap(([action, latest]) => doSomething(action, latest))
    )
)

We document this here: https://ngrx.io/guide/effects#incorporating-state

I think the behavior should be the same in either case, where the selector should not be triggered until the action is dispatched without having to nest it inside another flattening observable operator.

One idea would be to add a .pipe after the filter inside the ofType operator to map it to a new observable of the action to mimic the second scenario. There are probably other options also, and I haven't measured the impact of this change yet either.

[AdditionAddict]

I referenced this in a stackoverflow answer as I personally found the above explanation better than the docs.

With regards to the RFC, I think this is needed. Whilst you would guess ofType is a filter the behaviour above is still surprising and not what you would expect and try first.

[jqsjqs]

I agree. The official docs weren't clear on this and needs this type of explanation. Good job @brandonroberts (and thanks).

[sod]

We wrote a custom operator for that, that boils down to (typescript 4 for variadic tuple types required):

export function mergeTakeOne<SOURCE, MERGE extends ObservableInput<any>[]>(
    ...toBeMerged$: [...MERGE]
): OperatorFunction<SOURCE, [SOURCE, ...{[P in keyof MERGE]: MERGE[P] extends ObservableInput<infer U> ? U : never}]> {
    return mergeMap((source) =>
        combineLatest(toBeMerged$).pipe(
            take(1),
            map((source2): any => [source, ...source2]),
        ),
    );
}

used just like withLatestFrom, only that it subscribes once it's needed.

[ngfelixl]

I am sorry that I did not found this discussion and opened #2703.

In my opinion it would be really nice to have something like a withLatestFromStore operator provided by the effects package that combines the concatMap and the withLatestFrom. I've seen and I see many people just using withLatestFrom. For myself, I use NgRX almost every day and only knew about the performance issues about a month ago or so. Having an operator in place that prevents the devs from having to use the nested observables would be very beneficial. Therefore it would be great to have a withLatestFromStore operator in place that is usable like

createEffect(() => this.actions$.pipe(
  ofType(getSelectedBookDetails),
  withLatestFromStore(this.store.select(getSelectedBookId)),
  switchMap(([action, selectedBookId]) => this.bookService.getDetails(selectedBookId).pipe(
    map(...),
    catchError(...)
  )
);

or without having the store explicitly injected

createEffect(() => this.actions$.pipe(
  ofType(getSelectedBookDetails),
  withLatestFromStore(getSelectedBookId),
  switchMap(([action, selectedBookId]) => this.bookService.getDetails(selectedBookId).pipe(
    map(...),
    catchError(...)
  )
);

If the operator name contains the withLatestFrom string could help the dev understand how to use it. He or she can assume that it is similar to the corresponding RxJS operator. At least the first thing I would try as an output would be the array like function parameter [action, selectedBookId] if I am not familiar with it.

[david-shortman]

@ngfelixl The key distinction to note is that withLatestFromStore, as an operator, will not be able to reference an injectable Store, so it can't internally do this.store.select on each provided selector.

So to delay evaluation, you'd have to have some operator like delayedWithLatestFrom as such:

createEffect(() => this.actions$.pipe(
  ofType(getSelectedBookDetails),
  delayedWithLatestFrom(() => [this.store.select(getSelectedBookId)]),
  switchMap(([action, selectedBookId]) => this.bookService.getDetails(selectedBookId).pipe(
    map(...),
    catchError(...)
  )
);

[david-shortman]

At that point, I wonder if the use case would extend far enough to be included in RxJS instead. I'm sure there are other instances where it is desirable to delay evaluation of the params to an operator like withLatestFrom.

[david-shortman]

Created an issue in RxJS land to see where such an operator would better belong ReactiveX/rxjs#5845

[david-shortman]

Here's the operator I ended up writing (requires TS 4.0+). Gonna figure out how to open a PR:

import { Observable, of, OperatorFunction } from 'rxjs';
import { concatMap, withLatestFrom } from 'rxjs/operators';

type TypeOfObservable<T> = T extends Observable<infer U> ? U : never;

/*
 * RxJS operator that lazily evaluates the input Observables.
 *
 * Useful when there is a performance cost for evaluating a function which produces
 * an input observable, like in the case of NgRx's `store.select` method.
 */
export const concatLatestFrom = <T extends Observable<unknown>[], V>(o: () => [...T]) =>
  concatMap(value => of(value).pipe(withLatestFrom(...o()))) as OperatorFunction<
    V,
    [V, ...{ [i in keyof T]: TypeOfObservable<T[i]> }]
  >;

Jest proof:

import { of } from 'rxjs';
import { concatLatestFrom } from './concat-latest-from.operator';
import { skipWhile, tap, withLatestFrom } from 'rxjs/operators';
import { hot } from 'jasmine-marbles';

describe('concatLatestFrom', () => {
  it('should evaluate the array of observables only when the source emits a value', () => {
    // given
    let evaluated = false;
    const toBeLazilyEvaluated = () => {
      evaluated = true;
      return of(4);
    };

    // when
    const numbers$ = hot('abc', { a: 1, b: 2, c: 3 }).pipe(
      skipWhile(number => number < 3),
      tap(() => expect(evaluated).toBe(false)),
      concatLatestFrom(() => [toBeLazilyEvaluated()])
    );

    // then
    expect(evaluated).toBe(false);
    expect(numbers$).toBeObservable(hot('--d', { d: [3, 4] }));
    expect(evaluated).toBe(true);
  });
});

[jeetparikh]

Another question around this is once the withLatestFrom subscribes it's always listening. If a pipe and put a .tap to it I can see it logging all the time when that particular state slice changes. What I am interested in just the latest value of that state slice when THAT action triggers. And this is what I had originally intended too. Isn't this a performance hit if it's always listening? Any tips @brandonroberts or anyone . Thanks

[david-shortman]

Another question around this is once the withLatestFrom subscribes it's always listening. If a pipe and put a .tap to it I can see it logging all the time when that particular state slice changes. What I am interested in just the latest value of that state slice when THAT action triggers. And this is what I had originally intended too. Isn't this a performance hit if it's always listening? Any tips @brandonroberts or anyone . Thanks

Answered this question in the discussion in this PR #2799 (comment)

[niklas-wortmann]

just adding my 2 cents to this discussion. I usually just do

actions.pipe(
  ofType('SOME_ACTION'),
  withLatestFrom(store.select(someThing)),
  someMap(([action, latest]) => doSomething(action, latest))
)

My problem with having the withLatest in someMap operator is that it will create a new subscription to that observable on every emission. This works fine with ngrx selects which emits synchronously on subscribe, but this won't behave as expected if the observable emits asynchronous. The problem is actually displayed in the marble diagram of rxjs

If you would pass an observable that emits asynchronously by wrapping it in someMap you will likely lose the triggered action. Here you see that the a emission is lost.
This might be just one of my concerns, so ignore me if this does not apply :D

[david-shortman]

@niklas-wortmann You are right that if you delay the invitation of withLatestFrom, some Observables will not have a value readily available like is the case with store.select.

This is another good reason why the PR for concatLatestFrom is against NgRx and not proposed as an RxJs operator- it fits this use case well, but could get people into trouble in other scenarios if they don't understand the mechanics of their Observables and the operator.