Exploring the Power of Kotlin Flow: Combining, Merging, and Zipping Streams

This is part of the asynchronous flow series:

• Part 1 - Exploring Android LiveData Usages and Behaviors

• Part 2 - Introduction to Kotlin Flows and Channels

• Part 3 - Exploring Different Ways to Collect Kotlin Flow

• Part 4 - Convert Flow to SharedFlow and StateFlow

• Part 5 - Flow, SharedFlow, StateFlow Class Diagram

• Part 6 - Kotlin Flow - Combine, Merge and Zip

I have 2 flows here.

Flow1 emits int starting from 1 -> 1000 every 1 second.

private val flow1: Flow<Int> = flow {
    repeat(10000) { value ->
        delay(1000)
        Log.d(tag, "[Flow1]: emitting $value")
        emit(value)
    }
}

Flow 2 emits char from A to Z every 2 seconds.

private val flow2: Flow<Char> = flow {
    var value = 'A'
    while(true) {
        delay(2000)
        Log.d(tag, "[Flow2]: emitting $value")
        emit(value)
        if (value == 'Z') {
            value = 'A'
        } else {
            value += 1
        }
    }
}

Combine Flow

This combines flow1 and flow2 into combineFlow using the Flow.combine() extension function flow operator.

val combineFlow = flow1.combine(flow2) { flow1Value,flow2Value  ->
    "${flow1Value}_${flow2Value}"
}

The limitation of this Flow.combine() extension function is it is limited to combining 2 flows. To combine more than 2 flows, you can use combine() function directly which supports up to 5 flows.

val combineFlow = combine(flow1, flow2) { flow1Value, flow2Value ->
    "${flow1Value}_${flow2Value}"
}

To collect the flow, I use the LaunchedEffect() side effect for this demonstration purpose. The recommended way is either using collectAsStateWithLifeCylce()or repeatOnLifecycle(Lifecycle.State.STARTED), see here.

LaunchedEffect(true) {
    viewModel.combineFlow.collect { value ->
        Log.d(tag, "[Combine Flow]: $value")
    }
}

So this is the output. It combines the latest value from flow1 and flow2.

Merge Flow

This merges flow1 and flow2 into mergeFlow,

val mergeFlow = merge(flow1, flow2)

and here is the output:

1 and A are emitted at the same time, followed by 2, then 3 and B and so on.

Zip Flow

This zips flow1 and flow2 into zipFlow,

val zipFlow = flow1.zip(flow2) { flow1value,flow2value  ->
    "${flow1value}_${flow2value}"
}

and here is the output:

As you can see from this output, Zip pairs up the data from flow1 and flow2.

Conclusion

The above diagrams help me to understand the difference between combine, merge and zip flow operators.

I have been using combine() in this project here which combines the flow from a ROOM database(articles) with another flow from a Proto DataStore(user settings). I do not have any chance to use Merge() and Zip(). :)

Source Code

GitHub Repository: Demo_AsyncFlow (see the CombineMergeZipFlowActivity)

Originally published at https://vtsen.hashnode.dev.