Understanding Flow Operators in Kotlin: A Simple Guide

Transformed: 1
Additional: 1
Transformed: 2
Additional: 2
Transformed: 3
Additional: 3
Transformed: 4
Additional: 4
Transformed: 5
Additional: 5

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
  val sourceFlow: Flow<Int> = flowOf(1, 2, 3)

  val transformedFlow: Flow<String> = sourceFlow.flatMapConcat { number ->
    flow {
      emit("Start $number")
      delay(100)
      emit("End $number")
    }
  }

  transformedFlow.collect { value ->
    println(value)
  }
}

In this example, we start with a source Flow of integers (sourceFlow). We use the flatMapConcat operator to transform each emitted integer into a new Flow.

• Each integer generates a new Flow using the flow {} builder.
• Inside this nested Flow, two strings are emitted with a delay between them.
• Since flatMapConcat ensures sequential execution, the next integer’s Flow only starts after the previous one finishes.

Output:

Start 1
End 1
Start 2
End 2
Start 3
End 3

As you can see, the emitted values keep their original order, and all nested Flows are combined into one continuous stream.

The flatMapConcat operator is helpful when you need to run asynchronous tasks for each item in a Flow while keeping the order of the results.

flatMapMerge – Merging Multiple Flows

The flatMapMerge operator in Kotlin Flow combines multiple Flows into a single Flow, but unlike flatMapConcat, it does not preserve the order of emissions. Each item from the source Flow is transformed into a new Flow, and all the elements from these nested Flows are merged together in parallel. The final output may have interleaved results, depending on the execution timing.

How It Works:

• Source Flow: The original Flow where flatMapMerge is applied.
• Transformation Function: A function that takes each item from the source Flow and returns a new Flow.
• Resulting Flow: A merged Flow that emits all values from the nested Flows. Since the nested Flows run in parallel, their order is not guaranteed.

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
  val sourceFlow: Flow<Int> = flowOf(1, 2, 3)

  val transformedFlow: Flow<String> = sourceFlow.flatMapMerge { number ->
    flow {
      emit("Start $number")
      delay(100)
      emit("End $number")
    }
  }

  transformedFlow.collect { value ->
    println(value)
  }
}

In this example, we start with a Flow of integers (sourceFlow). We use the flatMapMerge operator on sourceFlow to create a new Flow for each emitted number. Each of these nested Flows emits two strings with a short delay.

When we collect and print the values from transformedFlow, the output may look like this:

Start 1
Start 2
Start 3
End 1
End 2
End 3

As you can see, the elements from different nested Flows are mixed together and may not follow the original order. This happens because each Flow runs independently, and their results appear as they finish.

The flatMapMerge operator is helpful when you need to process multiple items at the same time and combine the results into one Flow. It’s useful when order doesn’t matter, and you want to improve performance by running tasks in parallel.

flatMapLatest – Keeping Only the Latest Value

The flatMapLatest operator in Kotlin Flow merges multiple Flows into a single Flow but ensures that only the latest emitted value is processed. When a new item arrives from the source Flow, any previous transformation is canceled, and only the latest item is processed.

How it works:

• Source Flow: The original Flow that emits values.
• Transformation Function: A function that takes an item from the source Flow and returns another Flow.
• Cancels Previous Transformations: If a new item arrives, any ongoing processing for previous items is stopped.
• Resulting Flow: Emits values only from the most recent transformation.

Here’s a simple example in code:

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
  val sourceFlow: Flow<Int> = flowOf(1, 2, 3)

  val transformedFlow: Flow<String> = sourceFlow.flatMapLatest { number ->
    flow {
      emit("Start $number")
      delay(100)
      emit("End $number")
    }
  }

  transformedFlow.collect { value ->
    println(value)
  }
}

Start 1
Start 2
Start 3
End 3

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
  val sourceFlow: Flow<Int> = flowOf(1, 2, 3, 4, 5)

  val takenFlow: Flow<Int> = sourceFlow.take(3)

  takenFlow.collect { value ->
    println(value)
  }
}

1
2
3

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
  val sourceFlow: Flow<Int> = flowOf(1, 2, 3, 4, 5)

  val droppedFlow: Flow<Int> = sourceFlow.drop(2)

  droppedFlow.collect { value ->
    println(value)
  }
}

3
4
5

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
  val sourceFlow: Flow<Int> = flowOf(1, 2, 3, 4, 5)

  val modifiedFlow: Flow<Int> = sourceFlow.onEach { value ->
    // Perform a side effect, such as printing the element
    println("Element: $value")
  }

  modifiedFlow.collect { value ->
    // Process the elements (in this case, just collecting them)
  }
}

Element: 1
Element: 2
Element: 3
Element: 4
Element: 5

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
  val sourceFlow: Flow<Int> = flow {
    emit(1)
    emit(2)
    throw RuntimeException("Error occurred!")
    emit(3)
    emit(4)
 }

  val recoveredFlow: Flow<Int> = sourceFlow.catch { exception ->
    // Handle the exception gracefully
    println("Caught exception: ${exception.message}")
    emit(5) // Emit a recovery value
  }

  recoveredFlow.collect { value ->
    println("Received: $value")
  }
}

Received: 1
Received: 2
Caught exception: Error occurred!
Received: 5

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContext

fun main() = runBlocking {
  val sourceFlow: Flow<Int> = flow {
    for (i in 1..5) {
    println("Emitting $i in thread ${Thread.currentThread().name}")
    emit(i)
    delay(100)
   }
 }

  val transformedFlow: Flow<String> = sourceFlow.map { value ->
    "Transformed $value in thread ${Thread.currentThread().name}"
  }.flowOn(Dispatchers.IO) // Switch execution context to IO dispatcher

  withContext(Dispatchers.Default) {
    // Collect in a different context (Default dispatcher)
    transformedFlow.collect { value ->
      println("Received: $value in thread ${Thread.currentThread().name}")
    }
  }
}

Emitting 1 in thread DefaultDispatcher-worker-2
Received: Transformed 1 in thread DefaultDispatcher-worker-2 in thread DefaultDispatcher-worker-1
Emitting 2 in thread DefaultDispatcher-worker-2
Received: Transformed 2 in thread DefaultDispatcher-worker-2 in thread DefaultDispatcher-worker-2
Emitting 3 in thread DefaultDispatcher-worker-2
Received: Transformed 3 in thread DefaultDispatcher-worker-2 in thread DefaultDispatcher-worker-2
Emitting 4 in thread DefaultDispatcher-worker-2
Received: Transformed 4 in thread DefaultDispatcher-worker-2 in thread DefaultDispatcher-worker-1
Emitting 5 in thread DefaultDispatcher-worker-1
Received: Transformed 5 in thread DefaultDispatcher-worker-1 in thread DefaultDispatcher-worker-1

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
  val numbersFlow: Flow<Int> = flowOf(1, 2, 3, 4, 5)
  val lettersFlow: Flow<String> = flowOf("A", "B", "C", "D", "E")

  val zippedFlow: Flow<Pair<Int, String>> = numbersFlow.zip(lettersFlow) { number, letter ->
    Pair(number, letter)
  }

  zippedFlow.collect { pair ->
    println("Received: ${pair.first} - ${pair.second}")
  }
}

Received: 1 - A
Received: 2 - B
Received: 3 - C
Received: 4 - D
Received: 5 - E

import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
  val sourceFlow: Flow<Int> = flow {
    emit(1)
    emit(2)
    emit(2) // Duplicate
    emit(3)
    emit(3) // Duplicate
    emit(3) // Duplicate
    emit(4)
  }

  val distinctFlow: Flow<Int> = sourceFlow.distinctUntilChanged()

  distinctFlow.collect { value ->
    println("Received: $value")
  }
}

Received: 1
Received: 2
Received: 3
Received: 4

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
  val sourceFlow: Flow<Int> = flow {
  emit(1)
  emit(2)
  throw RuntimeException("Error occurred!")
    emit(3)
  }

  val retriedFlow: Flow<Int> = sourceFlow.retry(2) // Retry up to 2 times

  retriedFlow.collect { value ->
    println("Received: $value")
  }
}

Received: 1
Received: 2
Received: 1
Received: 2
Received: 1
Received: 2

val retriedFlow: Flow<Int> = sourceFlow.retry { cause, _ ->
  // Retry only if the cause of the exception is of a specific type
  cause is MyCustomException
}

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

fun main() = runBlocking {
  val flow = flow {
    emit(1)
    delay(100) // Simulate some processing time
    emit(2)
    delay(100) // Simulate some processing time
    emit(3)
  }

 flow
   .conflate() // Use the conflate operator
   .collect { value ->
     delay(200) // Simulate slow processing
     println(value)
   }
}

1
3