Hey all… I keep wrestling with this general proble...

# flow

[Alexander Black]

Hey all… I keep wrestling with this general problem… and I’m hoping someone can point me in the right direction. I think it’s just my lack of understanding.
If I do something like this:

val scope = MainScope() + Job()
val scope2 = MainScope() + Job()


val someEmitter = MutableStateFlow("")

fun test3()= runBlocking {
    scope.launch {
        someEmitter.emit("1")
        someEmitter.emit("2")
        someEmitter.emit("3")
    }
    scope2.launch {
        someEmitter.collect {
            println(it)
        }
    }
}

why does scope2 collect operation get blocked by scope? do they have to be on different threads? Am I missing something? I’ve gotten this type of thing to work before, but only by using another thread as the collector.

[Lefko]

Just curious what if you put the collect before the emit.

val scope = MainScope() + Job()
val scope2 = MainScope() + Job()


val someEmitter = MutableStateFlow("")

fun test3()= runBlocking {
    scope2.launch {
        someEmitter.collect {
            println(it)
        }
    }
    scope.launch {
         someEmitter.emit("1")
         someEmitter.emit("2")
         someEmitter.emit("3")
     }
}

[wasyl]

Your code with some changes (I don’t know MainScope does print 3 for me blob thinking upside down

@Test
fun name(): Unit = runBlocking {
    val scope = CoroutineScope(EmptyCoroutineContext) + Job()
    val scope2 = CoroutineScope(EmptyCoroutineContext) + Job()

    val someEmitter = MutableStateFlow("")

    scope.launch {
        someEmitter.emit("1")
        someEmitter.emit("2")
        someEmitter.emit("3")
    }
    scope2.launch {
        someEmitter.collect {
            println(it)
        }
    }
}

[wasyl]

Ah you expected each item to be emitted. While a `StateFlow`’s emit is suspendable, it doesn’t actually suspend as far as I understand. If there’s no suspension, the body of the second launch isn’t even called until after you emit the last value. This code:

@Test
fun name(): Unit = runBlocking {
    val scope = CoroutineScope(EmptyCoroutineContext) + Job()
    val scope2 = CoroutineScope(EmptyCoroutineContext) + Job()

    val someEmitter = MutableStateFlow("")

    println("Launching first scope")
    scope.launch {
        println("Launching code in first scope")
        println("Emitting 1")
        someEmitter.emit("1")
        println("Emitting 2")
        someEmitter.emit("2")
        println("Emitting 3")
        someEmitter.emit("3")
    }
    println("Launching second scope")
    scope2.launch {
        println("Launching code in second scope")
        someEmitter.collect {
            println(it)
        }
    }
}

will print

Launching first scope
Launching second scope
Launching code in first scope
Emitting 1
Emitting 2
Emitting 3
Launching code in second scope
3

and this is expected. First both launches are called, then the runBlocking suspends so the first launch executes its body. Nothing inside suspends, so the first launch completes before execution moves to the second launch.

[wasyl]

I may have some details wrong, but the bottom line is — you shouldn’t rely on getting individual emissions from MutableStateFlow anyway. It represents a state, and collectors aren’t guaranteed to get the intermediate emissions, but the final emitted state should always be correct

[Alexander Black]

@wasyl This is fantastic info! really appreciate you taking the time to explain that. Is there a flow type that will receive every emission that you are aware of?

[Alexander Black]

or would that maybe be more to do with like setting a buffer/replay value

[Alexander Black]

I think I figured it out… state flow was the wrong flow type for what I was trying to achieve… Looks like ShareFlow is what I was wanting here’s the code that achieved what I was trying to do:

fun test3()= runBlocking {
    val scope = CoroutineScope(EmptyCoroutineContext) + Job()
    val scope2 = CoroutineScope(EmptyCoroutineContext) + Job()

    val someEmitter = MutableSharedFlow<String>(1,0)

    scope2.launch {
        someEmitter.collect {
            println(it)
        }
    }
    scope.launch {
        someEmitter.emit("1")
        someEmitter.emit("2")
        someEmitter.emit("3")
    }

}

[Alexander Black]

thank you guys very much for your help! you got me going in the right direction. 🙂

[Raheel Naz [Amex]]

why use an EmptyCorutineContext?

[wasyl]

You can replace it with a Job() directly instead of plussing, I think