Flow<T>.distinctUntilChanged() 不起作用

Question

我试图在 SharedFlow 处于活动状态（subscriptionCount 大于零）时运行协程，并在计数下降时取消。但不知何故，即使像distinctUntilChanged() 这样简单的东西也不能正常工作，我很困惑。

为此，我正在制作这样的“onActive”扩展：

fun <T : Any> MutableSharedFlow<T>.onActive(
    block: suspend CoroutineScope.() -> Unit
): Flow<T> {

    val original = this

    val isActiveFlow: Flow<Boolean> = subscriptionCount
        .map {
            println("Class: Count is $it")
            it > 0
        }
        .distinctUntilChanged()

    return isActiveFlow.flatMapLatest { isActive ->
        println("Class: isActive is $isActive")
        // here would be the code that calls `block`
        // but just this exactly as is, already triggers the error

        original // still emits the original flow, 
                 // that is needed or else subscriptionCount never changes
    }
}

这最初似乎可行，但是在其上运行添加多个订阅者的测试时，将连续多次打印“isActive is true”。为什么distinctUntilChanged() 不起作用？这种重复调用会与编辑区域中的其余逻辑发生冲突。

测试是这样的：

    @Test
    fun `onActive is called only once with multiple subscribers`() = runBlocking {

        val flow = MutableSharedFlow<Int>(
            replay = 2,
            onBufferOverflow = BufferOverflow.DROP_OLDEST
        ).apply {
            repeat(5) { tryEmit(it) }
        }.onActive {

        }

        val jobs = mutableListOf<Job>()
        repeat(3) { count ->
            jobs.add(flow.onEach {
                println("Test:  Listener $count received $it")
            }.launchIn(this))
        }
        delay(100)
        jobs.forEach { it.cancel() }
        jobs.forEach { it.join() }
    }

运行这个输出是：

Class: Count is 0
Class: isActive is false
Class: Count is 1
Class: Count is 1
Class: isActive is true
Class: Count is 2
Class: Count is 2
Class: isActive is true
Class: Count is 3
Test:  Listener 0 received 3
Test:  Listener 0 received 4
Test:  Listener 1 received 3
Test:  Listener 1 received 4
Test:  Listener 2 received 3
Test:  Listener 2 received 4
Class: Count is 2
Class: isActive is true
Class: Count is 3
Class: Count is 3
Class: Count is 3
Test:  Listener 0 received 3
Test:  Listener 0 received 4

那么问题来了，为什么distinctUntilChanged() 不起作用，我该如何解决？

Answer 1

就distinctUntilChanged 而言，您看到的行为似乎实际上是正确的：

• 第一个注册订阅者收集原始 2 个重放元素，起始值为 isActive=false
• 然后isActive 因为第一次订阅而变为真，因此第一个订阅者由于flatMapLatest 重新收集原始流，从而再次获得重放的元素
• 其他 2 个订阅者在 subscriptionCount 已经为非 0 时到达，因此 isActive 对他们保持真实，直到他们被取消

如果您“在有订阅者时”启动的协程是为了在SharedFlow 中生成元素，我宁愿最初定义像channelFlow/callbackFlow 这样的流程，然后使用shareIn 和@ 987654331@ 有这种“有订阅者时运行”的行为。

如果它“只是在旁边”，您可能需要一个外部作用域并单独启动一个协程来监听 sharedFlow.subscribersCount 并启动/停止“sidecar”协程。