android - 处理RxJava的onErrorReturn和OnErrorNotImplementedException

Question

所以我试图用RxJava在android中实现MVI模式，但是我想在某种状态下处理抛出的error以及成功和加载，无论如何都可以处理错误而不是subscribe(onError = xxx)
处理

sealed class AuthResult : MviResult {
    sealed class LoadUserResult : AuthResult() {
        object Loading : LoadUserResult()
        data class Success(val user: User) : LoadUserResult()
        data class Fail(val error: Throwable) : LoadUserResult()
    }
}

private val loadUser =
    ObservableTransformer<LoadUserAction, LoadUserResult> { actions ->
        actions.flatMap {
            userManager.getCurrentUser()
                .map<LoadUserResult> { LoadUserResult.Success(it) }
                .onErrorReturn(LoadUserResult::Fail) // HERE? // EDIT FOR THE ANSWER: REMOVE THIS
                .subscribeOn(schedulerProvider.io())
                .observeOn(schedulerProvider.ui())
                .startWith(LoadUserResult.Loading)
        }.onErrorReturn(LoadUserResult::Fail) // ANSWER: ADD THIS TO CATCH API ERROR
    }

var actionProcess =
    ObservableTransformer<AuthAction, AuthResult> { actions ->
        actions.publish { s->
            Observable.merge(
                s.ofType(LoadUserAction::class.java).compose(loadUser),
                s.ofType(SignInWithGoogleAction::class.java).compose(signInWithGoogle)
            )
        }
    }

View 模型

fun combine(): Observable<AuthViewState> {
    return _intents
        .map(this::actionFromIntent)
        .compose(actionProcess)
        .scan(AuthViewState.idle(), reducer)
        .distinctUntilChanged()
        .replay(1)
        .autoConnect(0)
}

分段

disposable.add(viewModel.combine().subscribe(this::response))

private fun response(state: AuthViewState) {
    val user = state.user

    if (user.uid.isBlank() && user.email.isBlank() && user.username.isBlank()) {
        Timber.i("user: $user")
    } else {
        Timber.i("user: $user")
        Toast.makeText(requireContext(), "Will navigate to MainActivity", Toast.LENGTH_SHORT)
            .show()
    }

    // HANDLE THE ERROR HERE?
    if (state.error != null) {
        Toast.makeText(requireContext(), "Error fetching user", Toast.LENGTH_SHORT).show()
        Timber.e("Error loading user ${state.error.localizedMessage}")
    }
}

我得到的错误是

2020-06-03 22:42:15.073 25060-25060/com.xxx W/System.err: io.reactivex.exceptions.OnErrorNotImplementedException: The exception was not handled due to missing onError handler in the subscribe() method call. Further reading: https://github.com/ReactiveX/RxJava/wiki/Error-Handling | com.google.android.gms.tasks.RuntimeExecutionException: com.google.android.gms.common.api.ApiException: 10: 

Answer 1

您在此处收到的错误是由于您在.subscribe()中调用Fragment引起的。该.subscribe()的变体(仅接受一个参数的变体-onNext消费者回调)将仅在流成功发出项目(在本例中为AuthViewState)时通知消费者。但是，当您可观察的流遇到错误时，RxJava没有处理它的好方法，因为.subscribe()中没有提供错误回调。因此，它引发您在上面遇到的错误。

注意:RxJava具有Observable.subscribe()的许多重载，其中有些接受消费者回调以进行错误处理。

但是，如果您的目标是使Observable始终成功发出AuthViewState，即使遇到错误，您也可以使用Observable.onErrorReturn()(或RxJava提供的类似错误处理功能)。一个示例用法是:

sealed class ViewState {

    object Loading : ViewState()
    data class Success(val username: String) : ViewState()
    data class Error(val error: Throwable) : ViewState()
}

class UserProfileViewModel(
    private val userService: UserService
) {

    fun getViewState(): Observable<ViewState> {

        return Observable
            .merge(
                Observable.just(ViewState.Loading),
                userService
                    .getUserFromApi()
                    .map { user -> ViewState.Success(user.username) }
            )
            .onErrorReturn { error -> ViewState.Error(error) }
    }
}