Android Paging 库不适用于异步请求

Question

使用 Retrofit 进行异步网络调用时，Android 分页库不起作用。我在 Github 上使用 Google 的架构组件示例代码，并根据我的需要对其进行了修改。

我之前遇到过同样的问题，但由于用例允许，我通过同步调用解决了这个问题。但是在当前场景中，需要多个网络调用，并且数据存储库会返回组合结果。为此，我正在使用 RxJava。

最初它似乎是一个多线程问题，但this 的答案却暗示了其他问题。在主线程上观察 RxJava 调用也不起作用。

我在下面添加了相关代码。我在调试时进入了callback.onResult，一切都按预期工作。但最终它并没有通知Recycler View Adapter。

View Model snippet:

open fun search(query : String, init : Boolean = false) : Boolean {
    return if(query == searchQuery.value && !init) {
        false
    } else {
        searchQuery.value = query
        true
    }
}

fun refresh() {
    listing.value?.refresh?.invoke()
}

var listing : LiveData<ListingState<T>> = Transformations.map(searchQuery) {
    getList() // Returns the Listing State from the Repo snippet added below.
}

Repository snippet:

val dataSourceFactory = EvaluationCandidateDataSourceFactory(queryParams,
            Executors.newFixedThreadPool(5) )

    val pagelistConfig = PagedList.Config.Builder()
            .setEnablePlaceholders(true)
            .setInitialLoadSizeHint(5)
            .setPageSize(25)
            .setPrefetchDistance(25).build()

    val pagedList = LivePagedListBuilder<Int, PC>(
            dataSourceFactory, pagelistConfig)
            .setFetchExecutor(Executors.newFixedThreadPool(5)).build()

    val refreshState = Transformations.switchMap(dataSourceFactory.dataSource) {
        it.initialState
    }

    return ListingState(
            pagedList = pagedList,
            pagingState = Transformations.switchMap(dataSourceFactory.dataSource) {
                it.pagingState
            },
            refreshState = refreshState,
            refresh = {
                dataSourceFactory.dataSource.value?.invalidate()
            },

            retry = {
                dataSourceFactory.dataSource.value?.retryAllFailed()
            }
    )

Data Source snippet :

override fun loadInitial(params: LoadInitialParams<Int>, callback: LoadInitialCallback<Int, PC>) {
    try {
        queryMap = if (queryMap == null) {
            hashMapOf("page" to FIRST_PAGE)
        } else {
            queryMap.apply { this!!["page"] = FIRST_PAGE }
        }

        initialState.postValue(DataSourceState.LOADING)
        pagingState.postValue(DataSourceState.LOADING)
        val disposable : Disposable = aCRepositoryI.getAssignedAC(queryMap)
                .subscribeOn(AndroidSchedulers.mainThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe({
                    if(it.success) {
                        // remove possible retries on success
                        retry = null

                        val nextPage = it.responseHeader?.let { getNextPage(it, FIRST_PAGE) } ?: run { null }
                        val previousPage = getPreviousPage(FIRST_PAGE)
                        callback.onResult(it.response.pcList, previousPage, nextPage)
                        initialState.postValue(DataSourceState.SUCCESS)
                        pagingState.postValue(DataSourceState.SUCCESS)
                    } else {
                        // let the subscriber decide whether to retry or not
                        retry = {
                            loadInitial(params, callback)
                        }
                        initialState.postValue(DataSourceState.failure(it.networkError.message))
                        pagingState.postValue(DataSourceState.failure(it.networkError.message))
                        Timber.e(it.networkError.message)
                    }
                }, {
                    retry = {
                        loadInitial(params, callback)
                    }
                    initialState.postValue(DataSourceState.failure(it.localizedMessage))
                    pagingState.postValue(DataSourceState.failure(it.localizedMessage))
                })
    } catch (ex : Exception) {
        retry = {
            loadInitial(params, callback)
        }
        initialState.postValue(DataSourceState.failure(ex.localizedMessage))
        pagingState.postValue(DataSourceState.failure(ex.localizedMessage))
        Timber.e(ex)
    }
}

有人可以告诉这里的问题是什么。上面有一个similar issue，但是推荐使用同步调用。我们如何使用异步调用或使用 RxJava 来做到这一点。

Answer 1

我不明白你为什么要转到主线程。 DataSource 中的加载方法在后台线程中运行。这意味着你可以在这个线程上进行同步工作而不会阻塞主线程，这意味着你可以想出一个没有 RxJava 的解决方案。比如：

override fun loadInitial(params: LoadInitialParams<Int>, callback: LoadInitialCallback<Int, PC>) {
    try {
        val result = repository.fetchData(..)
        // post result values and call the callback
    catch (e: Exception) {
        // post error values and log and save the retry
    }
}

然后在您的存储库中您可以这样做，因为我们不在主线程上。

fun fetchData(...) {
    val response = myRetrofitService.someBackendCall(..).execute()
    response.result?.let {
        return mapResponse(it)
    } ?: throw HttpException(response.error)
}

我可能搞砸了语法，但我希望你明白这一点。没有回调，没有订阅/观察，但简单直接的代码。

此外，如果您在 loadInitial(..) 方法中开始线程化，您的初始列表将为空，因此同步执行操作也可以避免看到空列表。