使用场景

最近在做项目的时候遇到了一个需求。从DataStore获取最新的ID，并根据ID到Room数据库内获取相应的内容。由于DataStore和Room都可以通过Flow获取到最新的数据更新，因此我设想是否能通过数据驱动界面的方式，抛开以往的通知界面更新方法，直接对数据源的变动进行监听并设置界面。
需要实现以上的需求就需要对DataStore的ID进行监听的同时，还要对Room数据库内相应ID的数据进行同步监听，保证两个数据源的数据的变化都可以监听到。

Tips: 以下部分大都是问题发现与解决的步骤，符合从认识到实践的二次飞跃过程，因此想要直接看最终可用解决方案的可以直接翻到最后。

初步实现

最简单的设想是在transform方法中进行嵌套实现，代码如下：

// 使用时转换为LiveData后，在Activity中进行数据变化监听
val data = flow1.transform { id ->
    val flow2 = getFlow2(id)
    emitAll(flow2)
}.asLiveData()

实际使用以上代码时却出了大问题，A界面获取数据并初始化视图成功，但是在B界面修改数据后返回A界面后，数据更新并没有被监听到。以上的数据不更新情况还是不确定的，因为如果在B界面修改完数据后等待一段时间再返回A界面，数据更新就能够被监听到了。

初步解决方案

无法监听数据变化的情况让我抓狂了一阵子，最终“修改完数据后等待一段时间能够被监听到”还是在asLiveData方法的源码中找到了原因。源码如下：

// CoroutineLiveData.kt
internal const val DEFAULT_TIMEOUT = 5000L

// FlowLiveData.kt
@JvmOverloads
fun <T> Flow<T>.asLiveData(
    context: CoroutineContext = EmptyCoroutineContext,
    timeoutInMs: Long = DEFAULT_TIMEOUT
): LiveData<T> = liveData(context, timeoutInMs) {
    collect {
        emit(it)
    }
}

好家伙，这里有个DEFAULT_TIMEOUT的超时设定。掐指一算确实只需要在B界面等待5秒后再返回A界面，仅能够成功更新界面了。因此在使用asLiveData(timeoutInMs=0)后问题貌似就解决了。那么此处的超时时间是用来做什么的呢？继续看源码：

// CoroutineLiveData.kt
internal class BlockRunner<T>(
    private val liveData: CoroutineLiveData<T>,
    private val block: Block<T>,
    private val timeoutInMs: Long,
    private val scope: CoroutineScope,
    private val onDone: () -> Unit
) {
    ··· ···
    @MainThread
    fun cancel() {
        ··· ···
        cancellationJob = scope.launch(Dispatchers.Main.immediate) {
            delay(timeoutInMs)
            if (!liveData.hasActiveObservers()) {
                // one last check on active observers to avoid any race condition between starting
                // a running coroutine and cancelation
                runningJob?.cancel()
                runningJob = null
            }
        }
    }
}

首先asLiveData中的调用的Flow的collect方法是在一个lambda函数中的，CoroutineLiveData使用BlockRunner作为collect方法所在的lambda函数的运行器，监听LifeCycle的变化，以启动与取消collect方法。
而超时时间设置的原因，简单理解一下就是，由于担心界面变化太快（例如屏幕旋转）导致BlockRunner重复的启动与取消带来的任何竞争问题，所以此处等待指定时间后再检查是否还有活动的观察器，并最后取消collect的运行。
显然，超时时间并不是造成无法收到更新数据的主要原因，因为设置这个超时时间的出发点以及功能显然是好的。
但是以上源码却给我带来了启示，因为等待时间超过5秒后取消了对于collect方法的执行，后面再重新执行collect方法就可以获取到更新的数据了。

查找根本原因

此时，我突然想起collect方法是一个suspend方法，会挂起当前运行的协程，而对数据变化进行监听的话，collect方法显然会挂起当前运行的协程。会不会就是collect方法带来的协程挂起导致了以上的问题？开始试验：

val flow1 = flow {
    emit("F1-A")
    delay(2000L)
    emit("F1-B")
    delay(2000L)
    emit("F1-C")
    delay(10000000L)
}.transform {
    val flow2 = flow {
        emit("$it-F2")
        delay(10000000L)
    }
    emitAll(flow2)
}
runBlocking {
    launch {
        flow1.collect {
            println("${System.currentTimeMillis()} 1 $it")
        }
    }
    delay(3000L)
    launch {
        flow1.collect {
            println("${System.currentTimeMillis()} 2 $it")
        }
    }
}

delay(2000L)表示数据的更新间隔，delay(10000000L)模拟长时间不关闭的监听，两个独立的collect协程表示监听的先后。最后得到的结果如下：

1608255907350 1 F1-A-F2
1608255910333 2 F1-A-F2
// 后面就卡住了

实锤了collect方法的嵌套导致协程挂起，导致了无法收到更新数据的问题。

寻找解决方案

此时我突然想到了combine方法，同样都是将两个Flow合并在一起，为什么combine方法就没有问题？查找后源码如下：

@PublishedApi
internal suspend fun <R, T> FlowCollector<R>.combineInternal(
    flows: Array<out Flow<T>>,
    arrayFactory: () -> Array<T?>?, // Array factory is required to workaround array typing on JVM
    transform: suspend FlowCollector<R>.(Array<T>) -> Unit
): Unit = flowScope { // flow scope so any cancellation within the source flow will cancel the whole scope
    ··· ···
        // Coroutine per flow that keeps track of its value and sends result to downstream
        launch {
            try {
                flows[i].collect { value ->
                    resultChannel.send(Update(i, value))
                    yield() // Emulate fairness, giving each flow chance to emit
                }
            }
··· ···

恍然大悟，原来combine方法是将每个对collect的监听放到单独的Scope中才能够实现对多个flow的监听与同步更新。
那么这个flowScope以及combineInternal能不能改造后为我所用呢？不行，因为全都是内部方法，所以无法调用。那么只能模仿其写一个了。

fun <T1, T2> Flow<T1>.dependCombine(transform: (T1) -> Flow<T2>) = flow {
    var oldJob: Job? = null
    collect {
        oldJob?.cancel()
        oldJob = GlobalScope.launch {
            emitAll(transform(it))
        }
    }
}

然而，最终却出错了？！！

Exception in thread "DefaultDispatcher-worker-1 @coroutine#3" java.lang.IllegalStateException: Flow invariant is violated:
        Emission from another coroutine is detected.
        Child of "coroutine#3":StandaloneCoroutine{Active}@5bbc3ad9, expected child of "coroutine#2":StandaloneCoroutine{Active}@41f88e7d.
        FlowCollector is not thread-safe and concurrent emissions are prohibited.
        To mitigate this restriction please use 'channelFlow' builder instead of 'flow'

好吧，flow中不能够运行其他的协程，所以只能使用官方建议的channelFlow了。

最终解决方案

直接放代码：

fun <T1, T2> Flow<T1>.combine(transform: suspend (T1) -> Flow<T2>): Flow<T2> =
    combineTransform(transform) { _, t2 ->
        t2
    }

fun <T1, T2, R> Flow<T1>.combineTransform(combineTransform: suspend (T1) -> Flow<T2>, transform: suspend (T1, T2) -> R): Flow<R> =
    channelFlow {
        var oldJob: Job? = null
        collect { t1 ->
            oldJob?.cancel()
            oldJob = launch {
                combineTransform(t1).collect { t2 ->
                    send(transform(t1, t2))
                }
            }
        }
        awaitClose {
            oldJob?.cancel()
        }
    }

通过channelFlow即可正确的解决问题了
以上只是为本人的解决方案，并不代表只能这么解决，如果有更好的方法望不吝赐教。

One more thing

对于一个需要在多处被collect的Flow，与shareIn方法一起使用效果更佳。

---

Made By XFY9326