Блог

Я пытаюсь запустить сопрограмму, пока 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() не работает и как я могу это исправить?

Кажется, поведение, которое вы видите, на самом деле правильное, насколько distinctUntilChanged это касается:

• первый зарегистрированный подписчик собирает исходные 2 воспроизведенных элемента с начальным isActive=false значением
• затем isActive становится true из-за этого первого susbcription, так что первый подписчик вспоминает исходный поток из-за flatMapLatest и, таким образом, снова получает воспроизводимые элементы
• остальные 2 подписчика поступают, когда значение subscriptionCount уже не равно 0, поэтому isActive оно остается для них верным до тех пор, пока они не будут отменены

Если сопрограмма, которую вы запускаете «пока есть подписчики», предназначена для создания элементов в SharedFlow , я бы предпочел сначала определить поток как channelFlow / callbackFlow , а затем использовать shareIn с SharingStarted.WhileSubscribed , чтобы это поведение «запускалось при наличии подписчиков».

Если это «просто на стороне», вам, вероятно, нужна внешняя область и просто запустите сопрограмму отдельно, чтобы прослушать sharedFlow.subscribersCount и запустить / остановить сопрограмму «коляски».