Flatmaplatest Resubscribe Composite Key

Re-subscribe on a key change, gate on a composite key

Two flows, at very different rates. One is the key — the item currently in view, changing rarely. The other is fast and continuous — a position, a size, a progress reading. Some one-shot work needs both, and needs to run once per item, not once per tick.

flatMapLatest gives the re-subscription: when the key changes, the inner collection is cancelled and rebuilt around the new key, so the pair can never mix a new key with the old stream. distinctUntilChanged over a composite of the fields that actually matter then collapses the fast stream down to "something meaningful changed":

// adapted — names generalized
currentItemState
    .filterNotNull()
    .flatMapLatest { item ->
        timeline.map { t -> Pair(t, item) }
    }.distinctUntilChanged { (oldT, oldItem), (newT, newItem) ->
        // true = equivalent = DROP
        oldT.total == newT.total && oldItem.entity?.id == newItem.entity?.id
    }.collectLatest { (timeline, item) ->
        if (timeline.total > 0 && item.entity != null) {
            if (screenData.value.extraA == null) fetchA(item.id, timeline.total)
            if (screenData.value.extraB == null) fetchB(item.entity, timeline.total)
        }
    }

The composite is duration plus item id, because the work needs a usable duration: the item id alone changes too early (the duration is still unknown, and the fetch would be issued with a zero), and the duration alone changes on rewind. Neither field on its own is the event.

Traps

The distinctUntilChanged predicate answers "are these equivalent", so true drops. Take the implementation apart and the branch is unambiguous: when the predicate returns true it jumps past the emit, and only a false result updates the remembered key and forwards downstream. Written as old != new — the reading that feels natural — the operator inverts and drops everything except the changes. The symptom is a side effect that fires on every tick, which is the exact bug the line was added to fix, so it is easy to conclude the operator "doesn't work".

Do not build the composite key by concatenating strings and hashing. The pattern that shows up in practice is (a.toString() + b).hashCode() on both sides. It is lossy twice over: concatenation is ambiguous ("1" + "23" and "12" + "3" produce the same string), and a hash collapses distinct values onto one integer. A missed change here is silent — the work simply never runs for that item. Compare a Pair, a data class, or the fields themselves; they are already comparable and cost nothing:

.distinctUntilChangedBy { (t, item) -> t.total to item.entity?.id }

distinctUntilChangedBy is the same operator with the key extracted instead of a predicate, and it removes the chance of inverting the comparison at all.

A field left out of the key is a whole class of update you will never receive. This is the mirror of the hashing mistake and quieter still: whatever the key omits, the operator treats as noise. A snapshot key of (item, isPlaying, position) taken from a state that also carries the list behind the item drops every emission in which only that list changed — so the collector never learns the list arrived, and the symptom is a feature that is simply absent rather than one that misfires. Give each field a cheap projection rather than leaving it out: ids rather than objects, so the comparison is structural and does not depend on the upstream reusing instances.

// adapted — names generalized
.map { Snapshot(it.item, it.isPlaying, it.position, listIds = it.list.map { e -> e.id }) }
.distinctUntilChanged()

The rule that makes this decidable without thinking about it: the key must mention every field the collector body reads. A field the body uses and the key ignores is a silent drop; a field the key mentions and the body ignores costs only a redundant pass. The asymmetry is the whole argument for erring towards a wider key — and it is why the guard in the next trap is not optional.

The collector still needs its own idempotence guard. The composite key changes for reasons unrelated to the work — a duration correction arriving late will re-fire it for the same item. Each branch above therefore re-checks that the result it would produce is still missing (if (screenData.value.extraA == null)). The gate reduces the firing rate; the guard is what makes a re-fire harmless. Skipping the guard because the gate exists is how a duplicate request survives into release.

collectLatest cancels the previous action; it does not cancel what that action started. The inner work above is launched into the surrounding scope and outlives the cancelled action. If the one-shot work must be abandoned when the key changes, hold it in a named handle and cancel it — see named-job-lifecycle-discipline — or check the item id again at the point where the result is written, which is the pattern in distinct-by-key-reset-cancel-per-item.

combine is not a substitute here, and neither is zip. combine re-emits on either source, so the fast flow drives it and the key flow contributes nothing structural — you get the same rate with no re-subscription. zip pairs emissions positionally and stalls the moment the two rates diverge, which they do immediately.

flatMapLatest requires an explicit opt-in. It is annotated experimental in the coroutines library, so the file carries @OptIn(ExperimentalCoroutinesApi::class). That annotation is a real statement about API stability, not boilerplate: the surrounding code should not depend on the operator's exact cancellation timing beyond "the previous inner collection is cancelled".

Order the operators: filter, then flatMapLatest, then distinct, then collect. Putting distinctUntilChanged before the flatMapLatest deduplicates the key flow, which is nearly always a no-op, and leaves the fast flow undamped. The gate belongs on the merged stream.

Verifying it

Hand-written keys and projected ones need a census each, because the bare no-argument form is not a hand-written key and never appears in the first list:

grep -rnE 'distinctUntilChanged *\{|distinctUntilChangedBy' --include='*.kt' . | grep -v '/build/'
grep -rn -A12 -E '^ *\.map \{$' --include='*.kt' . | grep -v '/build/' | grep -E '\.map \{$|distinctUntilChanged\(\)'

From the first, expect a handful of hits (plus an import line per file that uses the By form). Read each one against the body of its own collector and answer the two questions above in order: does true mean drop here, and does the key mention every field the body reads? A predicate that compares one id while the body reads three fields is the silent-drop defect, invisible at the call site.

The second is where an omitted field actually hides, because a multi-line .map { feeding a bare distinctUntilChanged() is a hand-written key with its fields spread over several lines. Only the pairs in that output count: a .map { with no distinctUntilChanged() under it is an ordinary transformation, so read past it. For each pair, open the projection and ask the second question again.

Then instrument the gate, not the collector: log the composite key on every emission that gets past distinctUntilChanged. Let one item play through without touching anything. The count must be small and bounded — one line when the duration becomes known, and nothing for the rest of the item. A line every tick means the predicate is inverted or the composite includes a field that changes continuously; no line at all for a change you know happened is the omitted field.

Then switch items fast, before the first item's work returns, and confirm two things: a new line appears for the new item, and no result belonging to the old item is written afterwards. The second half fails independently of the gate and is the one this operator does not fix by itself.