Understanding flatMap Variants in Kotlin Flow

flatMapConcat, flatMapMerge, and flatMapLatest

Introduction

Reactive programming in Kotlin often involves transforming streams of data. When dealing with nested flows (a flow that emits other flows), we need operators to flatten them into a single stream. Kotlin Flow provides three powerful operators — flatMapConcat, flatMapMerge, and flatMapLatest — each with unique behavior. Understanding these differences is crucial for writing efficient, predictable asynchronous code.

1. flatMap (Collections)

Before diving into flows, let’s recall flatMap in collections:

• Behavior: Transforms each element into a collection and flattens the result.
• Use Case: Aggregating nested lists into a single list.

Example: Fetching user details and then their posts with RxJava

val apiService: ApiService = ...

apiService.getUsers() // returns Observable<List<User>>
    .flatMap { users ->
        // For each user, fetch their posts
        Observable.fromIterable(users)
            .flatMap { user ->
                apiService.getPosts(user.id) // returns Observable<List<Post>>
            }
    }
    .subscribe({ posts ->
        println("Received posts: $posts")
    }, { error ->
        println("Error: $error")
    })

• getUsers() emits a list of users.
• flatMap converts each user into another observable (getPosts(user.id)).
• The results are merged into one observable stream.

2. flatMapConcat

flatMapConcat is a Kotlin Flow operator that takes each emitted value, transforms it into a new Flow, and then concatenates those flows sequentially—meaning it waits for one flow to finish before starting the next.

• Definition: It’s an intermediate operator in Kotlin’s Flow API used to flatten flows.
• Behavior: For every element emitted by the upstream flow, it creates a new flow and collects them one after another in order.
• Key Point: flatMapConcat ensures strict sequential execution

Example: Fetching user and their posts using flatMapConcat

val usersFlow: Flow<User> = repository.getUsers()

val postsFlow: Flow<Post> = usersFlow
    .flatMapConcat { user ->
        repository.getPosts(user.id) // returns Flow<Post>
    }

lifecycleScope.launch {
    postsFlow.collect { post ->
        println("Received post: $post")
    }
}

• repository.getUsers() returns a Flow<User>.
• This means it will emit a sequence of User objects asynchronously (like a stream of users coming in one by one).
• For each user emitted by usersFlow, you call repository.getPosts(user.id).
• That function returns a Flow<Post> (all posts belonging to that user).
• flatMapConcat then concatenates these flows sequentially
• First, it collects all posts for the first user.
• Only after that flow completes, it moves to the next user’s posts.
• This ensures order is preserved and no overlap happens.
• Inside a coroutine (lifecycleScope.launch), you start collecting from postsFlow.
• As posts are emitted (one by one), you print them.

Characteristics

• Sequential execution: Each inner flow must complete before the next one starts.
• Order preservation: The results are emitted in the same order as the upstream flow.
• No concurrency: It doesn’t overlap flows, so it’s simpler but potentially slower if inner flows are long-running.

3. flatMapMerge

flatMapMerge in Kotlin Flow transforms each emitted value into a new Flow and then merges all those flows concurrently. Unlike flatMapConcat, which processes flows sequentially, flatMapMerge allows multiple inner flows to run in parallel, interleaving their emissions into the final stream

• Definition: flatMapMerge is an intermediate operator in Kotlin Flow.
• Behavior: For every element emitted by the upstream flow, it applies a transformation that returns another Flow. All these inner flows are then collected concurrently (up to a configurable concurrency limit).
• Result: The emissions from different flows are merged together into a single Flow, potentially interleaving values.

Example: Let’s consider a scenario where we want to search users and posts in parallel using flatMapMerge.

Repository

Suppose you’re working with two separate APIs: one that provides user data and another that provides post data. To organize these calls, you can create a SearchRepository class that exposes two methods—searchUsers() and searchPosts()—to fetch data from the respective APIs.

class SearchRepository(private val apiService: ApiService) {

    fun searchUsers(query: String): Flow<List<User>> = flow {
        delay(1000) // simulate network delay
        emit(apiService.searchUsers(query))
    }

    fun searchPosts(query: String): Flow<List<Post>> = flow {
        delay(1500) // simulate network delay
        emit(apiService.searchPosts(query))
    }
}

ViewModel

Create a SearchViewModel that depends on SearchRepository to perform searches for both users and posts. By using flatMapMerge, we can execute these two searches in parallel, allowing results from users and posts to be fetched concurrently and merged into a single flow.

class SearchViewModel(private val repository: SearchRepository) : ViewModel() {

    private val _searchQuery = MutableStateFlow("")
    val searchQuery: StateFlow<String> = _searchQuery

    val searchResults: Flow<SearchResult> = searchQuery
        .debounce(300)
        .filter { it.isNotBlank() }
        .flatMapMerge { query ->
            merge(
                repository.searchUsers(query).map { SearchResult.Users(it) },
                repository.searchPosts(query).map { SearchResult.Posts(it) }
            )
        }
        .flowOn(Dispatchers.IO)

    fun updateQuery(query: String) {
        _searchQuery.value = query
    }
}

sealed class SearchResult {
    data class Users(val users: List<User>) : SearchResult()
    data class Posts(val posts: List<Post>) : SearchResult()
}

• _searchQuery is a MutableStateFlow holding the current search text (initially empty).
• searchQuery is the read-only StateFlow exposed to observers (UI can collect it but not modify it).
• This is a reactive way to handle search input.
• debounce(300): Waits 300ms after the last input before emitting and prevents firing searches on every keystroke (useful for search bars).
• filter { it.isNotBlank() }: Ignores empty queries and ensures only meaningful search strings trigger API calls.
• flatMapMerge { query -> ... }: For each query, it starts multiple flows concurrently. It will merges results from: repository.searchUsers(query) → returns Flow<List<User>> repository.searchPosts(query) → returns Flow<List<Post>>
• Each result is wrapped into a SearchResult sealed class variant: SearchResult.Users(it) and SearchResult.Posts(it)
• merge(...): Combines both flows into a single flow. Results from users and posts can arrive interleaved (parallel execution).
• flowOn(Dispatchers.IO): Ensures the flow runs on the IO dispatcher (good for network calls).

Get Vaghela Mithun’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

Activity/Fragment

You can implement either an Activity or a Fragment to handle the search feature. Use an EditText to capture the user’s input, and display the search results in a RecyclerView for a clean, scrollable list presentation.

class SearchFragment : Fragment() {

    private val viewModel: SearchViewModel by viewModels {
        SearchViewModelFactory(SearchRepository(UserApi(), PostApi()))
    }

    override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
        val searchInput = view.findViewById<EditText>(R.id.searchInput)
        val resultsList = view.findViewById<RecyclerView>(R.id.resultsList)

        val adapter = ResultsAdapter()
        resultsList.adapter = adapter

        lifecycleScope.launch {
            viewModel.searchResults.collect { result ->
                when (result) {
                    is SearchResult.Users -> adapter.submitUsers(result.users)
                    is SearchResult.Posts -> adapter.submitPosts(result.posts)
                }
            }
        }

        searchInput.addTextChangedListener { text ->
            viewModel.updateQuery(text.toString())
        }
    }
}

Execution Flow Example

1. User types "android".
2. After 300ms pause, searchQuery emits "android".
3. flatMapMerge triggers both API calls in parallel: searchUsers("android") and searchPosts("android")
4. As soon as either API responds, results are emitted as SearchResult.Users or SearchResult.Posts.
5. The UI collects searchResults and renders both lists.

Benefits

• Parallel execution: Both APIs are queried at the same time.
• Faster results: UI can show partial results immediately (e.g., users first, then posts).
• Scalable: You can merge multiple sources (news, music, shopping, etc.).

👉 This pattern is great when you want multi-source search or parallel data fetching.

4. flatMapLatest

flatMapLatest in Kotlin Flow transforms each emitted value into a new Flow, but whenever a new value arrives, it cancels the previous inner flow and switches to the latest one. This makes it ideal for scenarios like search bars, where only the most recent query matters.

• Definition: Returns a flow that switches to a new inner flow every time the upstream emits.
• Behavior: If a new value arrives before the previous inner flow finishes, the old flow is canceled immediately.
• Result: Only emissions from the latest flow are collected.

A common real-world scenario for flatMapLatest is implementing a search bar that cancels outdated searches whenever the user types a new query. In this setup, we use a SearchRepository class that provides two methods—searchUsers() and searchPosts()—to fetch data from their respective APIs.

SearchViewModel

class SearchViewModel(private val repository: SearchRepository) : ViewModel() {

    // Flow that collects search queries from UI
    private val _searchQuery = MutableStateFlow("")
    val searchQuery: StateFlow<String> = _searchQuery

    // Expose search results as Flow
    val searchResults: Flow<List<Result>> = searchQuery
        .debounce(300) // wait for typing to pause
        .filter { query -> query.isNotBlank() } // ignore empty queries
        .flatMapLatest { query ->
            repository.searchUsers(query) // cancels old search if new query arrives
        }
        .flowOn(Dispatchers.IO) // run on background thread

    fun updateQuery(query: String) {
        _searchQuery.value = query
    }
}

flatMapLatest { query -> repository.searchUsers(query) }

• For each query, it calls repository.searchUsers(query) which returns a Flow<List<Result>>.
• flatMapLatestbehavior: If a new query arrives before the previous search finishes, the old search flow is canceled immediately.
• Only the latest query’s results are collected.
• This is perfect for search bars: you don’t want outdated results if the user keeps typing.

Execution Flow Example

Imagine the user types quickly: "a" → "an" → "and" → "android"

• "a" triggers a search, but before it completes, "an" arrives → cancel "a".
• "an" triggers a search, but "and" arrives → cancel "an".
• "and" triggers a search, but "android" arrives → cancel "and".
• Only the latest query "android" completes and emits results.

Benefits

• Efficient: Cancels outdated searches automatically.
• Responsive: UI always shows the latest results.
• Clean: Structured concurrency ensures no leaks or dangling jobs.

Comparison Table

Real-World Android Examples

• Search Bar: Use flatMapLatest to cancel old queries when the user types a new one.
• Multi-source Fetch: Use flatMapMerge to query multiple APIs in parallel (e.g., books + movies).
• Chained Operations: Use flatMapConcat when tasks must run sequentially (e.g., login → fetch profile → fetch settings).

📝 Conclusion

Choosing the right flattening operator is about intent:

• Use flatMapConcat for ordered, sequential tasks.
• Use flatMapMerge for parallel, independent tasks.
• Use flatMapLatest for real-time scenarios where only the latest result matters.

Mastering these operators makes your Kotlin Flow code cleaner, faster, and more predictable.