Angular NgRx-仅在首次调用时才继续轮询服务的效果

Question

我有一个应用程序，我刚刚在其中添加了NgRX，我希望使用效果来打开和关闭轮询。
样本大纲
我遵循了this post，这似乎是一种不错的方法。我有一个here的简化示例，其中大部分代码都在app.effects.ts中。
与示例类似，除了使用新的startPolling$工厂方法外，我还具有stopPolling$，continuePolling$和createEffect效果。
另外，我已将delay(2000)移到takeWhile()上方，因为我发现如果服务调用引发错误，那么catchError(err => of(appActions.getDataFail(err)))会导致效果进入连续的非常快速的循环而不会产生延迟。
开始和停止按钮调度轮询开始和停止...

public start() {
    console.log('dispatching start');
    this.store.dispatch(appActions.startPolling());
  }

  public stop() {
    console.log('dispatching stop');
    this.store.dispatch(appActions.stopPolling());
  }

我的问题
我有一些控制台日志，所以我们可以看到发生了什么。
当我们单击 开始按钮(恰好是 的第一个时间)时，我可以看到轮询开始，然后按预期继续。例如，我可以一遍又一遍地看到以下内容...

dispatching start
app effect started polling
app.service.getData
app effect continue polling
app.service.getData
app effect continue polling
app.service.getData
app effect continue polling

完美的。
当我停下来时，我看到

dispatching stop
app effect stop polling

也正确。
现在，问题 是当我尝试重新启动时。如果我现在再次单击开始按钮，那么我看到的只是初始开始轮询效果...

dispatching start
app effect started polling
app.service.getData

并且 continuePolling$中的代码不再称为，因此我没有轮询。
有谁知道为什么这种效果没有在秒时间内触发？我只是不知道为什么会这样。
更新1
我认为也许我的问题是，一旦将 isPollingActive设置为false，并且 takeWhile(() => this.isPollingActive),“停止”，则可观察项将不再处于事件状态，即 continuePolling$完成，因此将永远不会重新启动吗？
假设这样做，我尝试了以下操作，其中有2个不同的变量，一个变量“暂停”轮询(例如，如果我在离线模式下检测到该应用程序)，另一个变量取消(即，当用户从组件中导航时) 。
所以，我的整个效果现在变成了...

    @Injectable()
    export class AppEffects {
      private isPollingCancelled: boolean;
      private isPollingPaused: boolean;

      constructor(
        private actions$: Actions,
        private store: Store<AppState>,
        private appDataService: AppDataService
      ) { }

      public startPolling$ = createEffect(() => this.actions$.pipe(
        ofType(appActions.startPolling),
        tap(_ => console.log('app effect started polling')),
        tap(() => {
          this.isPollingCancelled = false;
          this.isPollingPaused = false;
        }),        
          mergeMap(() =>
            this.appDataService.getData()
              .pipe(                        
                switchMap(data => {              
                  return [appActions.getDataSuccess(data)
                  ];
                  }),
                catchError(err => of(appActions.getDataFail(err)))
              ))
        ));

         public pausePolling$ = createEffect(() => this.actions$.pipe(
            ofType(appActions.pausePolling),
            tap(_ => this.isPollingPaused = true),
            tap(_ => console.log('app effect pause polling')),       
         ));
      
      public cancelPolling$ = createEffect(() => this.actions$.pipe(
        ofType(appActions.cancelPolling),
        tap(_ => this.isPollingCancelled = true),
        tap(_ => console.log('app effect cancel polling')),
      ));

        public continuePolling$ = createEffect(() => this.actions$.pipe(
          ofType(appActions.getDataSuccess, appActions.getDataFail),    
          tap(data => console.log('app effect continue polling')),  
          takeWhile(() => !this.isPollingCancelled),    
          delay(3000),  
     
          mergeMap(() =>
            this.appDataService.getData()
              .pipe(   
                delay(3000),  
                tap(data => console.log('app effect continue polling - inner loop')),  
                takeWhile(() => !this.isPollingPaused), // check again incase this has been unset since delay 
                switchMap(data => {              
                  return [appActions.getDataSuccess(data)
                  ];
                  }),
                catchError(err => of(appActions.getDataFail(err)))
              ))
        ));    
    } 

我不建议运行上述命令，因为当我随后调度 pause polling action时，效果似乎陷入了无尽的循环，我不得不通过任务管理器杀死浏览器。
我不知道为什么会这样，但是我似乎比以前更远离解决方案。
更新2
我注意到我没有从暂停和取消效果中返回任何 Action 。
所以我更新了它们，我们关注...

 public pausePolling$ = createEffect(() => this.actions$.pipe(
    ofType(appActions.pausePolling),
    tap(_ => this.isPollingPaused = true),
    tap(_ => console.log('app effect pause polling')),
    map(_ => appActions.pausePollingSuccess())
  ));
  
  public cancelPolling$ = createEffect(() => this.actions$.pipe(
    ofType(appActions.cancelPolling),
    tap(_ => {
      this.isPollingCancelled = true;
      this.isPollingPaused = true;
    }),
    tap(_ => console.log('app effect cancel polling')),
    map(_ => appActions.cancelPollingSuccess())
  ));

现在暂停似乎可以正常工作，但是当我调度 appActions.cancelPolling时，我再次看到 app effect cancel polling无限循环被记录到控制台。
更新3
我发现了为什么得到无限循环以及如何停止它。根据doco here，我可以添加 dispatch:false ...

    public cancelPolling$ = createEffect(() => this.actions$.pipe(
        ofType(appActions.cancelPolling),
        tap(_ => {
          this.isPollingCancelled = true;
          this.isPollingPaused = true;
        }),
        tap(_ => console.log('app effect cancel polling')),
      ), { dispatch: false }); // <------ add this

这似乎解决了我的无限循环。
现在，我唯一的任务是能够确定如何启动，停止和重新启动轮询，从而处理对 appDataService.getData()的成功调用以及异常处理。
我可以让它为另一个服务(取决于我在哪里放置延迟和时间)，但不能同时为两个服务
更新4
我有最新的代码 here。
按原样运行，我使getData成功，并且令人惊讶的是，pause或stop操作将停止它并允许它重新启动。取消效果。
另外，如果将 takeWhile(() => !this.isPollingCancelled),传递给 true，这将导致它可观察到错误。轮询继续进行(根据需要，即，即使出现错误也要重试)，但是一旦我们现在分派(dispatch)暂停操作，它就不会停止轮询，而是分派(dispatch)停止，它确实会停止，但随后将不会重新启动。我赢不了。
更新5
我以为也许是因为继续轮询效果被取消了，所以我可以每次都重新创建它，如下所示。

    import { Injectable, OnInit, OnDestroy } from '@angular/core';
    import { createEffect, Actions, ofType } from '@ngrx/effects';
    import { select, Store } from '@ngrx/store';
    import { mergeMap, map, catchError, takeWhile, delay, tap, switchMap } from 'rxjs/operators';
    import { AppState } from './app.state';
    import { Observable, of } from 'rxjs';
    import { AppDataService } from '../app-data.service';
    import * as appActions from './app.actions';

    @Injectable()
    export class AppEffects {
      private isPollingCancelled: boolean;
      private isPollingPaused: boolean;

      constructor(
        private actions$: Actions,
        private store: Store<AppState>,
        private appDataService: AppDataService
      ) { }

      public startPolling$ = createEffect(() => this.actions$.pipe(
        ofType(appActions.startPolling),
        tap(_ => console.log('app effect started polling')),
        tap(() => {
          this.isPollingCancelled = false;
          this.isPollingPaused = false;
          this.createPollingEffect(); // <--- recreate the effect every time
        }),        
          mergeMap(() =>
            this.appDataService.getData()
              .pipe(                        
                switchMap(data => {              
                  return [appActions.getDataSuccess(data)
                  ];
                  }),
                catchError(err => of(appActions.getDataFail(err)))
              ))
        ));

      public pausePolling$ = createEffect(() => this.actions$.pipe(
        ofType(appActions.pausePolling),
        tap(_ => this.isPollingPaused = true),
        tap(_ => console.log('app effect pause polling')),
      ), { dispatch: false });
      
      public cancelPolling$ = createEffect(() => this.actions$.pipe(
        ofType(appActions.cancelPolling),
        tap(_ => {
          this.isPollingCancelled = true;
          this.isPollingPaused = true;
        }),
        tap(_ => console.log('app effect cancel polling')),
      ), { dispatch: false });

      public continuePolling$: any;

      private createPollingEffect(): void {
        console.log('creating continuePolling$');
        this.continuePolling$ = createEffect(() => this.actions$.pipe(
          ofType(appActions.getDataSuccess, appActions.getDataFail),
          tap(data => console.log('app effect continue polling')),
          delay(3000),
          takeWhile(() => !this.isPollingCancelled),
          mergeMap(() =>
            this.appDataService.getData(false)
              .pipe(
                tap(data => console.log('app effect continue polling - inner loop')),

                switchMap(data => {
                  return [appActions.getDataSuccess(data)
                  ];
                }),
                catchError(err => of(appActions.getDataFail(err)))
              ))
        ), { resubscribeOnError: true });
      } 
    }

因此，在 getData中，我调用 startPolling来创建继续轮询效果。
但是，当我尝试此操作时，轮询将永远不会开始。
更新6
我想出了一个似乎对我有用的解决方案。
我有以下

public startPolling$ = createEffect(() => this.actions$.pipe(
        ofType(dataActions.startPollingGetData),
        tap(_ => this.logger.info('effect start polling')),
        tap(() => this.isPollingActive = true),
        switchMap(_ => this.syncData())
      ), { dispatch: false });
      
    public continuePolling$ = createEffect(() => this.actions$.pipe(
        ofType(dataPlannerActions.DataSuccess,
          dataActions.DataFail),
        tap(_ => this.logger.debug('data effect continue polling')),
        tap(_ => this.isInDelay = true),
        delay(8000),
        tap(_ => this.isInDelay = false),
        switchMap(_ => this.syncData())
      ), { dispatch: false });


    public stopPolling$ = createEffect(() => this.actions$.pipe(
        ofType(dataActions.stopPollingData),
        tap(_ => this.isPollingActive = false),
        tap(_ => this.logger.info('data effect stop polling')),
        map(_ => dataActions.stopPollingDataSuccess())
      ), { dispatch: false });


    private syncData(): Observable<Action> {
        const result$: Observable<Action> = Observable.create(async subscriber => {
          try {
            // If polling "switched off", we just need to return anything (not actually used)
            // Id isInDelay, we may be restating while we still have a pending delay.
            // In this case we will exit, and just wait for the delay to restart
            // (otherwise we can end up with more than one call to this)
            if (this.isInDelay || !this.isPollingActive) {
              subscriber.next("");
              return;
            }

我在这里使用了几个“标志”，我敢肯定，这样做会更“rxy”。
实际上， see this post讨论如何摆脱 this.createPollingEffect()(我只需要解决这个问题，然后将其放入上面的生产代码中)

Answer 1

改用它:

public startPolling$ = createEffect(() => this.actions$.pipe(
  ofType(appActions.startPolling),    
  tap(_ => console.log('app effect started polling')),  
  tap(() => this.isPollingActive = true),        
  switchMap(() =>
    this.appDataSurvice.getData()
      .pipe(                        
        exhaustMap(data => {              
          return [appActions.getDataSuccess(data)];
        }),
        catchError(err => of(appActions.getDataFail(err)))
      ))
));