RXJava 2.0 深度嵌套链无法执行部分答案

【问题标题】：RXJava 2.0 deeply nested chain failing to execute partsRXJava 2.0 深度嵌套链无法执行部分
【发布时间】：2019-03-21 00:15:09
【问题描述】：

我有以下 RXJava 2.0 代码：

private fun <T> wrapApiRequestSingle(apiCall: () -> Single<T>, token: Token) : Single<T> =
        Single.defer {
            apiCall.invoke()
        }.retryWhen { obsError ->
            obsError.flatMap<Single<T>> { error ->
                when (error) {
                    is TokenExpiredException -> {
                        userRepository.getLoggedInUser().toFlowable().flatMap { userOptional ->
                            Publisher<Single<T>> {
                                if (userOptional.isPresent) {
                                    mobileRemote.swapRefreshTokenForAccessToken(token.refreshToken, userOptional.get().emailAddress)
                                            .onErrorResumeNext { refreshError ->
                                                EventReporter.e(TAG, "Failed to refresh JWT.", refreshError)
                                                tokenUseCases.deleteToken().andThen(preferences.singleOrError().flatMap { prefs ->
                                                    prefs.apply {
                                                        this.pushRegistrationId = ""
                                                        this.token = null
                                                    }.apply()

                                                    Single.error<Token>(NoLoggedInUserException())
                                                })
                                            }
                                } else {
                                    EventReporter.e(TAG, "No user was logged in.", error)
                                    tokenUseCases.deleteToken().andThen(preferences.singleOrError().flatMap { prefs ->
                                        prefs.apply {
                                            this.pushRegistrationId = ""
                                            this.token = null
                                        }.apply()

                                        Single.error<Token>(NoLoggedInUserException())
                                    })
                                }
                            }
                        }
                    }
                    else -> {
                        Flowable.error(error)
                    }
                }
            }
        }

这个想法是，所有的 API 调用都将被这个函数包裹起来。这个函数有4个主要的执行路径：

调用成功
调用因TokenExpiredException 而失败，当且仅当存在登录用户时，代码才会回退到尝试刷新。刷新成功，再次进行原来的调用。
调用因TokenExpiredException 而失败，当且仅当存在登录用户时，代码才会回退到尝试刷新。如果刷新失败，则删除一些本地数据并返回包含NoLoggedInUserException 的Single。
调用失败，没有登录用户，所以删除一些本地数据并返回一个包含NoLoggedInUserException的Single。

代码已编译，我已经阅读了我正在使用的所有函数的文档，但运行时无法为第 4 种情况返回 Single.error(NoLoggedInUserException)。

我决定编写一个测试用例来测试第 4 条路径，而无需使用实际的 API 或使用任何实际的服务。这是我的测试代码（它使用 Mockito 模拟出各种子系统，例如 mobileRemote 和 tokenUseCases：

/**
 * Set of tests to test the main presenter
 */
class ResourceInteractorTests : RobolectricTestBase() {
    @Mock
    private lateinit var injector: InjectorProvider

    @Mock
    private lateinit var preferences: Preferences

    @Mock
    private lateinit var userRepository: UserStorage

    @Mock
    private lateinit var tokenUseCases: TokenUseCases

    @Mock
    private lateinit var mobileRemote: MobileRemote

    @Before
    fun setup() {
        // Initialize all the mocks in this class
        MockitoAnnotations.initMocks(this)

        whenever(this.injector.providePreferences()).thenReturn(Observable.just(preferences))
        whenever(this.injector.provideUserStorage()).thenReturn(userRepository)
        whenever(this.injector.provideTokenUseCases()).thenReturn(tokenUseCases)
        whenever(this.injector.provideMobileRemote()).thenReturn(mobileRemote)
    }

    /**
     * Test that getLocations ultimately propagates a [NoLoggedInUserException]
     * When the remote call returns a [TokenExpiredException] and there is no logged in user
     */
    @Test
    fun onGetLocationsFailTokenExpiredNoLoggedInUser() {
        // ARRANGE
        whenever(this.tokenUseCases.getToken()).thenReturn(Single.just(Token("", Date(), "")))
        whenever(this.mobileRemote.getLocations("")).thenReturn(Single.error(TokenExpiredException()))
        whenever(this.userRepository.getLoggedInUser()).thenReturn(Single.just(Optional.absent()))
        whenever(this.tokenUseCases.deleteToken()).thenReturn(Completable.complete())

        val interactor = ResourceInteractor(this.injector)

        // ACT
        val shouldBeError = interactor.getLocations().test()
        shouldBeError.awaitTerminalEvent(3, TimeUnit.SECONDS)

        // ASSERT
        shouldBeError.assertError { it is NoLoggedInUserException }
    }
}

这个想法是只要我的 API 调用返回一个 TokenExpiredException，那么 retryWhen 块就会被命中（这是因为我在我的代码中放置了断点来验证）。模拟的userRepository 然后返回Optional.absent() 以使被测代码进入底部的else 块（它确实如此）。最后，tokenUseCases 模拟为deleteTokenOperation 返回Completable.complete()，这应该导致运行时进入andThen 块。但是，在运行时，永远不会到达andThen 块，并且整个链都没有错误地完成。我不知道为什么会这样，有人有什么想法吗？

编辑：

有人问我，为什么要使用Publisher<Single<T>>，那是因为Single 类型上的retryWhen 方法需要它：

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Single<T> retryWhen(Function<? super Flowable<Throwable>, ? extends Publisher<?>> handler) {
    return toSingle(toFlowable().retryWhen(handler));
}

而 Observable retryWhen 没有：

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Observable<T> retryWhen(
final Function<? super Observable<Throwable>, ? extends ObservableSource<?>> handler) {
    ObjectHelper.requireNonNull(handler, "handler is null");
    return RxJavaPlugins.onAssembly(new ObservableRetryWhen<T>(this, handler));
}

编辑 2：

调用私有函数 wrapApiRequestSingle 的测试代码在这里（澄清问题）：

override fun getLocations(): Single<Collection<Location>> =
        tokenUseCases.getToken()
                .flatMap { jwt ->
                    wrapApiRequestSingle({
                        mobileRemote.getLocations(jwt.accessToken)
                    }, jwt)
                }

编辑 3：

采用 TDD 方法编写函数，每次添加新行时完全重新开始并运行我的测试。现在函数看起来像这样：

private fun <T> wrapApiRequestSingle(apiCall: () -> Single<T>, token: Token) : Single<T> =
        Single.defer {
            apiCall.invoke()
        }.retryWhen { obsError ->
            obsError.flatMap<T> { error ->
                when (error) {
                    is TokenExpiredException -> {
                        userRepository.getLoggedInUser().toFlowable().flatMap { userOptional ->
                            if (userOptional.isPresent) {
                                mobileRemote.swapRefreshTokenForAccessToken(token.refreshToken, userOptional.get().emailAddress).toFlowable()
                                        .onErrorResumeNext {
                                            tokenUseCases.deleteToken().andThen(preferences.toFlowable(BackpressureStrategy.BUFFER).flatMap { prefs ->
                                                prefs.apply {
                                                    this.pushRegistrationId = ""
                                                    this.token = null
                                                }.apply()

                                                Flowable.error<T>(NoLoggedInUserException())
                                            })
                                        }
                            } else {
                                EventReporter.e(TAG, "No user was logged in.", error)
                                tokenUseCases.deleteToken().andThen(
                                        preferences.toFlowable(BackpressureStrategy.BUFFER).flatMap { prefs ->
                                            prefs.apply {
                                                this.pushRegistrationId = ""
                                                this.token = null
                                            }.apply()

                                            Flowable.error<T>(NoLoggedInUserException())
                                        })
                            }
                        }
                    } else -> {
                        Flowable.error<T>(error)
                    }
                }
            }
        }

但是，编译器在弄清楚如何解决我接下来要调用的 onErrorResume 的哪个重载时存在问题。我尝试通过在 lambda 参数上提供一个类型来显式声明重载，但编译器仍然抱怨模棱两可的类型 eval。

【问题讨论】：

为什么返回Publisher<Single<T>>？这是非常不寻常的。你实际上告诉retryWhen 通过给它一个对象实例来重试。它不会关心那个对象是什么，只是它表明应该发生重试。
因为 Single.retryWhen 的方法签名需要它（请参阅我对问题的编辑）
简而言之，这就是你写的：source.retryWhen(errors -> errors.flatMap(e -> Flowable.just(Single.error(new NoLoggedInUserException()))))。您通过返回未进一步扩展的对象 Single.error(...) 来响应错误。
您能详细说明一下吗？你是说我要返回一个 Flowable> ？
只需将obsError.flatMap<Single<T>> 更改为obsError.flatMap<T> 并在is TokenExpiredException -> { 的右大括号之前写入.flatMapSingle { it }。此外，如果您将大语句的一部分提取到它自己的方法中，它会更具可读性和可测试性。

标签： unit-testing kotlin mockito rx-java2

【解决方案1】：

根据我的评论，我的意思是：

private fun <T> wrapApiRequestSingle(apiCall: () -> Single<T>, token: Token) : Single<T> =
    Single.defer {
        apiCall.invoke()
    }.retryWhen { obsError ->
        obsError.flatMap<T> { error ->    // <---------------------------------------
            when (error) {
                is TokenExpiredException -> {
                    userRepository.getLoggedInUser()
                    .toFlowable()
                    .flatMap { userOptional ->
                        Publisher<Single<T>> {
                            if (userOptional.isPresent) {
                                mobileRemote.swapRefreshTokenForAccessToken(
                                    token.refreshToken, userOptional.get().emailAddress)
                                .onErrorResumeNext { refreshError ->
                                    EventReporter.e(TAG, "Failed to refresh JWT.", refreshError)
                                    tokenUseCases.deleteToken()
                                    .andThen(preferences
                                        .singleOrError()
                                        .flatMap { prefs ->
                                                prefs.apply {
                                                    this.pushRegistrationId = ""
                                                    this.token = null
                                                }.apply()

                                                Single.error<Token>(NoLoggedInUserException())
                                            })
                                        }
                            } else {
                                EventReporter.e(TAG, "No user was logged in.", error)
                                tokenUseCases.deleteToken()
                                .andThen(preferences
                                    .singleOrError()
                                    .flatMap { prefs ->
                                        prefs.apply {
                                            this.pushRegistrationId = ""
                                            this.token = null
                                        }.apply()

                                    Single.error<Token>(NoLoggedInUserException())
                                })
                            }
                        }
                    }.flatMapSingle { it } // <------------------------------------
                }
                else -> {
                    Flowable.error(error)
                }
            }
        }
    }

【讨论】：

是的，我做到了，但它没有用：-/我将函数剥离到一个单一的衬里并重新构建它，在每次迭代中运行我的测试以希望缩小范围问题是。
您有一堆日志记录语句，您可以添加更多以查看信号停止的位置以及更多日志记录。此外，对于基于模拟的测试，您可能正在模拟错误的对象，仅从代码中很难分辨。
还不清楚您在测试中调用wrapApiRequestSingle 的位置以及导致apiCall 失败的原因。
我将编辑问题以显示调用代码。我肯定在嘲笑所有必需的依赖关系，并且嘲笑是正确的。您应该能够从测试类中看出这一点。
那么剩下的就是在函数中添加更多日志并检查哪些 lambdas 被执行。