Flow & SharedFlow

What is Flow?

Flow is Kotlin's answer to reactive streams. It produces a sequence of values asynchronously using suspend functions. Unlike RxJava's Observable, Flow is cold — the producer code does not run until someone collects it.

fun fetchItems(): Flow<Item> = flow {
    val items = api.getItems()          // suspends until data arrives
    items.forEach { item ->
        emit(item)                      // sends one item to the collector
        delay(100)                      // simulate processing
    }
}

// Nothing happens here — flow is cold
val itemsFlow = fetchItems()

// Now the flow runs — collector triggers production
lifecycleScope.launch {
    itemsFlow.collect { item ->
        println(item.name)
    }
}

Flow Operators

Operators transform flows before or after collection. They fall into two categories:

Intermediate operators (transform the stream):

api.searchFlow(query)
    .debounce(300L)                     // wait for typing to stop
    .distinctUntilChanged()              // skip duplicate queries
    .map { it.results }                 // extract only results
    .filter { it.isNotEmpty() }         // skip empty results
    .catch { emit(emptyList()) }        // handle upstream errors
    .collect { results ->
        _uiState.value = results
    }

Terminal operators (trigger collection):

• collect — consume each emission
• toList — collect all values into a list
• first — get the first emission and cancel
• reduce — accumulate values into one result

Building Flows

// From values
val numbers = flowOf(1, 2, 3, 4, 5)

// From a sequence
val seq = sequenceOf("a", "b", "c").asFlow()

// From a suspend function
fun temperatureStream(): Flow<Float> = flow {
    while (true) {
        emit(sensor.readTemperature())
        delay(1000)
    }
}

// From callback
fun callbackFlow(): Flow<Event> = callbackFlow {
    val listener = object : EventListener {
        override fun onEvent(event: Event) {
            trySend(event)               // non-blocking send
        }
    }
    eventSource.register(listener)
    awaitClose { eventSource.unregister(listener) }  // cleanup
}

Flow vs suspend Functions

A suspend function returns one value. A Flow emits multiple values over time. Use Flow when you need a stream; use suspend when you need a single result.

StateFlow

StateFlow is a hot flow that always holds a current value. It only emits when the value changes, making it ideal for representing UI state in a ViewModel.

class SearchViewModel : ViewModel() {
    private val _uiState = MutableStateFlow(SearchUiState())
    val uiState: StateFlow<SearchUiState> = _uiState.asStateFlow()

    fun onQueryChanged(query: String) {
        _uiState.update { it.copy(query = query, isLoading = true) }
        viewModelScope.launch {
            val results = repository.search(query)
            _uiState.update {
                it.copy(results = results, isLoading = false)
            }
        }
    }
}

Key rules for StateFlow:

• Always has a value (initialize with a default)
• Uses equals to decide whether to emit — use data classes
• replay = 1 always — new collectors get the current value immediately

SharedFlow

SharedFlow is a hot flow with configurable replay and no required initial value. It broadcasts emissions to all collectors and is useful for events that should not be replayed.

// For one-time events like navigation or showing a snackbar
class EventHolder {
    private val _events = MutableSharedFlow<UiEvent>()
    val events: SharedFlow<UiEvent> = _events.asSharedFlow()

    fun showToast(message: String) {
        viewModelScope.launch {
            _events.emit(UiEvent.ShowToast(message))
        }
    }
}

// Collect in the Activity/Fragment
lifecycleScope.launch {
    eventHolder.events.collect { event ->
        when (event) {
            is UiEvent.ShowToast -> toast(event.message)
        }
    }
}

StateFlow vs SharedFlow Decision

Collecting Flows in Android

Always collect in a lifecycle-aware scope:

// In a Fragment
lifecycleScope.launch {
    viewLifecycleOwner.repeatOnLifecycle(Lifecycle.State.STARTED) {
        viewModel.uiState.collect { state ->
            render(state)
        }
    }
}

// In an Activity
lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        viewModel.uiState.collect { state ->
            render(state)
        }
    }
}

repeatOnLifecycle suspends collection when the lifecycle is below STARTED, preventing wasted work when the UI is not visible.