scala – combineLatest只在其中一个流发生变化时才会发出

发布时间：2020-12-16 18:50:53 所属栏目：安全来源：网络整理

导读：我有一个频繁值的流和一个慢的流.我希望将它们组合起来,但只在较慢的一个发出时才发出一个值.所以combineLatest不起作用. 像这样： a1a2b1(a2,b1)a3a4a5b2(a5,b2) 目前我正在这样做,有更清洁的方式吗？ withLatest[A,B](fast : Observable[A],slow : Observa

我有一个频繁值的流和一个慢的流.我希望将它们组合起来,但只在较慢的一个发出时才发出一个值.所以combineLatest不起作用.
像这样：

a1
a2
b1
(a2,b1)
a3
a4
a5
b2
(a5,b2)

目前我正在这样做,有更清洁的方式吗？

withLatest[A,B](fast : Observable[A],slow : Observable[B]): Observable[(A,B)] =
  Observable({ o =>
    var last : A
    fast.subscribe({a => last = a})
    slow.subscribe({b => o.onNext((last,b))})
  })

编辑：此运算符现在位于Rx中,称为withLatestFrom.

解决方法

您正在寻找的是我称之为“combinePrev”的组合器,它在API中不存在,但在许多情况下证明是非常必要的.示例运算符接近,但它不会合并两个流. I’ve also missed “combinePrev” in RxJS.事实证明,“combinePrev”(“withLatest”)的实现很简单,只取决于地图和开关：

withLatest[A,B)] = {
  val hotSlow = slow.publish.refCount
  fast.map({a => hotSlow.map({b => (a,b)})}).switch
}

这是在RxJS中实现的相同运算符的jsfiddle示例.

虽然运算符不在Rx中,但您可以使用隐式类,因此可以使用slow.withLatest(fast)：

implicit class RXwithLatest[B](slow: Observable[B]) {
  def withLatest[A](fast : Observable[A]) : Observable[(A,B)] = /* see above */
}

注意：慢必须很热.如果慢是一个冷Observable,withLatest不起作用.

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!