RxJava + Retrofit share() operator

Question

这里奇特的rx场景：

改造原料药：

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可观察的订阅将释放其源。当test2调用doSomething()时，将发生另一个API调用。第二次测试未能反映这一点。

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

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

希望有人能对此提供解释。

Answer 1

如评论中所述，问题是可观察的初始化。这里有一个更详细的解释。

问题就在这里：

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

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

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

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

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