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

Question

我在使用 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).

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

几乎（如果你知道 C#/.net）。就像manualResetEvent.WaitOne(8000)可以通过调用Set()manualResetEvent 对象取消的一样。

我希望我已经清楚地描述了这一点？

提前致谢

Answer 1

您可以创建一个在延迟后发出的 observable 使用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并确保控制流保持在可观察范围内。