使用 Kotlin 协程的 NetworkBoundResource答案

【问题标题】：NetworkBoundResource with Kotlin coroutines使用 Kotlin 协程的 NetworkBoundResource
【发布时间】：2019-10-21 12:30:32
【问题描述】：

您对如何使用 NetworkBoundResource 和 Kotlin 协程实现存储库模式有任何想法吗？我知道我们可以使用 GlobalScope 启动协程，但这可能会导致协程泄漏。我想将 viewModelScope 作为参数传递，但在实现方面有点棘手（因为我的存储库不知道任何 ViewModel 的 CoroutineScope）。

abstract class NetworkBoundResource<ResultType, RequestType>
@MainThread constructor(
    private val coroutineScope: CoroutineScope
) {

    private val result = MediatorLiveData<Resource<ResultType>>()

    init {
        result.value = Resource.loading(null)
        @Suppress("LeakingThis")
        val dbSource = loadFromDb()
        result.addSource(dbSource) { data ->
            result.removeSource(dbSource)
            if (shouldFetch(data)) {
                fetchFromNetwork(dbSource)
            } else {
                result.addSource(dbSource) { newData ->
                    setValue(Resource.success(newData))
                }
            }
        }
    }

    @MainThread
    private fun setValue(newValue: Resource<ResultType>) {
        if (result.value != newValue) {
            result.value = newValue
        }
    }

    private fun fetchFromNetwork(dbSource: LiveData<ResultType>) {
        val apiResponse = createCall()
        result.addSource(dbSource) { newData ->
            setValue(Resource.loading(newData))
        }
        result.addSource(apiResponse) { response ->
            result.removeSource(apiResponse)
            result.removeSource(dbSource)
            when (response) {
                is ApiSuccessResponse -> {
                    coroutineScope.launch(Dispatchers.IO) {
                        saveCallResult(processResponse(response))

                        withContext(Dispatchers.Main) {
                            result.addSource(loadFromDb()) { newData ->
                                setValue(Resource.success(newData))
                            }
                        }
                    }
                }

                is ApiEmptyResponse -> {
                    coroutineScope.launch(Dispatchers.Main) {
                        result.addSource(loadFromDb()) { newData ->
                            setValue(Resource.success(newData))
                        }
                    }
                }

                is ApiErrorResponse -> {
                    onFetchFailed()
                    result.addSource(dbSource) { newData ->
                        setValue(Resource.error(response.errorMessage, newData))
                    }
                }
            }
        }
    }
}

【问题讨论】：

恕我直言，存储库应公开 suspend 函数，或返回 Channel/Flow 对象，具体取决于 API 的性质。然后在视图模型中设置实际的协程。 LiveData 由视图模型引入，而不是存储库。
@CommonsWare 所以你建议重写 NetworkBoundResource 以返回实际数据（或 Resource），而不在其中和存储库中使用 LiveData？
你是想要使用NetworkBoundResource的人。我的 cmets 更通用：恕我直言，Kotlin 存储库实现应该公开与协程相关的 API。
我很想感谢大家帮助我解决这个问题和各种答案。感谢@CommonsWare，他的提示帮助我（再次）改进了我的代码
我会更多地将其表述为个人喜好。 LiveData 缺乏 RxJava 或 Kotlin 协程的功能。 LiveData 非常适合与活动或片段的“最后一英里”通信，它的设计考虑了这一点。对于小型应用程序，如果您想跳过存储库，只需 ViewModel 直接与 RoomDatabase 交谈，LiveData 就可以了。

标签： android kotlin kotlin-coroutines

【解决方案1】：

更新（2020-05-27）：

一种比我之前的示例更符合 Kotlin 语言的方式，使用 Flow API，并借用 Juan 的答案，可以表示为一个独立的函数，如下所示：

inline fun <ResultType, RequestType> networkBoundResource(
    crossinline query: () -> Flow<ResultType>,
    crossinline fetch: suspend () -> RequestType,
    crossinline saveFetchResult: suspend (RequestType) -> Unit,
    crossinline onFetchFailed: (Throwable) -> Unit = { Unit },
    crossinline shouldFetch: (ResultType) -> Boolean = { true }
) = flow<Resource<ResultType>> {
    emit(Resource.Loading(null))
    val data = query().first()

    val flow = if (shouldFetch(data)) {
        emit(Resource.Loading(data))

        try {
            saveFetchResult(fetch())
            query().map { Resource.Success(it) }
        } catch (throwable: Throwable) {
            onFetchFailed(throwable)
            query().map { Resource.Error(throwable, it) }
        }
    } else {
        query().map { Resource.Success(it) }
    }

    emitAll(flow)
}

上面的代码可以从一个类中调用，例如一个存储库，如下所示：

fun getItems(request: MyRequest): Flow<Resource<List<MyItem>>> {
    return networkBoundResource(
        query = { dao.queryAll() },
        fetch = { retrofitService.getItems(request) },
        saveFetchResult = { items -> dao.insert(items) }
    )
}

原答案：

这就是我使用livedata-ktx 工件的方式；无需传入任何 CoroutineScope。该类也只使用一种类型而不是两种类型（例如 ResultType/RequestType），因为我最终总是在其他地方使用适配器来映射它们。

import androidx.lifecycle.LiveData
import androidx.lifecycle.liveData
import androidx.lifecycle.map
import nihk.core.Resource

// Adapted from: https://developer.android.com/topic/libraries/architecture/coroutines
abstract class NetworkBoundResource<T> {

    fun asLiveData() = liveData<Resource<T>> {
        emit(Resource.Loading(null))

        if (shouldFetch(query())) {
            val disposable = emitSource(queryObservable().map { Resource.Loading(it) })

            try {
                val fetchedData = fetch()
                // Stop the previous emission to avoid dispatching the saveCallResult as `Resource.Loading`.
                disposable.dispose()
                saveFetchResult(fetchedData)
                // Re-establish the emission as `Resource.Success`.
                emitSource(queryObservable().map { Resource.Success(it) })
            } catch (e: Exception) {
                onFetchFailed(e)
                emitSource(queryObservable().map { Resource.Error(e, it) })
            }
        } else {
            emitSource(queryObservable().map { Resource.Success(it) })
        }
    }

    abstract suspend fun query(): T
    abstract fun queryObservable(): LiveData<T>
    abstract suspend fun fetch(): T
    abstract suspend fun saveFetchResult(data: T)
    open fun onFetchFailed(exception: Exception) = Unit
    open fun shouldFetch(data: T) = true
}

就像@CommonsWare 在 cmets 中所说，不过，最好只公开一个Flow<T>。这是我尝试过的方法。请注意，我没有在生产中使用此代码，所以买家要小心。

import kotlinx.coroutines.flow.*
import nihk.core.Resource

abstract class NetworkBoundResource<T> {

    fun asFlow(): Flow<Resource<T>> = flow {
        val flow = query()
            .onStart { emit(Resource.Loading<T>(null)) }
            .flatMapConcat { data ->
                if (shouldFetch(data)) {
                    emit(Resource.Loading(data))

                    try {
                        saveFetchResult(fetch())
                        query().map { Resource.Success(it) }
                    } catch (throwable: Throwable) {
                        onFetchFailed(throwable)
                        query().map { Resource.Error(throwable, it) }
                    }
                } else {
                    query().map { Resource.Success(it) }
                }
            }

        emitAll(flow)
    }

    abstract fun query(): Flow<T>
    abstract suspend fun fetch(): T
    abstract suspend fun saveFetchResult(data: T)
    open fun onFetchFailed(throwable: Throwable) = Unit
    open fun shouldFetch(data: T) = true
}

【讨论】：

Flow 代码会在数据库中的数据发生变化时再次发出网络请求，我发布了一个新的答案来说明如何处理它
在我的测试中，我在flatMapConcat 块内和query().map { Resource.Success(it) } 块内放置了一个断点，然后将一个项目插入到数据库中。只有后一个断点被击中。换句话说，当数据库中的数据发生变化时，网络请求不会再次发出。
如果你在这里设置断点if (shouldFetch(data))你会看到它被调用了两次。第一次从数据库获取结果，第二次调用saveFetchResult(fetch())
你说得对，我误解了代码。 flatMapConcat 将返回一个要观察的新流，因此将不再调用初始流。两个答案的行为方式相同，因此我将保留我的不同方式来实现它。很抱歉造成混乱，感谢您的解释！
@FlorianWalther 关于你的第二个问题，我在这里给出了答案：stackoverflow.com/a/65984833/2997980

【解决方案2】：

@N1hk 回答正确，这只是不使用flatMapConcat 运算符的不同实现（此时标记为FlowPreview）

@FlowPreview
@ExperimentalCoroutinesApi
abstract class NetworkBoundResource<ResultType, RequestType> {

    fun asFlow() = flow {
        emit(Resource.loading(null))

        val dbValue = loadFromDb().first()
        if (shouldFetch(dbValue)) {
            emit(Resource.loading(dbValue))
            when (val apiResponse = fetchFromNetwork()) {
                is ApiSuccessResponse -> {
                    saveNetworkResult(processResponse(apiResponse))
                    emitAll(loadFromDb().map { Resource.success(it) })
                }
                is ApiErrorResponse -> {
                    onFetchFailed()
                    emitAll(loadFromDb().map { Resource.error(apiResponse.errorMessage, it) })
                }
            }
        } else {
            emitAll(loadFromDb().map { Resource.success(it) })
        }
    }

    protected open fun onFetchFailed() {
        // Implement in sub-classes to handle errors
    }

    @WorkerThread
    protected open fun processResponse(response: ApiSuccessResponse<RequestType>) = response.body

    @WorkerThread
    protected abstract suspend fun saveNetworkResult(item: RequestType)

    @MainThread
    protected abstract fun shouldFetch(data: ResultType?): Boolean

    @MainThread
    protected abstract fun loadFromDb(): Flow<ResultType>

    @MainThread
    protected abstract suspend fun fetchFromNetwork(): ApiResponse<RequestType>
}

【讨论】：

在 ApiErrorResponse 情况下发出 Resource.error 不是更好吗？
改造服务返回类型应该是什么？
@MahmoodAli suspend fun someData(@Query/@Path): ApiResponse> ... 根据你的数据管理它
这种方法在加载过程中不会发出数据库更新，这有时是必要的

【解决方案3】：

我是 Kotlin 协程的新手。我这周刚遇到这个问题。

我认为，如果您使用上面帖子中提到的存储库模式，我认为可以随意将 CoroutineScope 传递到 NetworkBoundResource。 CoroutineScope可以是Repository中函数的参数之一，返回一个LiveData，如：

suspend fun getData(scope: CoroutineScope): LiveDate<T>

在 ViewModel 中调用 getData() 时，将内置范围 viewmodelscope 作为 CoroutineScope 传递，因此 NetworkBoundResource 将在 NetworkBoundResource 内工作strong>viewmodelscope 并与 Viewmodel 的生命周期绑定。 NetworkBoundResource 中的协程将在 ViewModel 死亡时取消，这将是一个好处。

要使用内置范围 viewmodelscope，请不要忘记在 build.gradle 中添加以下内容。

implementation 'androidx.lifecycle:lifecycle-viewmodel-ktx:2.2.0-alpha01'

【讨论】：