尝试使用 RxJava 和 Kotlin 在没有 Wifi 连接的情况下从 Internet 获取数据时出错答案

【问题标题】：Error while trying to fetch data from Internet without Wifi connection using RxJava and Kotlin尝试使用 RxJava 和 Kotlin 在没有 Wifi 连接的情况下从 Internet 获取数据时出错
【发布时间】：2018-09-18 10:25:01
【问题描述】：

大家好，我的 BaseActivity 中有以下功能。

override fun <T> subscribeToInternet(observable: Observable<Response<BaseResponse<T>>>, observer: Observer<BaseResponse<T>>) {
    observable.subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnSubscribe { observer.onSubscribe(it) }
            .doOnError {
                Log.d(TAG, it.message)
                observer.onError(it)
            }
            .doOnComplete { observer.onComplete() }
            .doOnNext {
                Log.d(TAG, "${it.body() ?: "no body"}")
                Log.d(TAG, "${it.errorBody()?.string() ?: "no error body"}")
                Log.d(TAG, it.code().toString())
                when {
                    it.code() == 401 -> {
                        view.userUnauthenticated()
                        observer.onNext(BaseResponse(false, "unauthenticated", null))
                        Log.d(TAG, "UNAUTHENTICATED")
                    }
                    it.code() == 423 -> {
                        view.userBlocked()
                        observer.onNext(BaseResponse(false, "blocked", null))
                        Log.d(TAG, "BLOCKED")
                    }
                    it.isSuccessful -> observer.onNext(it.body()!!)
                    it.code() == 429 -> observer.onNext(BaseResponse(false, "Too many attempts", null))
                    it.code() == 400 -> observer.onNext(BaseResponse(false, "Invalid Email or password", null))
                    else -> observer.onNext(BaseResponse(false, "", null))
                }
            }
            .subscribe()
}

如果服务器返回响应，我会在观察者的 onNext() 中处理错误，但是当设备上根本没有 Internet 连接时会出现问题！！它抛出以下异常

at io.reactivex.internal.operators.observable.ObservableDoOnEach$DoOnEachObserver.onError(ObservableDoOnEach.java:119)
    at io.reactivex.internal.observers.DisposableLambdaObserver.onError(DisposableLambdaObserver.java:64)
    at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.checkTerminated(ObservableObserveOn.java:276)
    at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.drainNormal(ObservableObserveOn.java:172)
    at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.run(ObservableObserveOn.java:252)

这就是前面提到的函数的用法

override fun sendLoginRequest(email: String, password: String, fcm_token: String) {
    subscribeToInternet(dataManager.sendLoginRequest(email, password, fcm_token), this)
}

override fun onComplete() {

}

override fun onSubscribe(d: Disposable) {
    DisposableManager.add(d)
}

override fun onNext(t: BaseResponse<LoginData>) {
    if(t.status) {
        Log.d(TAG, "${t.data}")
        dataManager.createLoginSession(t.data!!)
        view.loginSuccess()
    } else {
        Log.d(TAG, t.message)
        view.showError(t.message)
    }
}

override fun onError(e: Throwable) {
    view.showToastError()
    Log.d(TAG, e.message)
}

【问题讨论】：

标签： android kotlin rx-java2

【解决方案1】：

这个问题与您订阅 observable 的方式有关。根据使用subscribe() 而不传递处理错误的操作时的文档，当源抛出异常时，您应该收到OnErrorNotImplementedException - 这是因为使用了来自 RxJavaPlugins 的默认异常处理程序。

要解决该问题，请使用带有onError 参数的重载subscribe 方法之一。例如public final Disposable subscribe(Consumer onNext, Consumer onError)

【讨论】：

感谢您的准确回复。但是 doOnError 还不足以进行错误处理吗？
@TahaMalas，不，这还不够。 doOnError 只允许在发生错误时触发附加操作。根据可观察合同，当错误发生时，您的来源将被终止。因此，当您使用onError 参数调用subscribe 时，您可以确定正确处理源的终止（以防抛出异常）。
非常感谢它的作用就像一个魅力.. 那么我应该移动 doOnNext、doOnError、doOnComplete 和 doOnSubscribe 并在订阅参数中传递它们吗？
我建议你直接在subscribe() 中传递你的观察者并从你的方法中返回Subscription，因为在某些情况下你应该手动触发 Subscription.unsubbscribe() 以释放占用的资源。顺便说一句，您可以在 map 运算符中提取您的 switch/case 块。在我看来，它会更符合 Rx 的习惯。