fix(effects): resubscribe every time an error occurs (#2165)

Author: timdeschryver

modules/effects/spec/effect_sources.spec.ts

@@ -248,19 +248,18 @@ describe('EffectSources', () => {
       it('should resubscribe on error by default', () => {
         class Eff {
           @Effect()
-          b$ = hot('a--b--c--d').pipe(
+          b$ = hot('a--e--b--e--c--e--d').pipe(
             map(v => {
-              if (v == 'b') throw new Error('An Error');
+              if (v == 'e') throw new Error('An Error');
               return v;
             })
           );
         }
 
         const sources$ = of(new Eff());
 
-        //                       👇 'b' is ignored.
-        const expected = cold('a-----c--d');
-
+        //                       👇 'e' is ignored.
+        const expected = cold('a-----b-----c-----d');
         expect(toActions(sources$)).toBeObservable(expected);
       });
 
@@ -516,17 +515,18 @@ describe('EffectSources', () => {
         const sources$ = of(
           new class {
             b$ = createEffect(() =>
-              hot('a--b--c--d').pipe(
+              hot('a--e--b--e--c--e--d').pipe(
                 map(v => {
-                  if (v == 'b') throw new Error('An Error');
+                  if (v == 'e') throw new Error('An Error');
                   return v;
                 })
               )
             );
           }()
         );
-        //                       👇 'b' is ignored.
-        const expected = cold('a-----c--d');
+
+        //                       👇 'e' is ignored.
+        const expected = cold('a-----b-----c-----d');
 
         expect(toActions(sources$)).toBeObservable(expected);
       });

modules/effects/src/effects_resolver.ts

@@ -25,13 +25,7 @@ export function mergeEffects(
           : sourceInstance[propertyName];
 
       const resubscribable$ = resubscribeOnError
-        ? observable$.pipe(
-            catchError(error => {
-              if (errorHandler) errorHandler.handleError(error);
-              // Return observable that produces this particular effect
-              return observable$;
-            })
-          )
+        ? resubscribeInCaseOfError(observable$, errorHandler)
         : observable$;
 
       if (dispatch === false) {
@@ -56,3 +50,16 @@ export function mergeEffects(
 
   return merge(...observables$);
 }
+
+function resubscribeInCaseOfError<T extends Action>(
+  observable$: Observable<T>,
+  errorHandler?: ErrorHandler
+): Observable<T> {
+  return observable$.pipe(
+    catchError(error => {
+      if (errorHandler) errorHandler.handleError(error);
+      // Return observable that produces this particular effect
+      return resubscribeInCaseOfError(observable$, errorHandler);
+    })
+  );
+}