处理RxJava onErrorReturn和OnErrorNotImplementedException。

Question

所以我想用RxJava在安卓系统中实现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)
            )
        }
    }

VIEWMODEL

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

FRAGMENT

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) }
    }
}