使用 Paging3 库时更改请求并获取新数据流

Question

我在我的项目中使用了 Jetpack 的 Paging 3 库来处理数据分页。我有一个用例，当用户在搜索请求中更改某些内容（例如添加/删除一些过滤器）时，我必须调用 API 并根据新的搜索请求用新数据重新填充我的列表。但是当我创建新的Pager 实例并将其传递给我的PagingDataAdapter 适配器时，它会抛出：

java.lang.IllegalStateException: Collecting from multiple PagingData concurrently is an illegal operation.

我的实现是这样的：

存储库

class Repository {
    fun getDataStream(request: Request): Flow<PagingData<Response>> {
        return Pager(
            config = PagingConfig(
                pageSize = 10,
                initialLoadSize = 10,
                prefetchDistance = 3
            ),
            initialKey = 1,
            pagingSourceFactory = {
                DataPagingSource(
                    request = request,
                    repository = this
                )
            }
        ).flow
    }

    fun getData(page: Int, request: Request): Result<Response> {
        return remoteDataSource.getData(page, request)
    }
}

数据分页来源

class DataPagingSource(
    private val request: Request,
    private val repository: Repository
) : PagingSource<Int, Response>() {
    override suspend fun load(params: LoadParams<Int>): LoadResult<Int, Response> {
        val page = params.key ?: 1
        // Result is a sealed class which has two derived classes: Success and Error
        return when (val result = repository.getData(page, request)) {
            is Result.Success -> {
                LoadResult.Page(
                    data = result.data,
                    nextKey = page.inc(),
                    prevKey = null
                )
            }
            is Result.Error -> LoadResult.Error(
                result.error
            )
        }
    }
}

视图模型

class SomeViewModel(
    private val repository: Repository
): ViewModel() {
    private val _currentRequest = MutableLiveData<Request>()
   
    val data = _currentRequest
        .switchMap {
            repository
                .getDataStream(it)
                .cachedIn(viewModelScope)
                .asLiveData()
        }

    fun updateRequest(request: Request) {
        _currentRequest.postValue(request)
    }
}

片段

class SomeFragment: Fragment() {
    private lateinit var viewModel: SomeViewModel

    // ...
    
    override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
        // ...
        viewModel.data.observe(
            viewLifecycleOwner,
            Observer {
                lifecycleScope.launch {
                    adapter.submitData(it)
                }
            }
        )
    }
    // ...
}

如果有人帮助解决这个问题，那就太好了。
谢谢

Answer 1

我相信，如果您将 Pager 观察为 LiveData，则需要在 Fragment 中使用 adapter.submitData(lifecycle, data) 方法而不是 adapter.submitData(data)，但我也建议您在 _currentRequest LiveData 上尝试 distinctUntilChanged() 转换以限制创建来自重复请求的多个 PagingData 对象。

submitData(lifecycle: Lifecycle, pagingData: PagingData) 文档

此方法通常在观察 Pager 生成的 RxJava 或 LiveData 流时使用。对于 Flow 支持，使用 submitData 的挂起重载，它通过 CoroutineScope 自动取消，而不是依赖 Lifecycle

https://developer.android.com/reference/kotlin/androidx/paging/PagingDataAdapter#submitdata_1

片段

    viewModel.data.observe(
        viewLifecycleOwner,
        Observer {
            adapter.submitData(lifecycle, it)
        }
    )

视图模型

    val data = _currentRequest
        // Limit duplicate Requests (Request class should implement equals())
        .distinctUntilChanged()  
        .switchMap {
        // ...
    }

Answer 2

我找到了解决问题的方法。对于Paging 库要使用新的请求模型获取数据，您必须更改请求模型，然后在您的PagingDataSource 上调用invalidate。这是一个例子：
在 ViewModel 中，代码更改如下：

class SomeViewModel: ViewModel() {    
    private var _dataPagingSource: DataPagingSource? = null
    private val _requestChannel = ConflatedBroadcastChannel<Request>()
    
    val data = Pager(
        config = PagingConfig(
            pageSize = 10,
            initialLoadSize = 10,
            prefetchDistance = 3
        ),
        initialKey = 1,
        pagingSourceFactory = {
            DataPagingSource(
                request = _requestChannel.value,
                repository = repository
            ).also {
                dataSource = it
            }
        }
    ).flow.cachedIn(viewModelScope).asLiveData()

    // ...
    
    // subscribe on requestChannel and invalidate dataSource each time
    // it emits new value
    init {
        _requestChannel
            .asFlow()
            .onEach { _dataPagingSource?.invalidate() }
            .launchIn(viewModelScope)
    }

    // call this method with the new request 
    fun updateRequest(newRequest: Request) {
        _requestChannel.send(newRequest)
    }
}

Repository 变成这样：

class Repository {
    // we do not need getDataStream method here anymore

    fun getData(page: Int, request: Request): Result<Response> {
        return remoteDataSource.getData(page, request)
    }
}

我不知道是否有其他方法可以做到这一点。如果您知道其他方法，那就太好了。