Handling RxJava onErrorReturn and OnErrorNotImplementedExcep

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

视图模型

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

您在此处收到的错误是由于您在Fragment中调用.subscribe()。.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) }
}
}