RxJava + Retrofit share() 运算符答案

【问题标题】：RxJava + Retrofit share() operatorRxJava + Retrofit share() 运算符
【发布时间】：2019-02-23 03:51:52
【问题描述】：

这里有特殊的 rx 场景：

改造 API：

interface MyApi {
  @Headers("Content-Type: application/json")
  @POST("something")
  fun doSomething(@Body body: SomeRequestBody): Single<SomeResponse>
}

可以从多个地方调用此 API。所以，我想分享一下。我的存储库公开了这一点：

class Repository {

  private val observable: Observable<SomeResponse> by lazy {
    myApi.doSomething(SomeRequestBody())
        .toObservable()
        .share()
  }

  fun doSomething(): Completable {
    observable.flatMapCompletable { Completable.complete() }
  }
}

我正在使用以下内容进行测试：

// passes, as expected
@Test
fun `multiple api calls share`() {
  given(myApi.doSomething(any())).willReturn(Single.just(RESPONSE).delay(2, SECONDS))

  val test1 = repository.doSomething().test()
  val test2 = repository.doSomething().test()
  val test3 = repository.doSomething().test()

  test1.await(3, SECONDS)
  test2.await(3, SECONDS)
  test3.await(3, SECONDS)

  test1.assertNoErrors()
  test2.assertNoErrors()
  test3.assertNoErrors()

  test1.assertComplete()
  test2.assertComplete()
  test3.assertComplete()

  verify(myApi, times(1) /* for clarity */).doSomething(any())
}

// fails :(
@Test
fun `multiple api calls, one after the other`() {
  given(myApi.doSomething(any())).willReturn(Single.just(RESPONSE).delay(2, SECONDS))
      .willReturn(Single.just(OTHER_RESPONSE).delay(2, SECONDS))

  val test1 = repository.doSomething().test()

  test1.await(3, SECONDS)
  test1.assertNoErrors()
  test1.assertComplete()
  // even tried explicitly disposing here
  test1.dispose()

  val test2 = repository.doSomething().test()
  test2.await(3, SECONDS)
  test2.assertNoErrors()
  test2.assertComplete()

  // fails here
  verify(myApi, times(2)).doSomething(any())
}

我的理解是，如果所有订阅都已被处置，shared observable 将处置其来源。而当test2 调用doSomething() 时，会发生另一个API 调用。第二次测试未能反映这一点。

另外，如果我将 API 调用包装在 defer() 中，则两个测试都通过：

private val observable: Observable<SomeResponse> by lazy {
  Single.defer {
    myApi.doSomething(SomeRequestBody())
  }.toObservable().share()
}

希望有人可以对此提供解释。

【问题讨论】：

我可能没有正确理解这一点，但这不是因为您正在懒惰地初始化存储库内部可观察对象吗？它被初始化一次并调用doSomething 方法，但是当repo 的实例保持不变时，它永远不会再次调用它。将其包装在defer 中，使其成为订阅后执行的可观察对象的一部分，但defer 本身仅被调用一次。当您订阅时，不会对defer 进行新的调用，但它将再次运行源代码，执行doSomething 行并且测试通过。这可以理解吗？去掉惰性部分还能用吗？
@Fred 我想就是这样。我还注意到，即使没有Signle.defer/fromCallable，针对MockWebServer 的测试（而不是像上面那样模拟API）也可以工作。因此，Rx 调用适配器可能正在做类似的事情。如果不麻烦，您可以将其发布为答案吗？谢谢:)
当然！很高兴，我稍后会在我的笔记本电脑上做这件事 :) 很高兴它成功了。

标签： android rx-java retrofit retrofit2 rx-java2

【解决方案1】：

正如 cmets 中所讨论的，问题在于可观察初始化。这里有更详细的解释。

问题出在这里：

private val observable: Observable<SomeResponse> by lazy {
  myApi.doSomething(SomeRequestBody())
    .toObservable()
    .share()
}

变量observable被延迟初始化，这意味着只要我们使用相同的存储库实例，它只会被初始化一次。

因此，在测试中，您有一个存储库实例和多个测试。这意味着，对于整个测试类，lazy 块内的代码运行一次。意思是，myApi.doSomething(any()) 运行一次。当您尝试验证多个交互时，这会导致失败。

当您将其包装在 defer 中时它起作用的原因是 defer 创建了一个 observable 将在每次订阅者订阅时执行（在您的情况下，由于 share 运算符，它有点复杂，但想法是一样的）。像以前一样，defer 被懒惰地执行，并且在测试期间再也不会被调用。也就是说，如果可以验证对defer 的调用，那么结果将是相同的。但是，现在每次运行 observable 时都会调用 myApi.doSomething(any()) 并且测试将通过。

正如您已经想到的那样，您可以通过将调用包装在 defer 中来解决此问题。我认为您也可以简单地删除延迟初始化。也许，甚至使用依赖注入来初始化对象，而不是在测试中延迟初始化，而是在生产应用程序中保持延迟初始化。

【讨论】：