从 Futures/Iterables 测试 rx-observables答案

【问题标题】：Testing rx-observables from Futures/Iterables从 Futures/Iterables 测试 rx-observables
【发布时间】：2018-06-22 16:20:20
【问题描述】：

我有：

val observable: Observable[Int] = Observable.from(List(5))

我可以通过测试测试输入列表确实传递给了 observable：

materializeValues(observable) should contain (5)

materializeValues 在哪里：

def materializeValues[T](observable: Observable[T]): List[T] = {
  observable.toBlocking.toIterable.toList
}

现在，如果我从未来创建一个 observable，由于测试超时，我似乎无法使用 materializeValues 进行测试。所以如果我有：

val futVal = Future.successful(5)
val observable: Observable[Int] = Observable.from(futVal)
materializeValues(observable) should contain(5)

它超时并且没有通过测试。物化这两个可观察的过程有什么不同，导致我无法阻止它？

另外，测试 observable 的惯用方式是什么？有没有不打电话toBlocking的办法？

【问题讨论】：

由于Future 的Observable 超时，我无法重现此行为。也许您对隐含的ExecutionContext 有一些问题？同时提供minimal reproducible example 总是一个好主意。
这里有一个要点：gist.github.com/ShahOdin/74fb61e61f98f65261f491d73628d993 ExecutionContext 来自AsyncTestSuite

标签： scala rx-java reactive-programming scalatest rx-scala

【解决方案1】：

我认为问题在于您使用AsyncWordSpecLike（顺便问一下为什么AsyncWordSpecLike 而不是AsyncWordSpec？）。 AsyncWordSpecLike/AsyncWordSpec 旨在简化测试Future。不幸的是，Observable 是一个更强大的抽象，不能轻易映射到 Future。

特别是AsyncWordSpecLike/AsyncWordSpec 允许您的测试返回Future[Assertion]。为了使它成为可能，它提供了自定义隐式ExecutionContext，它可以强制执行所有内容并知道所有计划的作业何时完成。但是，相同的自定义ExecutionContext 是您的第二个代码不起作用的原因：仅在您的测试代码执行完成但您的代码在futVal 上阻塞后才开始处理计划作业，因为您的回调注册在Future.onComplete 计划在 ExecutionContext 上运行。这意味着您对自己的线程有一种死锁。

我不确定在 Scala 上测试 Observable 的官方方法是什么。在 Java 中，我认为 TestSubscriber 是建议的工具。正如我所说，Observable 从根本上来说比Future 更强大，所以我认为测试Observable 你应该避免使用AsyncWordSpecLike/AsyncWordSpec。如果你切换到使用FlatSpec 或WordSpec，你可以这样做：

class MyObservableTestSpec extends WordSpec with Matchers {

  import scala.concurrent.ExecutionContext.Implicits.global
  val testValue = 5

  "observables" should {

    "be testable if created from futures" in {
      val futVal = Future.successful(testValue)
      val observable = Observable.from(futVal)

      val subscriber = TestSubscriber[Int]()
      observable(subscriber)
      subscriber.awaitTerminalEvent
      // now after awaitTerminalEvent you can use various subscriber.assertXyz methods
      subscriber.assertNoErrors
      subscriber.assertValues(testValue)
      // or you can use Matchers as 
      subscriber.getOnNextEvents should contain(testValue)
    }

  }

}

【讨论】：