onEach change le dispatcher dans StateFlow (coroutines kotlin)

Question

Imaginez le scénario de test autonome suivant

@Test
fun `stateFlow in GlobalScope`() = runBlockingTest {

    suspend fun makeHeavyRequest(): String {
        return "heavy result"
    }

    val flow1 = flowOf(Unit)
        .map { makeHeavyRequest() }
        .onEach { logThread("1: before flowOn") }
        .flowOn(testDispatcher)
        .stateIn(GlobalScope, SharingStarted.Lazily, "init state")

    val flow2 = flowOf(Unit)
        .map { makeHeavyRequest() }
        .onEach { logThread("2: before flowOn") }
        .flowOn(testDispatcher)
        .stateIn(GlobalScope, SharingStarted.Lazily, "init state")
        .onEach { logThread("2: after stateIn") }

    val flow3 = flowOf(Unit)
        .map { makeHeavyRequest() }
        .onEach { logThread("3: before flowOn") }
        .flowOn(testDispatcher)
        .onEach { logThread("3: after flowOn") }
        .stateIn(GlobalScope, SharingStarted.Lazily, "init state")

    flow1.test {
        assertEquals("heavy result", expectItem())
        cancelAndIgnoreRemainingEvents()
    }

    flow2.test {
        assertEquals("heavy result", expectItem())
        cancelAndIgnoreRemainingEvents()
    }

    flow3.test {
        assertEquals("heavy result", expectItem())
        cancelAndIgnoreRemainingEvents()
    }

}

L'effet de son exécution sera :

Thread (1: before flowOn): Thread[main @coroutine#2,5,main]
Thread (2: before flowOn): Thread[main @coroutine#3,5,main]
Thread (2: after stateIn): Thread[main @coroutine#6,5,main]
Thread (3: before flowOn): Thread[DefaultDispatcher-worker-1 @coroutine#8,5,main]
Thread (3: after flowOn): Thread[DefaultDispatcher-worker-1 @coroutine#4,5,main]

org.opentest4j.AssertionFailedError: 
Expected :heavy result
Actual   :init state

Sur flow3 placer le onEach entre flowOn y stateIn change complètement le répartiteur et gâche le résultat. Pourquoi cela ?

Answer 1

Vous devez utiliser MainScope au lieu de GlobalScope . Global Scope est une portée non contrôlée.

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/-main-scope.html

Vous trouverez d'autres bons modèles dans les coroutines : https://medium.com/androiddevelopers/coroutines-patterns-for-work-that-shouldnt-be-cancelled-e26c40f142ad

Answer 2

La raison pour laquelle cela se produit est que le stateIn L'opérateur a quelques optimisations selon que le flux amont est un ChannelFlow .
.flowOn(...) renvoie un ChannelFlow tandis que .onEach(...) ne le fait pas.

En général, cela n'a pas vraiment d'importance. La raison pour laquelle c'est important dans votre cas est que vous vous attendez à ce que le flux retourné par stateIn pour ne jamais émettre la valeur initiale. Mais il y a une raison pour laquelle ce paramètre est obligatoire et vous devez vous attendre à recevoir la valeur initiale. Le fait que vous la receviez ou non dépend principalement de la capacité du flux amont à émettre une valeur sans se suspendre.

Maintenant, il semble que l'une des optimisations dans la stateIn c'est-à-dire qu'il pourrait consommer un ChannelFlow sans se suspendre. C'est la raison pour laquelle vous obtenez le comportement attendu en utilisant l'opérateur

.flowOn(testDispatcher) /*returns ChannelFlow*/
.stateIn(GlobalScope, SharingStarted.Lazily, "init state")