RxJS 按间隔和手动调用时轮询

Question

在我的 Angular 应用程序中，我正在处理通知，并且我有一个 REST API 来调用最新用户的通知。我需要在几分钟内调用此 API，因为用户实时获取通知并不重要（它们甚至可能不会出现得那么快）。然而，在客户端刷新通知的想法是下一个：

• 当用户登录时开始刷新通知 - 这是几分钟后开始刷新 API 的第一个手动调用
  • 如果用户让应用保持打开状态或只是在应用中导航，则不要更改计时器并等待剩余时间
  • 如果用户打开子页面可以执行与通知相关的操作并执行此操作，则刷新通知并重置计时器
• 在注销前刷新通知

我已经有了所描述过程的工作代码，但我不确定它是否适合我的需要。这是执行调用的代码（手动检查只有一个主题，停止检查有一个可观察的订阅 - 下面的代码实际上是分开的，但由于可读性，这里放在一个地方）：

// Subject for manual triggering
this.checkFeed = new Subject<void>();

// Call for refresh in own method
this.checkFeed.next();

// Waiting for manual refresh or triggering it on some interval after it was last triggered
this.feedSub = this.checkFeed.asObservable()
      .switchMap(() => Observable.timer(0, this.interval))
      .mergeMap(() => this.fetchChanges())
      .distinctUntilChanged(this.compareFeed)
      .subscribe(res => this.notify(res));

// Unsubscription when logging out
if (this.feedSub) this.feedSub.unsubscribe();

我最不确定的部分是.switchMap(() => Observable.timer(0, this.interval))，因为它需要0 才能立即开始（这没关系，但看起来仍然不正确？）。那么有没有更好的方法来实现我所描述的？

我还有另一个问题，如何开始检查来自另一个 observable 的通知 - 我应该使用哪个运算符。正如我所提到的，我已经在自己的方法中调用了主题的下一个，如下所示：

refreshFeed(): void {
  this.checkFeed.next();
}

因此，当有其他可观察到的执行（应刷新通知时的操作）时，我需要调用它。当其他 observable 有来自 API 的响应时，调用 void 方法的正确方法是什么？我在想这样的事情：

someActionThatCanChangeNotifications(): Observable<any> {
  return this.api.get('path/to/endpoint')
    .do(() => this.feedService.refreshFeed());
}

这样可以吗，还是有更好的办法？

提前感谢您的帮助！

Answer 1

所以基本上你有两个可观察的。

您手动调用的：

this.checkFeed

还有间隔（我们称之为intervalObs）：

this.intervalObs = Observable.timer(0, this.interval);

如果你看到这样最简单的方法是合并你的两个源流，然后做任何你想做的事情。

var mergedSource = Observable.merge(
    this.checkFeed,
    this.intervalObs)

subscription = mergedSource.subscribe(this.fetchChanges());

也许你需要在两者之间做一些更多的操作，但这应该会给你一个更易读的选择。

如果你想在https://plnkr.co/edit/n4nNFEMa4YOh2KSjDpSJ?p=preview987654321@附近玩，你可以试试这个工作的plunker

Answer 2

据我所知，您已经“正确”地完成了它。与一般的编程一样，单个问题有许多可能的（和正确的）解决方案。就个人而言，我也会这样做。

我也可以对你提到的两点给你一些评论：

.switchMap(() => Observable.timer(0, this.interval))

Observable.timer 几乎是Observable.interval，在第一个值之前有一个自定义超时。 Observable.timer(0, this.interval) 是正确的用法。

另一种选择是Observable.just(0).concat(Observable.interval(this.interval))，它立即返回一个值，然后开始间隔。但是，我更喜欢您的放置方式；我认为它清楚地说明了您的意图：“在0 毫秒后产生一个值，然后是this.interval 的间隔”。

.do(() => this.feedService.refreshFeed())

我会说这是完全正确的做法。 do 用于副作用，例如。发生在外部可观察的东西。

但我可以说，我不希望 someActionThatCanChangeNotifications 开始刷新提要。当一个函数返回一个 observable 时，我希望返回一个没有任何副作用的 observable。然而，由于我们生活在一个不完美的世界中，我们不可能总是拥有我们想要的。

你不能指望每个订阅者都记得做.do(() => this.feedService.refreshFeed())，相反我会在函数的文档注释中添加一条通知：“注意：返回的 observable 将在每个下一个信号时刷新提要”，或者那种东西。