ngrx effects error handling

问题:

I have a very basic question concerning @ngrx effects: How to ignore an error that happens during the execution of an effect such that it doesn't affect future effect execution?

My situation is as follows: I have an action (LOGIN) and an effect listening to that action. If an error happens inside this effect, I want to ignore it. When LOGIN is dispatched a second time after this error, the effect should be executed a second time.

My first attempt to do this was:

  @Effect()
  login$ = this.actions$
    .ofType('LOGIN')
    .flatMap(async () => {
      console.debug('LOGIN');
      // throw an error
      let x = [];x[0]();
    })
    .catch(err => {
      console.error('Error at login', err);
      return Observable.empty();
    });

Dispatching LOGIN the first time throws and catches the error, as expected. However, if I dispatch LOGIN a second time afterwards, nothing happens; the effect is not executed.

Therefore I tried the following:

    .catch(err => {
      return this.login$;
    });

, but this results in an endless loop... Do you know how to catch the error without preventing effect execution afterwards?

回答1:

The ngrx infrastructure subscribes to the effect via the provider imported into the app's NgModule using EffectsModule.run.

When the observable errors and catch returns a empty obervable, the composed observable completes and its subscribers are unsubscribed - that's part of the Observable Contract. And that's the reason you see no further handling of LOGIN actions in your effect. The effect is unsubscribed on completion and the ngrx infrastructure does not re-subscribe.

Typically, you would have the error handling inside the flatMap (now called mergeMap):

import { Actions, Effect, toPayload } from "@ngrx/effects";

@Effect()
login$ = this.actions$
  .ofType('LOGIN')
  .map(toPayload)
  .flatMap(payload => Observable
    .from(Promise.reject('Boom!'))
    .catch(error => {
      console.error('Error at login', error);
      return Observable.empty();
    })
  });

The catch composed into the inner observable will see an empty observable flattened/merged into the effect, so no action will be emitted.

回答2:

The @Effect stream is completing when the error occurs, preventing any further actions.

The solution is to switch to a disposable stream. If an error occurs within the disposable stream it's okay, as the main @Effect stream always stays alive, and future actions continue to execute.

@Effect()
login$ = this.actions$
    .ofType('LOGIN')
    .switchMap(action => {

        // This is the disposable stream!
        // Errors can safely occur in here without killing the original stream

        return Rx.Observable.of(action)
            .map(action => {
                // Code here that throws an error
            })
            .catch(error => {
                // You could also return an 'Error' action here instead
                return Observable.empty();
            });

    });

More info on this technique in this blog post: The Quest for Meatballs: Continue RxJS Streams When Errors Occur

回答3:

this is how I handle the option to be notified or not notified on Observables that didn't find any data:

 public listenCampaignValueChanged(emitOnEmpty: boolean = false): Observable<CampaignsModelExt> {
        var campaignIdSelected$ = this.ngrxStore.select(store => store.appDb.uiState.campaign.campaignSelected)
        var campaigns$ = this.ngrxStore.select(store => store.msDatabase.sdk.table_campaigns);
        return campaignIdSelected$
            .combineLatest(campaigns$, (campaignId: number, campaigns: List<CampaignsModelExt>) => {
                return campaigns.find((i_campaign: CampaignsModelExt) => {
                    return i_campaign.getCampaignId() == campaignId;
                });
            }).flatMap(v => (v ? Observable.of(v) : ( emitOnEmpty ? Observable.of(v) : Observable.empty())));
    }