如何实现Flow的emitDataOrLoading？超时自动发送Status.Loading状态

需求说明 #

针对返回Flow<Status<T>>的仓库，实现扩展方法emitDataOrLoading()，当Flow在指定毫秒数内未返回任何值时自动发送Status.Loading状态；收到数据时发送Status.Data状态。

定义Status密封类 #

先确保Status为密封类，约束状态类型：

sealed class Status<out T> {
    object Loading : Status<Nothing>()
    data class Data<out T>(val data: T) : Status<T>()
}

方案一：合并延迟Flow实现 #

通过合并原数据Flow和延迟发送Loading的Flow，实现超时触发逻辑：

fun <T> Flow<Status<T>>.emitDataOrLoading(timeoutMillis: Long = 1000): Flow<Status<T>> = flow {
    // 延迟指定时间后发送Loading的Flow
    val loadingTrigger = flow {
        delay(timeoutMillis)
        emit(Status.Loading)
    }

    // 合并两个Flow，优先传递先发出的状态
    merge(this@emitDataOrLoading, loadingTrigger)
        .distinctUntilChanged() // 过滤重复状态，减少无效UI更新
        .collect { emit(it) }
}

逻辑说明

• 若原Flow在timeoutMillis内发出Status.Data，合并后的Flow会优先推送该数据，后续原Flow的新数据正常传递；
• 若原Flow超时未发数据，loadingTrigger自动发送Status.Loading；
• distinctUntilChanged避免重复状态（如原Flow连续发送相同Data）导致的冗余回调。

方案二：超时异常捕获实现 #

通过timeout操作符捕获超时异常，超时且未收数据时发送Loading，之后继续监听原Flow：

import kotlinx.coroutines.TimeoutCancellationException
import kotlinx.coroutines.flow.onEach

fun <T> Flow<Status<T>>.emitDataOrLoading(timeoutMillis: Long): Flow<Status<T>> = flow {
    var receivedData = false

    try {
        // 尝试收集原Flow并设置超时
        this@emitDataOrLoading.onEach {
            receivedData = true
            emit(it)
        }.timeout(timeoutMillis)
    } catch (e: TimeoutCancellationException) {
        // 仅在未收到任何数据时发送Loading
        if (!receivedData) {
            emit(Status.Loading)
        }
        // 继续收集原Flow后续数据
        this@emitDataOrLoading.collect { emit(it) }
    }
}

逻辑说明

• 超时前收到数据则正常推送，不触发超时逻辑；
• 超时且未收数据时发送Loading，之后持续监听原Flow的后续数据；
• 该方案更严谨，避免原Flow超时后才发数据时，Loading与Data的顺序冲突问题。

使用示例 #

repository.fetchData()
    .emitDataOrLoading(timeoutMillis = 1500) // 设置1.5秒超时阈值
    .collect {
        when (it) {
            is Status.Loading -> {
                // 显示加载指示器
            }
            is Status.Data -> {
                // 处理并展示数据
            }
        }
    }

内容的提问来源于stack exchange，提问作者Johan Paul