Flowable 的 onErrorResumeNext、networkOnMainThread 出错

Question

我有以下 rxJava 链：

 override fun combineLocationToPlace(req: Flowable<Place>): Flowable<Place> {
        var combinedFlowable = Flowable
                .combineLatest(
                        req,
                        getLastLocation().lastOrError().toFlowable(),
                        BiFunction<Place, Location, Place> { t1, location ->
                            Timber.w("FIRSTINIT - Retrieved location $location")
                            var placeLocation = Location(t1.placeName)
                            placeLocation.latitude = t1.latitude
                            placeLocation.longitude = t1.longitude
                            t1.distance = location.distanceTo(placeLocation)
                            t1
                        })


        return combinedFlowable
                .onErrorResumeNext { t: Throwable ->
                    Timber.w(t, "FIRSTINIT - Could not retrieve location for place (${t.message}) returning original request")
                    req
                }
                .doOnError {
                    Timber.w("FIRSTINIT - did detect the error here...")
                }

        return combinedFlowable
    }

简而言之，我正在从本地数据库（一个地方）检索一些数据，我想将其与 GPS 的最新位置相结合：

 override fun getLastLocation(requestIfEmpty: Boolean): Observable<Location> {
        var lastLocation = locationProvider.lastKnownLocation
                .doOnNext {
                    Timber.w("Got location $it from last one")
                }
                .doOnComplete {
                    Timber.w("did i get a location?")
                }

        if (requestIfEmpty) {
            Timber.w("Switching to request of location")
            lastLocation = lastLocation.switchIfEmpty(requestLocation())
        }

        return lastLocation.doOnNext {
            Timber.w("Got something!")
            location = it
        }


    }

但我想考虑用户没有最后位置的场景，因此该行：

return combinedFlowable
                    .onErrorResumeNext { t: Throwable ->
                        Timber.w(t, "FIRSTINIT - Could not retrieve location for place (${t.message}) returning original request")
                        req
                    }
                    .doOnError {
                        Timber.w("FIRSTINIT - did detect the error here...")
                    }

它试图捕获任何错误并仅使用原始请求重试而不将其与任何内容组合。我这样称呼这段代码：

fun getPlace(placeId: String) {
        locationManager.combineLocationToPlace(placesRepository.getPlace(placeId))
                .onErrorResumeNext { t: Throwable ->
                    Timber.e(t, "Error resuming next! ")
                    placesRepository.getPlace(placeId)
                }.subscribeOn(schedulerProvider.io()).observeOn(schedulerProvider.ui())
                .subscribeBy(
                        onNext = {
                            place.value = Result.success(it)
                        },
                        onError = {
                            Timber.e("ERROR! $it")
                            place.value = Result.failure(it)
                        }
                )
                .addTo(disposables)

    }

但是，当没有位置抛出 NoSuchElementException 时，我的 flowable 切换到原始请求，然后在执行它时我得到一个 NetworkOnMainThread 异常。这个请求不应该在我放在那里的scheduler.io()上执行吗（因为我之前放了代码）？

如果您想知道，schedulerProvider.io() 转换为：

Schedulers.io()

获取地点：

  /**
     * Retrieves a single place from database
     */
      override fun getPlace(id: String): Flowable<Place> {
        return Flowable.merge(placesDao.getPlace(id),
                refreshPlace(id).toFlowable())
    }

    /**
     * Triggers a refreshPlace update on the db, useful when changing stuff associated with the place
     * itself indirectly (e.g. an experience)
     */
    private fun refreshPlace(id: String): Single<Place> {
        return from(placesApi.getPlace(id))
                .doOnSuccess {
                    placesDao.savePlace(it)
                }
    }

Answer 1

为了确保您的网络发生在主线程之外，请将其显式发送到后台线程。

使用 Rx 调度程序类中的 IO、新线程或计算调度程序：

subscribeOn(Schedulers.computation())

如果您不想这样做（或者如果您认为它应该已经在后台线程上并且只想调试），您可以按如下方式记录线程信息：

Thread.currentThread().getName()

当您使用 Schedulers.trampoline() 时，如果您有不同的调度程序用于观察和订阅，这对于跟踪正在发生的事情特别有用，例如您的示例：

.subscribeOn(schedulerProvider.io()).observeOn(schedulerProvider.ui())

Answer 2

你不能使用 onErrorResumeNext 来实现这个功能。您应该使用 retryWhen 运算符。

也许这篇文章对你有用。

https://medium.com/@v.danylo/server-polling-and-retrying-failed-operations-with-retrofit-and-rxjava-8bcc7e641a5a

此代码使用退避功能轮询服务器，直到收到不同于 204 的代码。 也许您可以使用 retryWhen 而不是 repeatWhen 来适应您的需求。

  fun pollServerWithBackoff(videoId: String, maxAttempts: Int, delay: Int): Flowable<Response<ResponseBody>> {
        return api.download(videoId)
                .subscribeOn(Schedulers.io())
                .repeatWhen {
                    it
                            .zipWith(Flowable.range(1, maxAttempts),
                                    BiFunction { _: Any?, attempt: Int -> attempt })
                            .flatMap {

                                Flowable.timer((it * delay).toLong(), TimeUnit.SECONDS);
                            }

                }
                .takeUntil({

                    it.code() != 204
                })
                .filter {

                    it.code() != 204
                }
                .map{
                    if(it.code() in 200..300)
                        it
                    else
                        throw IOException(it.errorBody()?.toString() ?: "Unkown Error")
                }
    }