Angular NgRx - 继续轮询仅第一次调用的服务的效果

Question

我有一个应用程序，我刚刚在其中添加了 NgRX，我希望在其中使用效果来打开和关闭轮询。

示例大纲

我关注了这篇文章，这似乎是一个很好的方法。我在这里有一个简化的例子，大部分代码在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),“停止”，observable 不再处于活动状态，即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)))
              ))
        ));    
    } 

我不建议运行上面的程序，因为当我发送 a 时pause polling action，效果似乎进入了无限循环，我必须通过任务管理器杀死浏览器。

我不知道为什么会发生这种情况，但我似乎比以前更远离解决方案。

更新 2

我注意到我没有从暂停和取消效果中返回任何动作。

所以我更新了它们，我们遵循...

 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

我已经找到了为什么会出现无限循环以及如何阻止它。根据这里的文档，我可以添加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

我这里有最新的代码。

按原样运行它，我让 getData 成功，令人惊讶的是，暂停或停止操作将停止它并允许它重新启动..我很惊讶停止操作允许它重新启动，因为我假设takeWhile(() => !this.isPollingCancelled),会取消效果.

此外，如果true传递给getDatathis 将导致它可观察到的错误。轮询继续（如需要，即即使出现错误仍重试），但是一旦我们现在调度暂停操作时，它不会停止轮询，并且当我们调度停止时，它确实停止，但不会重新启动。我赢不了。

更新 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 });
      } 
    }

因此，在startPollingI 调用this.createPollingEffect()中创建继续轮询效果。

但是，当我尝试这样做时，轮询从未开始。

更新 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”的方式来做到这一点。

事实上，请参阅这篇关于如何摆脱isInDelay（我只需要将其放入上面的生产代码中）的帖子

Answer 1

我将此作为我的问题/讨论的一部分，但我认为可以作为一种解决方案，使其更加明显......

我想出了一个似乎对我有用的解决方案。

我有以下内容

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”的方式。

事实上，请参阅这篇文章，了解如何摆脱isInDelay（我只需要把它放入上面的生产代码中）