Kotlin协程-返回Flow的Suspend函数将永远运行 [英] Kotlin Coroutines - Suspend function returning a Flow runs forever

问题描述

我正在制作一个支持多个数据检索配置的网络存储库，因此我想将这些配置的逻辑分离为功能.

I am making a network repository that supports multiple data retrieval configs, therefore I want to separate those configs' logic into functions.

但是，我有一个配置，可以按指定的时间间隔连续获取数据.当我将这些值发送到原始Flow时，一切都很好.但是，当我将逻辑带入另一个函数并通过它返回另一个Flow时，它不再关心它的协程范围.即使取消了作用域，它仍会继续获取数据.

However, I have a config that fetches the data continuously at specified intervals. Everything is fine when I emit those values to the original Flow. But when I take the logic into another function and return another Flow through it, it stops caring about its coroutine scope. Even after the scope's cancelation, it keeps on fetching the data.

TLDR:当使用currentCoroutineContext控制循环终止时，返回流的挂起函数将永远运行.

TLDR: Suspend function returning a flow runs forever when currentCoroutineContext is used to control its loop's termination.

我在这里做错了什么?这是我的代码的简化版本:

What am I doing wrong here? Here's the simplified version of my code:

片段调用了viewmodels函数，该函数基本上调用了getData()

 lifecycleScope.launch {
            viewModel.getLatestDataList()
        }

存储库

suspend fun getData(config: MyConfig): Flow<List<Data>>
{
    return flow {

        when (config)
        {
            CONTINUOUS ->
            {
                //It worked fine when fetchContinuously was ingrained to here and emitted directly to the current flow
                //And now it keeps on running eternally
                fetchContinuously().collect { updatedList ->
                    emit(updatedList)
                }
            }
        }
    }
}


//Note logic of this function is greatly reduced to keep the focus on the problem
private suspend fun fetchContinuously(): Flow<List<Data>>
{
    return flow {
        while (currentCoroutineContext().isActive)
        {

            val updatedList = fetchDataListOverNetwork().await()

            if (updatedList != null)
            {
                emit(updatedList)
            }

            delay(refreshIntervalInMs)
        }

        Timber.i("Context is no longer active - terminating the continuous-fetch coroutine")
    }
}


private suspend fun fetchDataListOverNetwork(): Deferred<List<Data>?> =

    withContext(Dispatchers.IO) {

        return@withContext async {

            var list: List<Data>? = null

            try
            {
                val response = apiService.getDataList().execute()

                if (response.isSuccessful && response.body() != null)
                {
                    list = response.body()!!.list
                }
                else
                {
                    Timber.w("Failed to fetch data from the network database. Error body: ${response.errorBody()}, Response body: ${response.body()}")
                }
            }
            catch (e: Exception)
            {
                Timber.w("Exception while trying to fetch data from the network database. Stacktrace: ${e.printStackTrace()}")
            }
            finally
            {
                return@async list
            }
            list //IDE is not smart enough to realize we are already returning no matter what inside of the finally block; therefore, this needs to stay here
        }

    }

推荐答案

我不确定这是否可以解决您的问题，但是您不需要具有返回Flow的挂起函数.您传递的lambda本身就是一个暂停函数:

I am not sure whether this is a solution to your problem, but you do not need to have a suspending function that returns a Flow. The lambda you are passing is a suspending function itself:

fun <T> flow(block: suspend FlowCollector<T>.() -> Unit): Flow<T> (source)

以下是重复使用(GraphQl)查询(简化-没有类型参数)的流程示例:

Here is an example of a flow that repeats a (GraphQl) query (simplified - without type parameters) I am using:

    override fun query(query: Query,
                       updateIntervalMillis: Long): Flow<Result<T>> {
    return flow {
        // this ensures at least one query
        val result: Result<T> = execute(query)
        emit(result)

        while (coroutineContext[Job]?.isActive == true && updateIntervalMillis > 0) {
            delay(updateIntervalMillis)

            val otherResult: Result<T> = execute(query)
            emit(otherResult)
        }
    }
}

这篇关于Kotlin协程-返回Flow的Suspend函数将永远运行的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！