拆分 Monix Observable

Question

我想为monix.reactive.Observable 写一个拆分函数。它应该根据谓词的值将源Observable[A] 拆分为一对新的(Observable[A], Observable[A])，并针对源中的每个元素进行评估。我希望拆分独立于源 Observable 是热的还是冷的。在源很冷的情况下，新的一对 Observable 也应该是冷的，而在源很热的情况下，新的一对 Observable 将是热的。我想知道这样的实现是否可行，如果可以，如何实现（我在下面粘贴了一个失败的测试用例）。

签名，作为隐式类的方法，看起来像或类似于

    /**
      * Split an observable by a predicate, placing values for which the predicate returns true
      * to the right (and values for which the predicate returns false to the left).
      * This is consistent with the convention adopted by Either.cond.
      */
    def split(p: T => Boolean)(implicit scheduler: Scheduler, taskLike: TaskLike[Future]): (Observable[T], Observable[T]) = {
      splitEither[T, T](elem => Either.cond(p(elem), elem, elem))
    }

目前，我有一个简单的实现，它使用源元素并将它们推送到PublishSubject。因此，这对新的 Observables 很热门。我对冷 Observable 的测试失败了。

import monix.eval.TaskLike
import monix.execution.{Ack, Scheduler}
import monix.reactive.{Observable, Observer}
import monix.reactive.subjects.PublishSubject

import scala.concurrent.Future

object ObservableOps {

  implicit class ObservableExtensions[T](o: Observable[T]) {

    /**
      * Split an observable by a predicate, placing values for which the predicate returns true
      * to the right (and values for which the predicate returns false to the left).
      * This is consistent with the convention adopted by Either.cond.
      */
    def split(p: T => Boolean)(implicit scheduler: Scheduler, taskLike: TaskLike[Future]): (Observable[T], Observable[T]) = {
      splitEither[T, T](elem => Either.cond(p(elem), elem, elem))
    }

    /**
      * Split an observable into a pair of Observables, one left, one right, according
      * to a determinant function.
      */
    def splitEither[U, V](f: T => Either[U, V])(implicit scheduler: Scheduler, taskLike: TaskLike[Future]): (Observable[U], Observable[V]) = {
      val l = PublishSubject[U]()
      val r = PublishSubject[V]()

      o.subscribe(new Observer[T] {
        override def onNext(elem: T): Future[Ack] = {
          f(elem) match {
            case Left(u) => l.onNext(u)
            case Right(v) => r.onNext(v)
          }
        }

        override def onError(ex: Throwable): Unit = {
          l.onError(ex)
          r.onError(ex)
        }

        override def onComplete(): Unit = {
          l.onComplete()
          r.onComplete()
        }
      })

      (l, r)
    }
  }
}

//////////


import ObservableOps._

import monix.execution.Scheduler.Implicits.global
import monix.reactive.Observable
import monix.reactive.subjects.PublishSubject

import org.scalatest.FlatSpec
import org.scalatest.Matchers._
import org.scalatest.concurrent.ScalaFutures._

class ObservableOpsSpec extends FlatSpec {

  val isEven: Int => Boolean = _ % 2 == 0

  "Observable Ops" should "split a cold observable" in {
    val o = Observable(1, 2, 3, 4, 5)

    val (l, r) = o.split(isEven)

    l.toListL.runToFuture.futureValue shouldBe List(1, 3, 5)
    r.toListL.runToFuture.futureValue shouldBe List(2, 4)
  }

  "Observable Ops" should "split a hot observable" in {
    val o = PublishSubject[Int]()

    val (l, r) = o.split(isEven)
    val lbuf = l.toListL.runToFuture
    val rbuf = r.toListL.runToFuture

    Observable.fromIterable(1 to 5).mapEvalF(i => o.onNext(i)).subscribe()
    o.onComplete()

    lbuf.futureValue shouldBe List(1, 3, 5)
    rbuf.futureValue shouldBe List(2, 4)
  }
}

我希望上面的两个测试用例都能通过，但 "Observable Ops" should "split a cold observable" 失败了。

编辑：工作代码

两个测试用例都通过的实现如下：

import monix.execution.Scheduler
import monix.reactive.Observable

object ObservableOps {

  implicit class ObservableExtension[T](o: Observable[T]) {

    /**
      * Split an observable by a predicate, placing values for which the predicate returns true
      * to the right (and values for which the predicate returns false to the left).
      * This is consistent with the convention adopted by Either.cond.
      */
    def split(
        p: T => Boolean
    )(implicit scheduler: Scheduler): (Observable[T], Observable[T]) = {
      splitEither[T, T](elem => Either.cond(p(elem), elem, elem))
    }

    /**
      * Split an observable into a pair of Observables, one left, one right, according
      * to a determinant function.
      */
    def splitEither[U, V](
        f: T => Either[U, V]
    )(implicit scheduler: Scheduler): (Observable[U], Observable[V]) = {

      val oo = o.map(f)

      val l = oo.collect {
        case Left(u) => u
      }

      val r = oo.collect {
        case Right(v) => v
      }

      (l, r)
    }
  }
}

Answer 1

class ObservableOpsSpec extends FlatSpec {

  val isEven: Int => Boolean = _ % 2 == 0

  "Observable Ops" should "split a cold observable" in {
    val o = Observable(1, 2, 3, 4, 5)
   val o2 =  o.publish
    val (l, r) = o2.split(isEven)

  val x=   l.toListL.runToFuture
  val y =   r.toListL.runToFuture
    o2.connect()
    x.futureValue shouldBe List(1, 3, 5)
    y.futureValue shouldBe List(2, 4)
  }

  "Observable Ops" should "split a hot observable" in {
    val o = PublishSubject[Int]()

    val (l, r) = o.split(isEven)
    val lbuf = l.toListL.runToFuture
    val rbuf = r.toListL.runToFuture

    Observable.fromIterable(1 to 5).mapEvalF(i => o.onNext(i)).subscribe()
    o.onComplete()

    lbuf.futureValue shouldBe List(1, 3, 5)
    rbuf.futureValue shouldBe List(2, 4)
  }
}

Answer 2

根据定义，Cold observable 对每个订阅者进行惰性评估。如果不对所有内容进行两次评估或将其转换为热门，就无法拆分它。

如果您不介意对所有内容进行两次评估，只需使用两次.filter。 如果您不介意转换为热门，请使用.publish（或.publish.refCount，这样您就不需要手动connect）。 如果您想保留冷/热属性并并行处理两个部分，有一个publishSelector 方法可以让您在有限范围内将任何可观察对象视为热对象：

coldOrHot.publishSelector { totallyHot =>
  val s1 = totallyHot.filter(...).flatMap(...) // any processing
  val s2 = totallyHot.filter(...).mapEval(...) // any processing 2
  Observable(s1, s2).merge
}

除了作用域之外，它的限制是内部 lambda 的结果必须是另一个 Observable（将从 publishSelector 返回），因此您无法获得具有所需签名的助手。但是如果原件是冷的，结果仍然是冷的。