ReadThat 10: Kotlin Flows in Practice

Part 10 of the ReadThat case study.

📲 Try it live: Download the ReadThat APK (8.8 MB, Android 8+). Open the file on your phone and allow “install unknown apps” if prompted.

Flows are the wiring of the whole client: Room emits, repositories transform, ViewModels derive, Compose collects. ReadThat also keeps a standalone :flows module, 29 tests of the patterns in isolation, deliberately outside the app graph, and this page pairs those distilled forms with the production call sites.

Example 1: Observing UiState with combine(...).stateIn(...)

The comments screen is the canonical case. One immutable CommentsUiState, derived from every input it depends on, but in two stages, because the inputs have two speeds:

1
// Stage 1 — the expensive derivation. Runs only when structure or collapse
2
// actually changes, and runs off the main thread.
3
private val render = combine(tree, userCollapsed, autoCollapsed) { t, user, auto ->
4
    t?.let { CommentFlattener.flatten(it, user + auto) }
5
        ?: CommentRenderList(emptyList(), 0, 0)
6
}.flowOn(flattenDispatcher)
7

8
// Stage 2 — cheap assembly. Flag churn re-executes THIS, never the flatten;
9
// the render instance passes through untouched, so LazyColumn's inputs stay
10
// referentially stable across spinner ticks.
11
val uiState: StateFlow<CommentsUiState> =
12
    combine(render, tree, loading, loadMoreStates, header) { r, t, flags, more, h ->
13
        CommentsUiState(render = r, header = h, loadMoreStates = more, ...)
14
    }.stateIn(
15
        scope = viewModelScope,
16
        started = SharingStarted.WhileSubscribed(STOP_TIMEOUT_MS),
17
        initialValue = CommentsUiState(),
18
    )

CommentsViewModel.kt#L105-L133

Why each piece is there:

• One state object, not a bundle of flows: an emission can’t pair new rows with a stale loading flag.
• WhileSubscribed(5_000): the only correct SharingStarted for UI. Eagerly runs while backgrounded forever; Lazily never stops; the 5-second grace window spans a rotation so the upstream doesn’t restart and re-fetch. This only works because the UI collects with collectAsStateWithLifecycle(): plain collectAsState() keeps collecting in the background and the timeout never fires.
• The two-stage split is the production-grade move: toggling a collapse re-derives the render list without refetching anything (there’s a test asserting the network call count is unchanged), and a spinner tick can’t trigger a 5,000-row flatten.

A test pins the flatten count too: flag churn executes stage 2 only. That’s a property you can only assert because derivation is a flow graph, not ad-hoc mutation.

Example 2: Observing Room

Room queries returning Flow/PagingSource are the mechanism that makes “Room is the source of truth” reactive rather than aspirational. The feed path:

1
@Transaction
2
@Query(
3
    """SELECT * FROM feed_groups
4
       WHERE accountId = :accountId AND feedId = :feedId
5
       ORDER BY sortIndex ASC"""
6
)
7
fun pagingSource(accountId: String, feedId: String): PagingSource<Int, GroupWithState>

FeedDao.kt#L17-L46 (condensed)

Room invalidates this PagingSource on any write to either joined table. That single fact powers the app’s best trick, the optimistic vote:

1
toggleLike(itemId)
2
  └─ one Room transaction: update item_state + insert vote_outbox
3
       └─ Room invalidates the PagingSource
4
            └─ Paging re-emits the affected page
5
                 └─ Compose recomposes the one changed row

No manual list surgery, no second in-memory copy to keep consistent, and the same write is instantly visible in feed, MediaFeed, and detail because they all observe the same tables. The ViewModel’s entire exposure is one line: repository.feed().cachedIn(viewModelScope): and cachedIn matters: without it, rotation re-collects the Pager from scratch, refetching page one and losing scroll position (FeedViewModel.kt#L29-L39).

Non-paging Room observation follows the same shape elsewhere: the community drawer renders its account-scoped Room snapshot immediately from a DAO flow, then silently refreshes behind it.

Example 3: A more complex combine (three sources, one snapshot)

From the :flows module, the distilled multi-source pattern: posts filtered by settings, joined with connectivity:

1
fun visiblePosts(): Flow<List<Post>> =
2
    combine(local.cachedPosts, local.settings) { posts, settings ->
3
        posts.filterNot { it.subreddit in settings.blockedSubreddits }
4
    }.distinctUntilChanged()
5

6
fun feedSnapshot(connectivity: Flow<Connectivity>): Flow<FeedSnapshot> =
7
    combine(visiblePosts(), local.settings, connectivity) { posts, settings, network ->
8
        FeedSnapshot(
9
            posts = posts,
10
            settings = settings,
11
            connectivity = network,
12
            // A derived field computed in one place instead of in every collector.
13
            canAutoplay = settings.autoplayVideo && network == Connectivity.ONLINE,
14
        )
15
    }.distinctUntilChanged()

flows FeedRepository.kt#L37-L61

The traps this shape defuses, each pinned by a test:

• combine emits nothing until every input has emitted. One never-emitting input silently stalls the whole graph. It is the single most common combine bug. Every input here is a StateFlow (always has a value) or emits an initial value up front. Tested explicitly with emptyFlow().
• distinctUntilChanged stops churn that doesn’t change the result: a settings write that blocks a subreddit not in the list produces no downstream emission (pinned with Turbine’s expectNoEvents()).
• Derived fields live in the combine (canAutoplay), not in N collectors that would each re-implement the rule slightly differently.

The connectivity input is itself the fourth foundational pattern: a listener API bridged with callbackFlow, using trySend (callbacks aren’t suspend contexts), an initial value pushed before waiting, and awaitClose { unregister }. Omit that and you leak the listener; there’s a test asserting the listener count returns to zero after cancellation (Sources.kt#L84-L90). This is exactly how metered-network state reaches the preload policy in part 9.

Honorable mention: the search pipeline

Every operator load-bearing, straight from flows/SearchViewModel.kt#L26-L82:

1
query
2
  .debounce(300)             // 4 rapid keystrokes -> 1 network call (tested)
3
  .distinctUntilChanged()    // cursor-move IME churn must not re-query
4
  .flatMapLatest { search }  // a new query CANCELS the in-flight one
5
  //   catch INSIDE the inner flow: a failed search kills that query,
6
  //   not the pipeline for the rest of the session
7
  .stateIn(...)

flatMapLatest is the important choice: flatMapConcat would queue stale results in order; flatMapMerge would race them and let the slowest response win. The production search feature runs this shape over its Paging mediator (SearchViewModel.kt).

Full pattern catalog: cold vs hot, Channel vs SharedFlow vs StateFlow for one-shot events, retryWhen/catch/flowOn ordering, virtual-time testing: flows/README.md.

← Part 9: Networking deep dive · Next: Part 11: Kotlin Multiplatform →