android - 使用 RxJava2 和 Retrofit2 Mosby MVI 的 OnErrorNotImplementedException

Question

尽管在下游处理了错误(？)，但我还是抛出了 OnErrorNotImplementedException 并且应用程序崩溃了。

异常

E/AndroidRuntime: FATAL EXCEPTION: RxCachedThreadScheduler-1
Process: pl.netlandgroup.smartsab, PID: 9920
io.reactivex.exceptions.OnErrorNotImplementedException: HTTP 401 Unauthorized
    at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:704)
    at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:701)
    at io.reactivex.internal.observers.LambdaObserver.onError(LambdaObserver.java:74)
    at io.reactivex.internal.operators.observable.ObservableDoOnEach$DoOnEachObserver.onError(ObservableDoOnEach.java:119)
    at io.reactivex.internal.operators.observable.ObservableDoOnEach$DoOnEachObserver.onError(ObservableDoOnEach.java:119)
    at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeOnObserver.onError(ObservableSubscribeOn.java:63)
    at retrofit2.adapter.rxjava2.BodyObservable$BodyObserver.onNext(BodyObservable.java:56)
    at retrofit2.adapter.rxjava2.BodyObservable$BodyObserver.onNext(BodyObservable.java:37)
    at retrofit2.adapter.rxjava2.CallExecuteObservable.subscribeActual(CallExecuteObservable.java:43)
    at io.reactivex.Observable.subscribe(Observable.java:10838)
    at retrofit2.adapter.rxjava2.BodyObservable.subscribeActual(BodyObservable.java:34)
    at io.reactivex.Observable.subscribe(Observable.java:10838)
    at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeTask.run(ObservableSubscribeOn.java:96)
    at io.reactivex.Scheduler$DisposeTask.run(Scheduler.java:452)
    at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:61)
    at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:52)
    at java.util.concurrent.FutureTask.run(FutureTask.java:237)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:272)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1133)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:607)
    at java.lang.Thread.run(Thread.java:761)
 Caused by: retrofit2.adapter.rxjava2.HttpException: HTTP 401 Unauthorized
    at retrofit2.adapter.rxjava2.BodyObservable$BodyObserver.onNext(BodyObservable.java:54)
    at retrofit2.adapter.rxjava2.BodyObservable$BodyObserver.onNext(BodyObservable.java:37) 
    at retrofit2.adapter.rxjava2.CallExecuteObservable.subscribeActual(CallExecuteObservable.java:43) 
    at io.reactivex.Observable.subscribe(Observable.java:10838) 
    at retrofit2.adapter.rxjava2.BodyObservable.subscribeActual(BodyObservable.java:34) 
    at io.reactivex.Observable.subscribe(Observable.java:10838) 
    at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeTask.run(ObservableSubscribeOn.java:96) 
    at io.reactivex.Scheduler$DisposeTask.run(Scheduler.java:452) 
    at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:61) 
    at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:52) 
    at java.util.concurrent.FutureTask.run(FutureTask.java:237) 
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:272) 
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1133) 
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:607) 
    at java.lang.Thread.run(Thread.java:761) 

改造存储库:

class RetrofitRepository (retrofit: Retrofit) {

    val apiService: ApiService = retrofit.create(ApiService::class.java)
    var size: Int = 0

    fun getMapResponse(pageIndex: Int = 0): Observable<MapResponse> {
        return apiService.getMapResponse(pageIndex = pageIndex)
                .doOnError { Log.d("error", it.message) }
                .doOnNext { Log.d("currThread", Thread.currentThread().name) }

    }

    fun getItemsFormResponses(): Observable<List<Item>> {
        val list = mutableListOf<Observable<List<Item>>>()
        val resp0 = getMapResponse()
        resp0.subscribe { size = it.totalCount }
        var accum = 0
        do {
            list.add(getMapResponse(accum).map { it.items })
            accum++
        } while (list.size*200 < size)
        return Observable.merge(list)
    }
}

Interactor 观察到了这个结果:

class MapInteractor @Inject constructor(private val repository: RetrofitRepository) {

    fun getMapItems(): Observable<MapViewState> {
        return repository.getItemsFormResponses()
                .map {
                    if(it.isEmpty()) {
                        return@map MapViewState.EmptyResult()
                    } else {
                        val mapItems = it.map { it.toMapItem() }
                        return@map MapViewState.MapResult(mapItems)
                    }
                }
                .doOnNext { Log.d("currThread", Thread.currentThread().name) }
                .startWith(MapViewState.Loading())
                .onErrorReturn { MapViewState.Error(it) }
    }
}

onErrorReturn { MapViewState.Error(it) } 正确发出(就在应用程序崩溃之前我可以看到屏幕上呈现的正确内容)。如何在保持 MVI 架构的同时避免此异常？

编辑

answer provided by dimsuz是正确的解决方案，尽管要实现合并和返回一个包含所有项目的 Observable，它必须修改为:

fun getMapItems(): Observable<List<Item>> {
    return getMapResponse().map {
        val size = it.totalCount
        val list = mutableListOf<Observable<List<Item>>>()
        var accum = 0
        do {
            list.add(getMapResponse(accum++).map { it.items })
        } while (list.size*200 < size)
        return@map list.zip { it.flatten() }
    }.mergeAll()
}

Answer 1

解决方案已在上一个答案中显示，但是您获得“OnErrorNotImplementedException”的确切原因是您可能仅订阅了“onSuccess”消费者。

您应该添加另一个消费者来处理错误事件。

getMapResponse()
            .subscribe ( 
                     { size = it.totalCount },
                     {/* Do something with the error*/}
            )

'doOnError' 不会阻止您获得异常。