【问题标题】:Angular NgRx - Effect to continue polling a service only called the first timeAngular NgRx - 仅在第一次调用时继续轮询服务的效果
【发布时间】:2020-02-14 00:01:32
【问题描述】:

我有一个应用程序,我刚刚添加了 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),“停止”,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)))
              ))
        ));    
    } 

我不建议运行上面的,因为当我发送 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

我找到了为什么会出现无限循环以及如何停止它。根据文档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 成功了,令人惊讶的是,暂停或停止操作都会停止它并允许它重新启动。我很惊讶 stop 操作允许它重新启动,因为我假设 takeWhile(() =&gt; !this.isPollingCancelled), 会取消效果。

另外,如果 true 被传递给 getData 这将导致它可以观察到错误。轮询继续(如所愿,即即使出现错误仍会重试),但是一旦我们现在调度暂停操作,它不会停止轮询,并且我们调度停止,它确实停止了,但是它不会重新启动。我赢不了。

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

所以,在startPolling 中我调用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”的方式。

事实上,see this post 关于如何摆脱 isInDelay(我只需要把它放到我上面的生产代码中)

【问题讨论】:

    标签: angular rxjs ngrx ngrx-effects


    【解决方案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)))
          ))
    ));
    

    【讨论】:

    • @MoxxiMangarn 感谢回复,但是我改成上面的,第二次轮询还是没有开始。我有你的修改folked here
    【解决方案2】:

    您解决问题的方式值得称赞。我在重新启动轮询时遇到了完全相同的问题,这篇文章帮助了我。

    我现在面临的一个问题是,如果轮询在不到 3 秒(指定计时器)内重新启动,则会多次调用服务。换句话说,轮询仅在间隔过去后才完全暂停/停止。因此,如果您尝试在计时器结束之前再次启动它,则会运行多个线程。刚刚在服务调用中添加了时间戳@https://angular-ngrx-polling3-j7b8st.stackblitz.io

    每次轮询都会调用服务两次。

    【讨论】:

    • 谢谢你。如果你看一下问题中的 UPDATE6,我已经把我最终使用的东西放了出来。从rxjs 的角度来看,它可能不是最纯粹的,但到目前为止的所有测试都对我有用。为了停止“多次执行”,您在延迟期间停止/启动轮询,我添加了 this.isInDelay 标志。请参阅链接以获取看起来更好的“更多 rx”解决方案,但到目前为止,上述内容对我有用,我不再获得“线程”的构建(是的,我知道它们不是真正的线程)做投票
    • 非常感谢您的回复。这有帮助:)
    • 如果有帮助,我可能会将其作为解决方案,这样对其他人来说更明显。
    • 是的。我使用了stackoverflow.com/questions/60220897/… 提供的解决方案。到目前为止,我已经对所有情况进行了正面测试。
    【解决方案3】:

    我将此作为我的问题/讨论的一部分,但我认为会作为一种解决方案来提高知名度......

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

    我有以下

    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 关于如何摆脱 isInDelay(我只需要把它放到我上面的生产代码中)

    【讨论】:

      【解决方案4】:

      根据@peterc 和@Ingo Bürk 的输入,我能够对所有情景进行正面测试。下面是我的代码的外观。

      @Effect()
            getPageData$ = this.actions$.pipe(
              ofType(actions.StartLoading),
              tap(() => {
                this.appService.isPollingActive = true;
              }),
              mergeMap(() =>
                this.appService.getData().pipe(
                  switchMap((response: GridDataResponse) => {
                    return [new actions.DoneLoading(response.data)];
                  }),
                  retry(1),
                  catchError(err => {
                    return of(new actions.FailedLoading());
                  })
                ))
            );
      
            @Effect()
            public stopPolling$ = this.actions$.pipe(
              ofType(actions.StopPolling),
              tap(_ => {
                this.appService.isPollingActive = false;
              }),
              mergeMap(() => {
                return [new actions.ResetLoading()];
              })
            );
      
            @Effect()
            public continuePolling$ = this.actions$.pipe(
              ofType(actions.DoneLoading,
                actions.FailedLoading),
              switchMap(_ =>
                timer(this.appService.pollingTimer).pipe(
                  takeUntil(this.actions$.pipe(ofType(actions.StopPolling))),
                  mergeMap(() =>
                  this.appService.getData().pipe(
                    takeWhile(() => this.appService.isPollingActive),
                    switchMap((response: GridDataResponse) => {
                      return [new actions.DoneLoading(response.data)];
                    }),
                    catchError(err => {
                      return of(new actions.FailedLoading());
                    })
                  ))
                )
            )
            );
      

      【讨论】:

        猜你喜欢
        • 1970-01-01
        • 2020-01-20
        • 1970-01-01
        • 1970-01-01
        • 2017-07-08
        • 2022-10-16
        • 2018-09-11
        • 1970-01-01
        • 1970-01-01
        相关资源
        最近更新 更多