Understanding the Architectural Timing of Flow Conversion in Kotlin and Android.
Kotlin Flow makes it easy to work with asynchronous data streams. We can use it to observe database changes, listen for network updates, combine multiple data sources, and expose reactive data to the UI.
But there is a design question that often gets overlooked:
Where should we actually convert data into Flow?
It’s easy to wrap an API in flow {}or callcackFlow and call it a day. But doing so in the wrong layer can make an architecture unnecessarily complicated. We may end up moving callback management into the Repository, wrapping one-time operations in streams, or creating Flow objects that serve no real purpose.
A better approach is to introduce Flow close to the source of the data, then let higher layers work with that stream.
Let’s look at what this means in practice.
Start with the Data Source
Different data sources expose data in different ways.
Some provide callbacks, others return values through suspend functions, and some already support Flow.
The first step is to understand how the underlying API delivers data.
Database: Use Flow Directly
Room already supports observable queries.
@Query("SELECT * FROM items")
fun observeItems(): Flow<List<ItemEntity>>We don’t need to create another Flow around this query. Room already handles change observation and emits updated results when the underlying data changes.
The same principle applies to DataStore:
val showAds: Flow<Boolean> = dataStore.data.map { preferences ->
preferences[SHOW_ADS] ?: true
}The data source already provides a stream of changes. We should use it directly rather than introducing another abstraction.
Callback APIs: Use callbackFlow
Android APIs often rely on callbacks or listeners to report changes.
For example, consider a location API that delivers updates through a listener.
fun observeLocation(): Flow<Location> = callbackFlow {
val listener = LocationListener { location ->
trySend(location)
}
locationManager.registerUpdates(listener)
awaitClose {
locationManager.removeUpdates(listener)
}
}callbackFlow bridges the callback-based API into Kotlin Flow.
The important part is awaitClose. It keeps the callback registered while the Flow is being collected and unregisters it when collection is cancelled.
This is exactly where the conversion belongs: next to the API that owns the listener lifecycle.
We don’t want the Repository or ViewModel to know how to register and unregister platform callbacks.
The same approach works for connectivity monitoring, sensor updates, Bluetooth events, and other listener-based APIs.
Suspend Functions: Don’t Wrap Everything in Flow
Not every asynchronous operation needs to become a Flow.
Consider a typical API request:
suspend fun getUser(): User
This represents a one-time operation. A suspend function already provides a straightforward way to retrieve the result.
We could wrap it in Flow:
fun observeUser(): Flow<User> = flow {
emit(api.getUser())
}It simply executes the request when the Flow is collected. Every new collection triggers another request.
If we only need to fetch a user once, keeping the suspend function is usually the better choice.
Use flow {}when you actually need to produce a stream, for example when combining multiple suspend operations into a Flow-based pipeline.
fun observeUserData(): Flow<UserData> = flow {
val user = api.getUser()
val settings = api.getSettings()
emit(UserData(user, settings))
}The distinction is simple: use suspend functions for one-time results and Flow for streams of values over time.
Convert at the Source, Not in the Repository
Now let’s consider where these conversions should happen in a layered architecture.
Suppose we have a callback-based network monitor.
One approach is to expose the callback API from the Data Source and convert it into Flow inside the Repository.
class NetworkDataSource {
fun registerListener(
listener: NetworkListener
) {
// Register callback
}
}class NetworkRepository(
private val dataSource: NetworkDataSource
) {
fun observeConnectivity(): Flow<Boolean> = callbackFlow {
val listener = NetworkListener { connected ->
trySend(connected)
}
dataSource.registerListener(listener)
awaitClose {
dataSource.unregisterListener(listener)
}
}
}
This works, but the Repository now needs to understand the callback lifecycle.
It knows how to register and unregister listeners, even though that responsibility belongs to the component interacting with the underlying API.
A cleaner design is to convert the callback into Flow inside the Data Source.
class NetworkDataSource {
fun observeConnectivity(): Flow<Boolean> = callbackFlow {
val listener = NetworkListener { connected ->
trySend(connected)
}
registerListener(listener)
awaitClose {
unregisterListener(listener)
}
}
}The Repository can simply expose or transform the resulting Flow.
class NetworkRepository(
private val dataSource: NetworkDataSource
) {
fun observeConnectivity(): Flow<Boolean> = dataSource.observeConnectivity()
}
Now each layer has a clear responsibility:
- Data Source handles the underlying API and its lifecycle.
- Repository provides data through an application-facing interface.
- ViewModel derives the state required by the UI.
The Repository doesn’t need to know whether the original data came from a callback, listener, or database.
It only needs to work with the resulting Flow.
Let the Repository Transform and Combine Data
Once data sources expose Flow, the Repository can focus on data-related operations.
For example, database entities can be converted into domain models using map.
fun observeItems(): Flow<List<Item>> = dao.observeItems()
.map { entities ->
entities.map { it.toDomain() }
}
We can also combine multiple sources when the resulting data belongs to the same repository responsibility.
fun observeHome(): Flow<HomeData> = combine(
observeItems(),
observeCategories()
) { items, categories ->
HomeData(items, categories)
}
Or switch between data streams based on changing input with flatMapLatest.
fun search(query: Flow<String>): Flow<List<Item>> = query
.flatMapLatest { keyword ->
dao.searchItems(keyword)
}
These operators don’t create the original data source. They transform existing streams into something more useful for the application.
That distinction helps keep the architecture understandable.
Convert Flow into StateFlow in the ViewModel
After the Repository provides Flow, the ViewModel can use it to derive UI state.
This is where stateIn() becomes useful.
val uiState: StateFlow<UiState> = combine(
repository.items,
query,
repository.showAds
) { items, query, showAds ->
UiState(
items = items.filter { it.matches(query) },
showAds = showAds
)
}.stateIn(
viewModelScope,
SharingStarted.WhileSubscribed(5_000),
UiState()
)
Notice the difference between the two conversions:
- Data Source → Flow: Represents changes coming from the underlying source.
- ViewModel → StateFlow: Represents the current UI state derived from application data and user input.
They solve different problems and belong to different architectural responsibilities.
We don’t need to manually update UiState every time an input changes. combine derives a new state whenever one of its upstream sources emits.
This keeps the UI state declarative and makes dependencies explicit.
Common Ways to Create Flow
Here is a quick reference for choosing the right Flow builder based on the original data source.
For example, converting an existing collection:
val itemsFlow: Flow<Item> = items.asFlow()
Creating a Flow from fixed values:
val statusFlow: Flow<String> = flowOf("Loading", "Success")Returning an empty Flow:
val emptyItems: Flow<Item> = emptyFlow()
These functions are useful, but they shouldn’t be used just to make every API return Flow.
The goal is not to maximize the number of Flow objects in an application. It’s to introduce reactive streams where they provide real value.
Putting the Responsibilities Together
The overall architecture can be summarized as follows:
Data Sources
|
+----------+----------+
| | |
Room Callback suspend API
| | |
Flow callbackFlow suspend
| | |
+----------+----------+
|
↓
Repository
|
transform / combine
|
Flow<T>
|
↓
ViewModel
|
map / combine / ...
|
stateIn()
|
StateFlow<UiState>
|
↓
UI
collectAsStateWithLifecycle()
One important detail: a suspend API doesn’t have to become Flow. It can remain a suspend function and be called from the Repository or ViewModel, depending on the responsibility.
The diagram shows the typical path for observable data, not a rule that every operation must follow.
Conclusion
Creating a Flow is easy. Deciding where to create it is the more important architectural decision.
When working with Kotlin Flow, think about the original source first:
- If the source already provides Flow, use it directly.
- If it uses callbacks or listeners, bridge it with callbackFlowclose to the API.
- If it provides a one-time result through a suspend function, keep it as suspend unless a stream is genuinely needed.
- If multiple sources need to be combined, use Flow operators in the appropriate Repository or ViewModel.
- When exposing derived UI state, convert the resulting Flow into StateFlow with stateIn().
The principle is straightforward:
Convert at the source. Transform in the Repository. Derive UI state in the ViewModel.
This isn’t about forcing every layer to use Flow. It’s about introducing Flow at the right architectural boundary, where it makes data changes easier to observe and the responsibilities of each layer easier to understand.
Enjoyed this article? Here’s how you can support my work:
- 👏 Clap (up to 50 times!) if you found this insightful.
- 👤 Follow me and turn on notifications 🔔 so you never miss the next deep dive.
- 📧 Subscribe to get my latest stories delivered directly to your inbox.
You can find all my articles on Android architecture, Jetpack Compose, and Kotlin in this curated reading list.
Don’t Just Create a Flow. Know Where to Create It. was originally published in ProAndroidDev on Medium, where people are continuing the conversation by highlighting and responding to this story.
