scala - combineLatest 仅在其中一个流发生变化时发出

Question

我有一个带有频繁值的流和一个带有较慢值的流。我想将它们组合起来，但只在较慢的一个发出时才发出一个值。所以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 .

Answer 1

您正在寻找的是我称为“combinePrev”的组合器，它在 API 中不存在，但在许多情况下非常必要。 sample运算符很接近，但它不会组合两个流。 I've also missed "combinePrev" in RxJS .事实证明，“combinePrev”(“withLatest”)的实现很简单，只依赖于 map 和 switch:

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

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

虽然运算符不在 Rx 中，但您可以使用隐式类，以便使用 slow.withLatest(fast) :

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

注: slow必须是 hot .如 slow是一个冷 Observable， withLatest不起作用。