Custom RxJS operator for safe calling the service in the Effect

[alex-okrushko]

Many Effects are used to call services, and most of the time they should result in SUCCESS or ERROR actions.
From the code that I review it's quite often that error case is either:

• not handled
• handled incorrectly:
  • error handling is put onto main Observable, instead of piped from service call
  • error is handled, but without wrapping Error action into observable

Even people who are familiar with all of these mistakes still make them sometimes.

Custom operator might help here, e.g.:

const effect$: Observable<FooSuccess|FooError> = actions$.pipe(
  ofType<Foo>(FOO),
  concatMap(a => this.service.fetch(a.bar).pipe(
    map(resp => new FooSuccess(resp)),
    catchError(e => of(new FooError()))
  )),
);

could turn into

const effecOp$: Observable<FooSuccess|FooError> = actions$.pipe(
  ofType<Foo>(FOO),
  safeCall(
    action => this.service.fetch(a.bar).pipe(map(resp => new FooSuccess(resp))),
    error => new FooError()),
)

where the safeCall (not crazy about the naming) would be that safety operator.

the operator itself could be implemented like this:

function safeCall<A extends Action, T extends Action, V extends Action>(
  callFn: (action: A) => Observable<T>,
  errorHandlingFn: (error: any) => V
): (source: Observable<A>) => Observable<T | V> {
  return (source: Observable<A>) =>
    source.pipe(
      concatMap(a =>
        callFn(a).pipe(catchError(err => of(errorHandlingFn(err))))
      )
    );
}

Thoughts?
Should we add this to ngrx?

If accepted, I would be willing to submit a PR for this feature

[ x ] Yes (Assistance is provided if you need help submitting a pull request)
[ ] No

[alex-okrushko]

Ah, here is the stackblitz: https://stackblitz.com/edit/typescript-snyfkg

[timdeschryver]

I get the why, but I'm not really in favor of this.

• peoples would have to "learn" this new operator and it's not really that hard to do in RxJS
• not all service calls would want to use concatMap, yes we could make this a parameter

But maybe I'm just a sucker for the RxJS operators 😅

[brandonroberts]

@MikeRyanDev had a similar idea for a pipeable operator like this one. I think it has some value for repetitive mapping and error handling.

[MikeRyanDev]

Here's the proposal I submitted to redux-observable: redux-observable/redux-observable#457

They accepted it, I just haven't gotten around to implementing it in both locations. I welcome anyone to make the PR.

[alex-okrushko]

peoples would have to "learn" this new operator and it's not really that hard to do in RxJS

This won't be enforced onto people, it's just safer and helpful alternative :)

not all service calls would want to use concatMap, yes we could make this a parameter

That's true, but concatMap being the default choice is only making it safer against other mistakes, right?

Honestly, in the internal code reviews every time I see switchMap I ask to explain why it was used, and in most cases it's changed to concatMap.

But maybe I'm just a sucker for the RxJS operators 😅

I love RxJS operators :) but every time I create an "interesting" combination/solution, I comment every line of it for my future self.

@MikeRyanDev I can definitely create a PR for this.

• Do we agree that it should be part of NgRx? @timdeschryver waiting for your reconsideration as well.

• Should it be called mapToAction?

• Effects should not complete, so I assume just success and error callbacks are fine?

[timdeschryver]

@alex-okrushko Yea of course I'm not blocking.
I honestly think this will start growing on me, especially after seing it in action 😃

[MikeRyanDev]

Some inner observables in effects notify multiple times and then complete. Thinking of web socket connections and Http requests with status notifications.

…

On Thu, Aug 2, 2018 at 12:04 PM Tim Deschryver ***@***.***> wrote: @alex-okrushko <https://github.com/alex-okrushko> Yea of course I'm not blocking. I honestly think this will start growing on me, especially after seing it in action 😃 — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#1224 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/ADilFXRvSIfRy1iMH3PmDJHDV_LnxfoVks5uMyMbgaJpZM4VrhwB> .

[sangecz]

is it possible to implement custom operator which would look at action's payload and if there is operator (switchMap, concatMap,..) present use it otherwise use some chosen default op.?
It would be great for libs which generate store (eg. from swagger). They also generate effects and they could give opportunity to lib users to override default operator used for API calls.

[jnorkus]

Our app is heavily relying on effects for all the business logic. That's why a global error handler is critical because any exception stops the actions stream. Here is my solution.

effects.service.ts

import { Injectable, ErrorHandler } from '@angular/core';
import { getEffectsMetadata } from '@ngrx/effects';
import { Observable } from 'rxjs';
import { catchError } from 'rxjs/operators';

@Injectable()
export class EffectsService {
    constructor(
        private errorHandler: ErrorHandler
    ) {}

    addErrorHandler(instance) {
        const metadata = getEffectsMetadata(instance);
        for (let key in metadata) {
            instance[key] = (<Observable<any>>instance[key]).pipe(
                catchError((error, source) => {
                    this.errorHandler.handleError(error);
                    return source; // return actions observable so the stream is effectively continued
                })
            )
        }
    }
}

home.effects.ts

constructor(
    effects: EffectsService
) {
    effects.addErrorHandler(this);
}

This will handle any unhandled exceptions and continue the actions stream.

[dummdidumm]

Regarding the requirement "I want unhandled errors not to completely break my action stream": I wonder why this is not part of the effects library itself? Wouldn't a appended catchError to all @Effect()-streams work?

[jnorkus]

@dummdidumm actually there is a catchError in the effects library. It calls the error handler and then completes. In my sample I return the source parameter which allows the stream to continue instead of completing. I wonder if that would actually make sense to include in the lib.

[dummdidumm]

Maybe (if they don't want it to be the default) there could be a new property to turn it on.

@Effect({resubscribeAfterError: true})
streamThatWillContinueToWorkAfterError$ = ...

[alex-okrushko]

@dummdidumm

Wouldn't a appended catchError to all @Effect()-streams work

No, it wouldn't work. It needs to be handled on the Observable before it's flattened into the main stream.

[dummdidumm]

I imagine something like this
https://stackblitz.com/edit/rxjs-e1jvlk
Could @ngrx/effects not wrap all effects in such a way inside the effects_resolver before merging it back into the main stream? Is there anything I'm missing?

[jnorkus]

@alex-okrushko have you tried my solution? It works for me so it could work for others if implemented in effects lib. You only need to return the second argument of catchError and the stream continues.

[awerlang]

@jnorkus you're using a loose definition of "works" here.

catchError((error, source) => source) doesn't continue the original stream. Instead, it completes and restarts. In order to ignore the error (like ignoreElements, but for errors) and continue, pipe a catchError() to the origin stream like @alex-okrushko explained. Restarting the observable in combination with a few operators (distinctUntilChanged, concat, startWith, etc) would lead to unexpected results.

Your idea is neat though. As a default selector I'd prefer returning EMPTY from it, so only the effect in err would unsubscribe from the action stream.

[alex-okrushko]

Reviving this thread.
With mapToAction fn we'll have the following changes (using the pre-8.0 action creators):

BEFORE:

const effect$: Observable<FooSuccess|FooError> = actions$.pipe(
  ofType(FOO),
  concatMap(a => this.service.fetch(a.bar).pipe(
    map(resp => new FooSuccess(resp)),
    catchError(e => of(new FooError()))
  )),
);

AFTER:

const effect$: Observable<FooSuccess|FooError> = actions$.pipe(
  ofType(FOO),
  concatMap(a => this.service.fetch(a.bar).pipe(
    mapToAction(
      resp => new FooSuccess(resp),
      e => new FooError(),
      optionalComplete => new FooCompleted() // could be a good addition as Mike mentioned above?
   ))),
);

It's not as 'in your face' as I'd like it to be (given how often I see this error), but at least promoting this new operator might bring some awareness.

Alternatively, or in addition to it, we can have 4 operators that will wrap the flattening operators, concatMapToAction,switchMapToAction,exhaustMapToAction and mergeMapToAction e.g.:

const effecOp$: Observable<FooSuccess|FooError> = actions$.pipe(
  ofType(FOO),
  concatMapToAction( // <-- concatMap-based implementation
    action => this.service.fetch(a.bar).pipe(map(resp => new FooSuccess(resp))),
    error => new FooError()
  ),
);

or even split the observable-provider function:

const effecOp$: Observable<FooSuccess|FooError> = actions$.pipe(
  ofType(FOO),
  concatMapToAction( // <-- concatMap-based implementation
    action => this.service.fetch(a.bar),
    success => new FooSuccess(success),
    error => new FooError()
  ),
);

@timdeschryver @brandonroberts Thoughts? Also, ideally I'd like to get it into 8.0.

[brandonroberts]

I'd like to keep the API surface small, keep the number of operators to update/add low, and not make users have to decide on which *MapToAction to use. I propose a separate option that combines the two ideas, with a default of concatMap if its the most common choice out of the box. In most cases, users won't deviate from the default. If you want to use something other than concatMap, provide your own map, error, complete operator sequence.

const effecOp$: Observable<FooSuccess|FooError> = actions$.pipe(
  ofType(FOO),
  mapToAction( // <-- concatMap-based implementation
    fromAction => this.service.fetch(a.bar).pipe(map(resp => new FooSuccess(resp))),
    toError => new FooError(),
    toOptionalComplete => new FooComplete()
  ),
);

I think it would help streamline the sequence as a primary operator, and users have to worry less about nesting observable within a flattening operator.

[wesleygrimes]

+1 to the mapToAction solution

[wesleygrimes]

I'm also proposing a new higher-order effect called createApiEffect here #1826 that will utilize mapToAction

[alex-okrushko]

I was thinking more about @jnorkus suggestion from #1224 (comment) and I think I like it now.

ErrorHandler is already used to catch any unhandled error coming from effects but re-subscribing by default could be a good idea.

If anything, users would be able to remove that behavior with {resubscribeAfterError: false} (so it's true by default) either in @Effect decorator or createEffect(...) fn.