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

Closes #2023

Author: peterreisz

modules/effects/spec/effect_sources.spec.ts

@@ -248,18 +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);
       });

modules/effects/src/effects_resolver.ts

@@ -24,14 +24,19 @@ export function mergeEffects(
           ? sourceInstance[propertyName]()
           : sourceInstance[propertyName];
 
+      const resubscribeInCaseOfError = (
+        observable$: Observable<any>
+      ): Observable<any> =>
+        observable$.pipe(
+          catchError(error => {
+            if (errorHandler) errorHandler.handleError(error);
+            // Return observable that produces this particular effect
+            return resubscribeInCaseOfError(observable$);
+          })
+        );
+
       const resubscribable$ = resubscribeOnError
-        ? observable$.pipe(
-            catchError(error => {
-              if (errorHandler) errorHandler.handleError(error);
-              // Return observable that produces this particular effect
-              return observable$;
-            })
-          )
+        ? resubscribeInCaseOfError(observable$)
         : observable$;
 
       if (dispatch === false) {