scala - 拆分 Monix Observable
问题描述
我想为monix.reactive.Observable
. 它应该根据谓词的值将源Observable[A]
拆分为新的 pair ,并针对源中的每个元素进行评估。(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)
}
}
}
解决方案
根据定义,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 返回),因此您无法获得具有所需签名的助手。但是如果原件是冷的,结果仍然是冷的。
推荐阅读
- django - 尝试使用 Django、Redis 和 Apache wsgi 运行 2 个 Celery 守护进程和队列
- c# - Windows 服务中未调用 UDP 侦听器回调
- python - 获取按钮和表单的黑色屏幕截图
- r - 如何以第三个变量(日期)为条件使新变量等于现有变量(MA)
- c - 试图复制迷宫算法
- javascript - 如何让计时器显示和隐藏 div?
- swift - 如何在swiftui中使两个具有可观察对象协议的类相互访问
- r - 在 R / 编写内部带有外部函数的循环函数
- spring-boot - 为什么在启动 Java 应用程序时出现以下错误?
- java - Hofstadter 在 Java 中的 G 序列递归