RxJs/NgRx - 有没有办法在延迟运算符之后“取消”流答案

【问题标题】：RxJs/NgRx - is there a way to "Cancel" a stream after the delay operatorRxJs/NgRx - 有没有办法在延迟运算符之后“取消”流
【发布时间】：2020-02-14 06:07:51
【问题描述】：

我正在使用 NgRx 在我的 Angular 应用程序中使用轮询方案。

为了简化事情，我有类似以下的内容......

    public stopPolling$ = createEffect(() => this.actions$.pipe(
        ofType(actions.stopPolling),
        tap(_ => this.isPollingActive = false),    
        map(_ => actions.stopPolling())
      ), { dispatch: false });

     public continuePolling$ = createEffect(() => this.actions$.pipe(
        ofType(actions.getData),
        tap(_ => this.logger.debug('continue polling')),    
        delay(8000),    
        switchMap(_ => this.pollData())
      ), { dispatch: false });


    private pollData() {
       if (!this.isPollingActive)
         return;
    }

在我的“StopPolling”中，我设置了一个标志，但如果在我处于delay(8000) 时重新启动它（即 isPollingActive 重新设置为 true），延迟将退出，我最终会调用 getData多次。

所以，我的问题是，有没有办法在延迟后调用switchMap(_ => this.pollData()) - 即有没有办法在超时之前“强制延迟退出”？

几乎（如果您了解 C#/.net）。就像 manualResetEvent.WaitOne(8000) 可以通过在 manualResetEvent 对象上调用 Set() 来取消。

我希望我已经描述清楚了吗？

提前致谢

【问题讨论】：

标签： angular rxjs ngrx

【解决方案1】：

您可以使用timer 创建一个延迟后发射的可观察对象，并使用takeUntil 取消订阅以提前退出：

this.actions$.pipe(
  ofType(actions.getData),
  tap(_ => this.logger.debug('continue polling')),   
  switchMap(_ =>
    timer(8000).pipe(
      takeUntil(this.actions$.pipe(ofType(actions.stopPolling))),
      concatMap(() => this.pollData())
    )
)

这还可以让您移除副作用this.isPollingActive = false 并确保控制流保持在可观察对象内。

【讨论】：