动态更改的轮询间隔

Question

我需要前端应用程序来轮询后端应用程序并刷新显示的数据。我以某种方式解决了它，但是我感觉可以对其进行优化。这是我的解决方案：

        this.pollDataSource$ = new BehaviorSubject<number>(this.pollingInterval);

        this.pollDataSource$
            .pipe(
                switchMap(duration =>
                    of(duration)
                        .pipe(
                            filter(duration => duration !== 0),
                            delay(duration),
                            concatMap(() => this.kubernetesObjectList$)
                        )
                )
            )
            .subscribe({
                next: (kubernetesObjectList) => {
                    this.dataSource.data = kubernetesObjectList;
                    if (this.pollingInterval > 0) {
                        this.pollDataSource$.next(this.pollingInterval);
                    }
                }
            });

        this.pollDataSource$.next(-1);

所以，我有一个影响this.pollingInterval的下拉选择器。另外，我有this.pollDataSource$，它是BehaviorSubject<number>。它发出一个number，用作下一次轮询之前的持续时间。

[当this.pollDataSource$发出-1时（发生在用户单击刷新按钮时发生），必须立即轮询数据源，而不考虑设置了什么轮询间隔。

[this.pollDataSource$发出某个正数（当用户从​​下拉选择器中选择某个轮询间隔时发生），此数字必须用作下一次刷新之前的持续时间。

[当this.pollDataSource$发出0时（当用户在同一下拉选择器中选择停止轮询选项时发生），我们必须停止轮询，直到用户选择新的轮询间隔。

并且一切正常。页面已加载，默认情况下this.pollingInterval的值为10000，因此用户可以立即获取数据，但会在10秒内对其进行更新。当用户点击Refresh时，数据将被更新，并且自此之后的10秒后将进行下一次自动刷新。当用户切换到停止轮询时，数据保持静止。当用户切换到另一个时间间隔时，数据将再次被更新。一切都很棒！但是我觉得我的解决方案不是最佳的。我只是不喜欢这种结构：...pipe...switchMap...of...pipe...有没有一种方法可以简化它？

预先感谢所有线索。

Answer 1

我之所以提到expand的原因是可以使用此运算符代替构造函数

someSubject.pipe(
  // do stuff
).subscribe(
  next: data => {
    // do some more stuff
    someSubject.next(something)
  }

这是您实际使用的结构，因此是关于expand的建议。

然后，在不介绍expand的情况下回到您的问题，您可能会认为是类似的

this.pollDataSource$
        .pipe(
            switchMap(duration =>
                return duration === 0 ?
                  NEVER :
                  this.kubernetesObjectList$.pipe(delay(duration));
            )
        )
        .subscribe({
            next: (kubernetesObjectList) => {
                this.dataSource.data = kubernetesObjectList;
                if (this.pollingInterval > 0) {
                    this.pollDataSource$.next(this.pollingInterval);
                }
            }
        });

    this.pollDataSource$.next(-1);

使用expand可能看起来像

this.pollDataSource$
            .pipe(
                expand(duration =>
                    return duration === 0 ?
                      NEVER :
                      this.kubernetesObjectList$.pipe(
                         delay(duration),
                         map(() =>  duration)
                      );
                )
            )
            .subscribe({
                next: (kubernetesObjectList) => {
                    this.dataSource.data = kubernetesObjectList;
                }
            });

        this.pollDataSource$.next(-1);

这些版本的代码是否更干净，以至于您发现我的原始代码清晰易读，这可能是个人喜好问题。

我使用的所有“可能”和“将会”是由于我没有在操场上测试此代码的事实，因此很有可能在这里发现不起作用的地方。

Answer 2

如果您想按一定的时间间隔进行轮询，请尝试以下操作：

const intervalDelays$ = new BehaviorSubject(10000);

intervalDelays.pipe(
  switchMap(ms => ms === 0 ? EMPTY : interval(ms).pipe(
    concatMap(() => fetchDataObservableHere$)
  ))
);

interval在引擎盖下使用setInterval。只要确保ms长于fetchDataObservableHere$到complete的时间即可。否则，您将面临背压问题。

如果担心背压，您可以这样做：

const intervalDelays$ = new BehaviorSubject(10000);

intervalDelays.pipe(
  switchMap(ms => ms < 0
    // No wait? We'll assume it's "off"
    ? EMPTY
    // just an observable to start the recursive expand call.
    // 'start' doesn't matter. Could be `true` or anything, really.
    : of('start').pipe(
      expand(() => timer(ms).pipe(
        switchMap(() => fetchDataObservableHere$),
        tap(data => {
          // PROCESS DATA HERE
          // this ensures that it's done processing before you move on.
        }),
      )),
  ))
);

上面的代码将看到间隔延迟的传入变化，并启动一个新的内部Observable，该内部Observable将以如下方式递归执行，即它将等待每个值返回并在发出另一个请求之前对其进行处理。