Avoid @ngrx/effects to cancel same actions

Question

Is there a way to avoid effects to cancel previous one ?

I need to do:

this.tagsStoreService.initTagsForFamilyId(this.tagFamily.id)

I have this effects:

@Effect() initTagsForFamilyId$: Observable<Action> = this.actions$
    .pipe(
        ofType<InitTagsForFamilyIdAction>(TagsStateActionTypes.INIT_TAGS_FOR_FAMILY_ID_ACTION),
        switchMap(params => {
            // this.loadingService.showLoading();
            return this.tagsService.initTagsForFamilyId(params.payload)
                .pipe(
                    exhaustMap((data) => {
                        this.loadingService.hideLoading();

                        return [
                            new InitTagsForFamilyIdSuccessAction({ ...data }),
                        ];
                    }),
                    catchError(error => {
                        // this.loadingService.hideLoading();
                        return of(new TagsFailureAction({ error }));
                    }),
                );
        }),
    );

Thanks in advance for help ;-)

Answer 1

You should use mergeMap instead switchMap

let {
  interval,
  of
} = rxjs
let {
take,
tap,
switchMap,
delay,
finalize
} = rxjs.operators

const switchMapCase = interval(500).pipe(
  take(5),
  tap(() => console.log("new request")),
  switchMap(() =>
    of(null).pipe(
      delay(800),
      tap(() => console.log("end")),
      finalize(() => console.log("request stream complete"))
    )
  )
);

switchMapCase.subscribe();

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.6.6/rxjs.umd.min.js"></script>

If you run the snippet above (the one with switchMap) you will notice that every 500 ms a new event is emitted (new request), this event will go inside the switchMap and return a new observable that will be resolved after 800 ms , because 800 > 500 before the inner observable is resolved new observable will come from the source stream, which will cancel the current inner stream (the one with 800 ms delay) and a new stream that also has 800 ms delay will be returned. This thing will happen 5 times (because of the take(5) operator, this is equivalent to dispatching the TagsStateActionTypes.INIT_TAGS_FOR_FAMILY_ID_ACTION action 5 times, with 500 ms delay between each dispathc). In the 5-th iteration the inner observable will complete, as there will be no new dispatches that will cancel it.

let {
  interval,
  of
} = rxjs
let {
take,
tap,
mergeMap,
delay,
finalize
} = rxjs.operators

const mergeMapCase = interval(500).pipe(
  take(5),
  tap(() => console.log("new request")),
  mergeMap(() =>
    of(null).pipe(
      delay(800),
      tap(() => console.log("server response recieved")),
      finalize(() => console.log("request stream complete"))
    )
  )
);

mergeMapCase.subscribe();

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.6.6/rxjs.umd.min.js"></script>

In the snippet above (the mergeMap) we have pretty much we have the same structure with the only difference that we use mergeMap instead switchMap the difference will be that unlike switchMap (that cancels the previous stream) mergeMap will add the new "innerStream" to the source observable without canceling the old ones.

*Side Note: * The thing that actually is getting canceled in the effects is not the action or the effect it is the this.tagsService.initTagsForFamilyId observable, that on completion will trigger another another action.