hey all — i have a pretty common construction where i’m in an Android ViewModel and need to observe a Flow, do some stuff, and then push the result into a 

MutableStateFlow

. Looks something like this:

fun whatever() { 
	viewModelScope.launch { 
		inputFlow
			.map {...}
			.collect { value -> 
				mutableStateFlow.value = value
			}
	}	
}

(Assume that I need to push other values into the 

MSF

 outside this 

collect

 block) One issue I’ve run into with the above construction is that when the observer of the

MSF

cancels collection (e.g. the Fragment observing the ViewModel falls off the screen), it doesn’t cancel this intermediate collection…this makes absolute sense — that cancellation signal is not being propagated upstream! So digging into the

stateIn

sources, I’ve come up with what I think is a way to push the cancellation “upstream” — can y’all give thoughts/comments/suggestions on the following approach?

fun <T> Flow<T>.stateInto(downstream: MutableStateFlow<T>, scope: CoroutineScope) {
    val upstream = this
    scope.launch {
        downstream.subscriptionCount
            .mapLatest { count -> //this is a simplified version of [StartedWhileSubscribed#command]
                if (count > 0) true
                else {
                    delay(800.milliseconds)
                    false
                }
            }
            .dropWhile { active -> !active }
            .distinctUntilChanged()
            .collectLatest { active ->
                when (active) {
                    true -> upstream.collect(downstream) //will be cancelled upon new emission
                    false -> Unit /*just cancel and do nothing*/ 
                }
            }
    }
}

Personally, in these situations I would always try to think in the reactive paradigm first. That's a pretty abstract statement so let's break it down: By collecting one flow and mutating another, you're kind of creating this imperative 'break' in what could otherwise be one continuous Flow.

yeah I can’t combine/merge them in because they’re not Flows, they’re single methods that are driven by user interaction that push new state into the MutableStateFlow. (Also FWIW I’ve been doing Rx since 2015, so I’m familiar with the notion of “no side effects! pure functions only!” but that’s not always realistic in the real world 😕 ) oh yeah btw if you find the proposed code a bit odd, it’s actually distilled (and structurally the same) from

Flow.stateIn(...)

Ah yeah fair enough 👍 I just found this is too often the answer to peoples probs in the past ☝️

Sometimes I'll turn use a MutableState/SharedFlow to represent the method calls and then I can go full reactive. Seems like something akin to this could work for you:

private val nonInputStateFlow = MutableStateFlow<ValueType>
val outputFlow = merge(inputFlow.map {...}, nonInputStateFlow)
fun otherMethod1(value: Value) {
    otherMethodFlow.tryEmit(value)
}
fun otherMethod2(somethingElse: SomethingElse) {
    otherMethodFlow.tryEmit(calculateValue(somethingElse))
}