Combining Multiple Kotlin Flows

1. Introduction

In Kotlin, coroutines and Flows provide powerful tools for handling asynchronous and stream-based programming. Combining multiple flows allows us to orchestrate complex asynchronous operations efficiently.

In many real-world scenarios, developers often encounter the need to combine multiple flows, either to synchronize their emissions or to perform operations that involve data from multiple sources. In this tutorial, we’ll explore various techniques for combining multiple flows in Kotlin to address these common use cases.

2. Introduction to Flows

Before we dive into combining flows, let’s have a brief overview of what flows in Kotlin are. Flows are a part of Kotlin Coroutines that emit multiple asynchronously-produced values. They are similar to sequences but designed to handle asynchronous computations.

3. Creating Sample Flows

For demonstration purposes, let’s create two simple Flows that emit some integers asynchronously.

suspend fun sampleFlow1(): Flow<Int> = flow {
    repeat(5) {
        delay(1000)
        emit(it)
    }
}

suspend fun sampleFlow2(): Flow<Int> = flow {
    repeat(5) {
        delay(1500)
        emit(it * it)
    }
}

In this code, we’ve defined two suspending functions sampleFlow1() and sampleFlow2(), each returning a Flow of Integers. These flows emit values asynchronously using the flow builder. sampleFlow1 emits integers from 0 to 4 with a delay of 1000 milliseconds between emissions, while sampleFlow2 emits squares of integers from 0 to 4 with a delay of 1500 milliseconds between emissions.

3.1. The zip() Method

suspend fun main() {
    val combinedFlow = sampleFlow1().zip(sampleFlow2()) { first, second ->
        "($first, $second)"
    }
    combinedFlow.collect { println(it) }
}

In this method, we use the zip() function to combine emissions from both flows. The zip() function waits for both flows to emit a value, then combines corresponding values from each flow using the provided lambda function { first, second -> … }. In this case, it combines integers emitted from sampleFlow1 with squares of integers emitted from sampleFlow2. The combined values are printed as pairs.

Let’s take a look at the instances where we would use this zip() method to combine our flows:

• We use zip() when we want to combine corresponding elements from multiple flows into pairs
• It waits for both flows to emit a value before combining them
• Useful when we need to synchronize the emissions of multiple flows

The output for this code will be:

(0, 0)
(1, 1)
(2, 4)

3.2. The combine() Method

suspend fun main() {
    val combinedFlow = sampleFlow1().combine(sampleFlow2()) { first, second ->
        "($first, $second)"
    }
    combinedFlow.collect { println(it) }
}

In this code, we utilize the combine() function to merge emissions from both flows. Unlike zip(), combine() produces a new value every time either of the flows emits a value. It combines the latest values emitted by each flow. The combined values are printed as pairs.

Let’s take a look at the instances where we’d use this the combine() method to combine our Flows:

• We use combine() when we want to merge two flows and emit a new value every time either of the flows emits a value.
• It combines the latest values emitted by each flow
• Suitable for scenarios where we need to react to changes in any of the flows independently

The output for this code will be:

(0, 0)
(1, 0)
(2, 0)
(2, 1)
(2, 4)

3.3. The flatMapConcat() Method

suspend fun main() {
    val combinedFlow = sampleFlow1().flatMapConcat { value1 ->
        sampleFlow2().map { value2 ->
            "($value1, $value2)"
        }
    }
    combinedFlow.collect { println(it) }
}

In this method, we use flatMapConcat() to sequentially combine emissions from both flows. It waits for each emission from sampleFlow1. For each emission, it collects emissions from sampleFlow2, combining them sequentially. The combined values are printed as pairs.

Let’s take a look at the instances where we would use the flatMapConcat() method to combine our flows:

• We use flatMapConcat() method when one flow depends on the emissions of another flow.
• It concatenates the emissions of multiple flows sequentially.
• Useful when we need to process emissions from one flow before processing emissions from another flow.

The output for this code will be:

(0, 0)
(0, 1)
(0, 4)
(1, 0)
(1, 1)
(1, 4)
(2, 0)
(2, 1)
(2, 4)

3.4. The flatMapMerge() Method

suspend fun main() {
    val combinedFlow = sampleFlow1().flatMapMerge { value1 ->
        sampleFlow2().map { value2 ->
            "($value1, $value2)"
        }
    }
    combinedFlow.collect { println(it) }
}

Similar to flatMapConcat(), flatMapMerge() concatenates the emissions of multiple flows. However, it does so concurrently, potentially interleaving them depending on their respective execution times. The combined values are printed as pairs.

Let’s take a look at the instances where we would use the flatMerge() method to combine our flows:

• We use flatMapMerge() when you want to concatenate the emissions of multiple flows concurrently.
• It interleaves emissions from multiple flows, potentially producing a result faster than flatMapConcat().
• Suitable for scenarios where we want to maximize parallelism while processing emissions from multiple flows.

The output for our code sample will be:

(0, 0)
(1, 0)
(0, 1)
(2, 0)
(1, 1)
(0, 4)
(2, 1)
(1, 4)
(2, 4)

3.5. The flattenConcat() Method

suspend fun main() {
    val combinedFlow = listOf(sampleFlow1(), sampleFlow2()).flattenConcat()
    combinedFlow.collect { println(it) }
}

Here, flattenConcat() is used to concatenate the emissions of both flows sequentially into a single flow. It’s a shorthand for flatMapConcat(), specifically designed for flows. The combined values are printed as pairs.

Let’s take a look at the instances where we would use the flattenConcat() method to combine our flows:

• We use flattenConcat() when we have a list of flows and want to concatenate their emissions sequentially into a single flow.
• It’s a shorthand for flattening a flow of flows into a single flow by concatenating them sequentially.
• Useful when we have a collection of flows and want to treat them as a single flow.

The output for our code will be:

0
1
2
0
1
4

3.6. The merge() Method

suspend fun main() {
    val combinedFlow = merge(sampleFlow1(), sampleFlow2())
    combinedFlow.collect { println(it) }
}

We use merge() to combine emissions from both flows concurrently, potentially interleaving them depending on their respective execution times. It emits values as soon as they are available from any of the flows. The combined values are printed as pairs.

Let’s take a look at the instances where we would use the merge() method to combine our flows:

• We use merge() when we want to combine emissions from multiple flows concurrently.
• It emits values as soon as they are available from any of the flows, potentially interleaving emissions.
• Suitable for scenarios where we want to process emissions from multiple flows concurrently without waiting for any particular flow to complete.

The output for this code will be:

0
0
1
1
2
4

4. Conclusion

Combining multiple flows in Kotlin provides a flexible and powerful way to handle asynchronous data streams. In this tutorial, we explored various methods including zip(), combine(), flatMapConcat(), flatMapMerge(), flattenConcat() and merge(). Understanding these methods and when to use them will empower us to effectively orchestrate complex asynchronous operations in our Kotlin applications.

The full implementation of these examples is available over on GitHub.