Angular 2 - 如何更改 RxJS Observable 的间隔

Question

我正在使用 rxJS Observable Interval 来刷新正在获取的数据。我无法弄清楚更改间隔设置的方法。我已经看到了一些关于使用 rxJS 提供的 Subject 类的东西，但我无法让它工作。

我在这个plunk中提供了一个简化的例子

在 AppComponent 我有这个方法。

getTime() {
        this.timeService.getTime(this.refreshInterval)
          .subscribe(t => {
            this.currentTime = t;
            console.log('Refresh interval is: ' + this.refreshInterval);
          }
        );
}

在服务组件中，我目前有此代码。

getTime(refreshInterval: number) {
  return Observable.interval(refreshInterval)
        .startWith(0)
        .map((res: any) => this.getDate())
        .catch(this.handleError)
}

也许有人可以为我提供一个工作示例，那就太好了！

Answer 1

据我从您的 plnkr 中了解到，您的目标是允许用户修改计时器间隔。

如你所料，refreshInterval 的改变会改变声明的 rxJs 流：

    this.timeService.getTime(this.refreshInterval)
      .subscribe(t => {
        this.currentTime = t;
        console.log('Refresh interval is: ' + this.refreshInterval);
      }
    );

这是错误的。

每次更新refreshInterval，都需要：

• 取消订阅或销毁之前的流。
• 创建新的流和 再次订阅

Answer 2

您无需销毁并重新创建整个 Observable 流即可更改 refreshInterval。您只需要更新取决于更改间隔的部分流。

首先简化您的服务的getTime()，使其不负责确定输出频率。它所做的只是返回时间：

getTime() { return (new Date()).toString(); }

现在调用代码将确定时间表。只需 3 个简单的步骤：

1.调整到所需间隔的源函数：

/** Observable waits for the current interval, then emits once */
refreshObs() {return Observable.timer(this.refreshInterval)}

2. 使用repeat operator 不断重新执行流的可观察链：

getTime$ = Observable.of(null)
            .switchMap(e=>this.refreshObs()) // wait for interval, then emit
            .map(() => this.timeService.getTime()) // get new time
            .repeat(); // start over

3.订阅触发整个事情：

ngOnInit(){
    this.getTime$.subscribe(t => {
        this.currentTime = t;
        console.log('refresh interval = '+this.refreshInterval);
    });
}

这是因为 refreshObs() 每次重复流时都会返回一个新的 observable，并且该新的 observable 将根据当前设置的间隔在发射前等待。

Live demo

Answer 3

我想在这里（以及 Stack Overflow 上的其他地方）建立以前的答案。我的示例有一个常见的RefreshService，各种组件都可以使用它进行订阅。这样，一个站点可以有一个“每 X 秒刷新一次”组件，每个组件都可以订阅该组件。

https://plnkr.co/edit/960yztjl3dqXQD2XPSei?p=preview

该服务提供函数withRefresh，该函数提供Observable。它利用了BehaviorSubject，它将立即触发订阅事件。

export class RefreshService {

  static interval$: BehaviorSubject<number> = new BehaviorSubject<number>(30000);

  setInterval(newInterval: number){
    RefreshService.interval$.next(newInterval);
  }

  public withRefresh() {
    return RefreshService.interval$
      .switchMap((int: number) => Observable
        .interval(int)
        .startWith(0)
      );
  }
}

那么任何组件都可以使用这个服务，如下：

this.refreshService
  .withRefresh()
  .switchMap(() => /* do something on each interval of the timer */);