问使用RxJava和Kotlin在没有Wifi连接的情况下尝试从互联网获取数据时出错

EN

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

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