时间戳

首页 > 解决方案 > Kotlin Coroutines - 返回流的暂停函数永远运行

android - Kotlin Coroutines - 返回流的暂停函数永远运行

问题描述

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

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

TLDR：当 currentCoroutineContext 用于控制其循环的终止时，返回流的暂停函数将永远运行。

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

调用基本上调用 getData() 的 viewmodels 函数的片段

 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
        }

    }

标签： androidmultithreadingkotlinkotlin-coroutines

解决方案

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)

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)
        }
    }
}

推荐阅读