Differences in methods of collecting Kotlin Flows

Some of you may have recently started using Kotlin Flow, the new framework by JetBrains to handle observable streams.

As you probably already know from many articles on the internet, in order to collect a flow, you have two main options. You can either use .collect() , or .launchIn() . (For the purposes of this article we avoid other terminal operators such as .toList(), etc)

What are the differences between these two methods?

There are some obvious ones and one is very subtle, not immediately obvious to everyone and you should pay attention to it, if you want your code to behave as expected.

Obvious differences

For one, if we take a look at method signatures of these two methods, we’ll learn, that .collect() is suspending method, while .launchIn() is not:

public suspend fun collect(collector: FlowCollector<T>)
public fun <T> Flow<T>.launchIn(scope: CoroutineScope): Job

The practical difference then is, that you can call collect() method only from another suspending function or from a coroutine. For example like this:

coroutineScope.launch {
    flowOf(1, 2, 3)
        .collect { println(it) }
}

whereas .launchIn() can be called like this in any regular function:

flowOf(1, 2, 3)
    .onEach { println(it) }
    .launchIn(coroutineScope)

With the .launchIn() method, you also get a Job as return value, so you could cancel the flow by cancelling the job:

val job = flowOf(1, 2, 3)
    .onEach { println(it) }
    .launchIn(coroutineScope)

job.cancel()

There’s one more difference between these two methods, that may not be so obvious however and it returns to the fact, that .collect() is a suspending function.

A more subtle difference

Let’s take a short quiz here. What do you think the output of this little piece of code will be?

runBlocking {
    flowOf(1, 2, 3)
        .onEach { delay(100) }
        .collect { println(it) }
    flowOf("a", "b", "c")
        .collect { println(it) }
}

Since there’s a delay on each of the emissions in the first Flow and the second Flow emits and prints immediately one might expect this output:

a
b
c
1
2
3

However, the resulting output is actually this:

1
2
3
a
b
c

Why?

Collect is a suspending function

The reason for this behaviour is, that .collect() is a suspending function. It suspends coroutine, until it is finished doing its own thing. So in case of our code snipped, the coroutine is suspended on every delay that occurs and the coroutine will not continue executing any other code, until the first flow has been collected. That means, that the second flow won’t be collected at all, until first flow is completed.

Ok, so in practical scenarios, this is probably not what you want. You probably do expect to see first the letters a, b, c in the output and only then 1, 2, 3.

So how can we achieve this? We can use .launchIn()

But why does it work?

Let’s look into the implementation of .launchIn() method:

public fun <T> Flow<T>.launchIn(scope: CoroutineScope): Job =     scope.launch {
    collect() // tail-call
}

As you can see from the code above, .launchIn() does call .collect() method internally. However it collects the flow in a scope.launch {} block. This code means, that a new child coroutine is launched within the coroutine scope specified in the parameter.

So in this code:

runBlocking {
    flowOf(1, 2, 3)
        .onEach { delay(100) }
        .onEach { println(it) }
        .launchIn(this)
    flowOf("a", "b", "c")
        .onEach { println(it) }
        .launchIn(this)
}

when the first flow is collected, the coroutine launched by runBlocking {} is NOT suspended. Instead, a new child coroutine was launched in the scope of the blocking coroutine (launched via runBlocking {}) and this new child coroutine is the one, that will be suspended.

That means, that the second flow is not blocked from being collected. And that means, the output is what you’d expect:

a
b
c
1
2
3

Conclusion

So in conclusion I’d recommend preferring .launchIn() over .collect() for collecting your flows, to avoid unexpected bugs. Use .collect() only if you are absolutely sure, that you don’t mind and will not mind in the future it suspending your coroutine (for example because the coroutine was launched to collect this one specific flow and nothing else is ever going to be executed in it, not even in the future you’d want to add anything else into that coroutine).