Kotlin Flow Interview Questions 2025

// Cold flow — producer block runs for each collector independently
val coldFlow: Flow<Int> = flow {
    println("Producer starting")   // Printed once per collector
    emit(1)
    emit(2)
    emit(3)
}

// Two collectors → producer block runs twice
coldFlow.collect { println("Collector A: $it") }
coldFlow.collect { println("Collector B: $it") }

// Hot flow — always active, values shared across collectors
val _state = MutableStateFlow(0)
val hotFlow: StateFlow<Int> = _state.asStateFlow()

// Both collectors share the same value — no duplication of logic
hotFlow.collect { println("Collector A: $it") }
hotFlow.collect { println("Collector B: $it") }

// StateFlow — for UI state
class SearchViewModel : ViewModel() {
    private val _uiState = MutableStateFlow(SearchUiState())
    val uiState: StateFlow<SearchUiState> = _uiState.asStateFlow()
}

// SharedFlow — for one-shot navigation events
class LoginViewModel : ViewModel() {
    // replay=0: new subscribers won't get past events (correct for navigation)
    private val _events = MutableSharedFlow<LoginEvent>(replay = 0)
    val events: SharedFlow<LoginEvent> = _events.asSharedFlow()

    fun onLoginSuccess() {
        viewModelScope.launch {
            _events.emit(LoginEvent.NavigateToDashboard)
        }
    }
}

// Search with flatMapLatest — cancels previous search on new query
val searchQuery: StateFlow<String> = // comes from UI text field

val searchResults: StateFlow<List<Result>> = searchQuery
    .debounce(300)           // Wait 300ms after user stops typing
    .distinctUntilChanged()  // Ignore if same query typed again
    .flatMapLatest { query ->
        if (query.isBlank()) {
            flowOf(emptyList())      // Empty query → clear results immediately
        } else {
            repository.search(query) // Cancels previous search when new query arrives
        }
    }
    .stateIn(
        scope = viewModelScope,
        started = SharingStarted.WhileSubscribed(5_000),
        initialValue = emptyList()
    )

// build.gradle.kts
// testImplementation("app.cash.turbine:turbine:1.1.0")

@Test
fun `search results update on new query`() = runTest {
    val viewModel = SearchViewModel(
        repository = FakeSearchRepository(),
        dispatcher = UnconfinedTestDispatcher(testScheduler)
    )

    viewModel.searchQuery.test {
        // Initial empty state
        assertEquals(emptyList<Result>(), awaitItem())

        // Type a query
        viewModel.onQueryChanged("kotlin")
        val results = awaitItem()
        assertThat(results).isNotEmpty()
        assertThat(results.first().title).contains("Kotlin")

        cancelAndIgnoreRemainingEvents()
    }
}

// Testing error states
@Test
fun `network error shows error state`() = runTest {
    val repo = FakeRepository(shouldFail = true)
    val viewModel = MyViewModel(repo)

    viewModel.uiState.test {
        assertEquals(UiState.Loading, awaitItem())
        val errorState = awaitItem()
        assertThat(errorState).isInstanceOf(UiState.Error::class.java)
        cancelAndIgnoreRemainingEvents()
    }
}

// Without stateIn: every collector starts its own database query
// val messages: Flow<List<Message>> = repository.getMessages()  // cold

// With stateIn: all collectors share one upstream query
val messages: StateFlow<List<Message>> = repository
    .getMessages()           // cold Room Flow
    .stateIn(
        scope = viewModelScope,
        started = SharingStarted.WhileSubscribed(stopTimeoutMillis = 5_000),
        initialValue = emptyList()
    )

// SharingStarted options:
// - Eagerly: starts immediately, never stops (use in tests or singletons)
// - Lazily: starts on first subscriber, never stops after that
// - WhileSubscribed(N): starts on first subscriber, stops N ms after last unsubscribes
//   → Use WhileSubscribed(5000) in ViewModels for production code

// Wrapping OkHttp WebSocket in callbackFlow
fun observeWebSocket(url: String): Flow<WebSocketMessage> = callbackFlow {
    val client = OkHttpClient()
    val request = Request.Builder().url(url).build()

    val listener = object : WebSocketListener() {
        override fun onMessage(webSocket: WebSocket, text: String) {
            // trySend is non-suspending — safe to call from OkHttp's thread
            trySend(WebSocketMessage.Text(text))
        }

        override fun onMessage(webSocket: WebSocket, bytes: ByteString) {
            trySend(WebSocketMessage.Binary(bytes))
        }

        override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
            close(t)   // Terminates the Flow with an error
        }

        override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
            close()    // Terminates the Flow normally
        }
    }

    val webSocket = client.newWebSocket(request, listener)

    // awaitClose is called when the collector cancels
    // Use it to clean up — cancel the WebSocket connection
    awaitClose {
        webSocket.cancel()
    }
}

// Shared database query — one Room query, many UI subscribers
val sharedMessages: SharedFlow<List<Message>> = repository
    .getMessages()
    .shareIn(
        scope = viewModelScope,
        started = SharingStarted.WhileSubscribed(5_000),
        replay = 1   // New subscribers immediately get the latest list
    )

// Event bus pattern with replay=0
val networkEvents: SharedFlow<NetworkEvent> = networkMonitor
    .observe()
    .shareIn(
        scope = applicationScope,
        started = SharingStarted.Eagerly,
        replay = 0  // Events are fire-and-forget, no caching
    )

// replay=1 vs StateFlow:
// StateFlow requires initial value + enforces same-value de-duplication
// shareIn(replay=1) is more flexible but lacks the current value guarantee
// Use StateFlow for UI state, shareIn for general event sharing

// debounce — wait 300ms after last keystroke
searchQuery
    .debounce(300)
    .collect { query -> search(query) }

// conflate — collector only processes latest, drops intermediates
progressFlow
    .conflate()
    .collect { progress -> updateProgressBar(progress) }

// buffer — producer and collector run concurrently
dataSource
    .buffer(capacity = 64)
    .collect { item -> processItem(item) }   // processing can lag without blocking producer

// combine — latest from each flow, emits on any change
val uiState = combine(
    userFlow,
    settingsFlow,
    networkStatusFlow
) { user, settings, networkStatus ->
    UiState(user = user, settings = settings, isOnline = networkStatus.isConnected)
}

// zip — strict one-to-one pairing
val zipped = flowA.zip(flowB) { a, b -> Pair(a, b) }
// If flowA emits 1, 2, 3 and flowB emits "a", "b":
// zipped emits (1,"a"), (2,"b") — stops because flowB completes

// Pattern 1: catch operator — emit fallback on error
repository.getData()
    .catch { e ->
        Timber.e(e, "Failed to load data")
        emit(emptyList())   // Fallback value
    }
    .collect { data -> updateUi(data) }

// Pattern 2: Result wrapping — cleanest for UI state
sealed class Result<out T> {
    data class Success<T>(val data: T) : Result<T>()
    data class Error(val exception: Throwable) : Result<Nothing>()
    object Loading : Result<Nothing>()
}

fun getData(): Flow<Result<List<Item>>> = flow {
    emit(Result.Loading)
    try {
        val data = api.getItems()
        emit(Result.Success(data))
    } catch (e: IOException) {
        emit(Result.Error(e))
    }
}

// Pattern 3: retry with exponential backoff
repository.getStream()
    .retryWhen { cause, attempt ->
        if (cause is IOException && attempt < 3) {
            delay(2.0.pow(attempt.toInt()).toLong() * 1000)   // 1s, 2s, 4s
            true   // retry
        } else {
            false  // do not retry, propagate the error
        }
    }
    .collect { /* ... */ }

// collect — processes every item, even if new ones arrive
flow.collect { value ->
    delay(1000)  // Slow processing
    updateUi(value)  // Will process all values, even stale ones
}

// collectLatest — cancels current block when new value arrives
flow.collectLatest { value ->
    delay(1000)  // If new value arrives during this delay...
    updateUi(value)  // ...this line may never execute for old values
}

// Practical example: typing in a search box
searchQuery.collectLatest { query ->
    // If user types "k", then "ko", then "kot" within 1 second:
    // - "k" processing is cancelled when "ko" arrives
    // - "ko" processing is cancelled when "kot" arrives
    // - Only "kot" processing completes
    val results = repository.search(query)  // network call
    _results.value = results
}

// Automatic cancellation via viewModelScope
class MyViewModel : ViewModel() {
    init {
        viewModelScope.launch {
            // This collection is automatically cancelled when ViewModel is cleared
            repository.getUpdates().collect { update ->
                _state.value = update
            }
        }
    }
}

// Manual cancellation via Job
val job = viewModelScope.launch {
    someFlow.collect { /* ... */ }
}

// Cancel when needed (e.g., user navigates away before ViewModel clears)
job.cancel()

// Compose — lifecycle-aware collection (preferred over LaunchedEffect + collect)
@Composable
fun MyScreen(viewModel: MyViewModel) {
    // Automatically starts/stops collection based on Lifecycle.State.STARTED
    val state by viewModel.uiState.collectAsStateWithLifecycle()
}

// Without distinctUntilChanged — UI redraws even if value is the same
userPreferences
    .collect { prefs ->
        updateTheme(prefs.theme)   // Called every time prefs emits, even if theme unchanged
    }

// With distinctUntilChanged — only triggers when theme actually changes
userPreferences
    .map { it.theme }
    .distinctUntilChanged()   // Drop if same theme emitted twice in a row
    .collect { theme ->
        updateTheme(theme)   // Only called when theme actually changes
    }

// Custom equality — only distinct if both fields change
userPreferences
    .distinctUntilChangedBy { it.theme }   // Only compare theme field
    .collect { /* ... */ }

// With a custom comparator
userPreferences
    .distinctUntilChanged { old, new ->
        old.theme == new.theme && old.language == new.language
    }
    .collect { /* ... */ }

// flow { } — sequential, single-coroutine producer
val sequentialFlow = flow {
    emit(fetchFromDatabase())   // Sequential
    emit(fetchFromNetwork())    // Runs after database fetch completes
}

// channelFlow — concurrent producers
val concurrentFlow = channelFlow {
    // Launch two concurrent fetches
    launch {
        val dbResult = fetchFromDatabase()
        send(dbResult)   // safe to call from launched coroutine
    }
    launch {
        val networkResult = fetchFromNetwork()
        send(networkResult)   // concurrent with the database fetch
    }
    // Both results emitted as they complete, in whichever order finishes first
}

// Real-world use: merge local cache and network in parallel
fun getProducts(): Flow<List<Product>> = channelFlow {
    // Immediately send cached data
    launch { send(cache.getProducts()) }
    // Concurrently fetch fresh data and send when ready
    launch {
        val fresh = api.getProducts()
        cache.save(fresh)
        send(fresh)
    }
}

import kotlinx.coroutines.flow.merge

// merge — interleave all emissions from multiple flows as they arrive
val allEvents: Flow<Event> = merge(
    locationEvents,
    networkEvents,
    authEvents
)
// Emits whichever event arrives first, from any source

// combine — emit derived state when any source changes
val dashboardState: Flow<DashboardState> = combine(
    userFlow,        // latest user
    metricsFlow,     // latest metrics
    alertsFlow       // latest alerts
) { user, metrics, alerts ->
    DashboardState(user = user, metrics = metrics, alerts = alerts)
}
// Re-emits whenever user, metrics, OR alerts changes

// flattenMerge — flatten a Flow of Flows with concurrency limit
val results: Flow<SearchResult> = queries
    .map { query -> repository.search(query) }  // Flow<Flow<SearchResult>>
    .flattenMerge(concurrency = 4)              // Run up to 4 searches concurrently

// flowOn — run the upstream producer on a different dispatcher
val processedData: Flow<Data> = repository
    .getRawData()                         // IO operations inside (Room/network)
    .map { raw -> cpuIntensiveTransform(raw) }  // CPU-intensive
    .flowOn(Dispatchers.Default)          // Run everything above this on Default
    // Everything below flowOn() still runs on the collector's dispatcher

// Collect on Main (correct for ViewModels)
viewModelScope.launch {    // viewModelScope uses Main dispatcher
    processedData.collect { data ->
        _uiState.value = data  // Main-thread safe
    }
}

// Room Flow — already dispatched correctly, no flowOn needed
val messages: Flow<List<Message>> = messageDao.getAll()
// Room uses Dispatchers.IO internally

// What NOT to do:
viewModelScope.launch(Dispatchers.IO) {   // Wrong!
    uiStateFlow.collect { state ->
        _uiState.value = state  // Crashes — cannot update StateFlow from IO thread in UI context
    }
}