I’d like to convert a `Flow` to a `StateFlow` with...

# coroutines

[Mark]

I’d like to convert a Flow to a StateFlow without using a CoroutineScope. The idea is that the underlying Flow is only collected when the StateFlow is collected. Before that, the StateFlow would just act as an initial state holder. Is this possible?

[Sam]

I think you could achieve this with a MutableStateFlow. In the upstream flow, add an onEach block that updates the MutableStateFlow.

[Sam]

Wait... maybe that doesn't exactly match what you asked for 🤔

[efemoney]

Yes (ish), by implementing your own StateFlow subclass. (even though its not “safe” for extesion)

[efemoney]

something like:

class DerivedStateFlow<T>(
  private val getValue: () -> T,
  private val flow: Flow<T>,
) : StateFlow<T> {

  override val replayCache: List<T>
    get() = listOf(value)

  override val value: T
    get() = getValue()

  @InternalCoroutinesApi
  override suspend fun collect(collector: FlowCollector<T>): Nothing {
    flow.distinctUntilChanged().conflate().collect(collector)
    awaitCancellation()
  }
}

(credit @billjings)

[Mark]

Thanks, but I don’t understand the point behind getValue() here. It seems to be totally unrelated to flow. I would expect an initialValue: T to be passed in. And the collect() to return that value first. value would initially return initialValue and then the values as collected from flow.

[efemoney]

Ah youre right of course, I didnt do any code review, just copy & pasted 😄. Adjust to fulfil your needs 👍🏾

[efemoney]

(iirc this was a solution to mapping stateflows, so not exactly your use case; so that getValue was usually { otherStateFlow.map{…} }. )

[Mark]

Maybe this?

class DerivedStateFlow<T>(
    private val initialValue: T,
    private val flow: Flow<T>,
) : StateFlow<T> {

    override val replayCache: List<T>
        get() = listOf(_value)

    private var _value: T = initialValue

    override val value: T
        get() = _value

    override suspend fun collect(collector: FlowCollector<T>): Nothing {
        flow {
            emit(initialValue)
            emitAll(flow)
        }.distinctUntilChanged()
            .conflate()
            .onEach { _value = it }
            .collect(collector)
        awaitCancellation()
    }
}

[franztesca]

My attempt: https://pl.kotl.in/GmxEkWWiz
You still need a coroutine scope to collect the upstream though. That is the correct way of modelling a stateflow, which is not a cold flow.
What if there are two subscribers to your StateFlow?. When should the upstream flow be collected and when should it be cancelled? Without specifying the scope of the upstream collection it's hard to adhere to the stateflow sematics.

[Mark]

@franztesca can’t you avoid the class declaration? https://pl.kotl.in/JGzmBaMMl but generally I’m not convinced how useful a MutableStateFlow is in this case. Surely you would want a StateFlow only, in which case just use stateIn

[franztesca]

Yes, you can avoid it indeed kodee happy
For the mutability, I though you needed to hold a (mutable) initial state and wanted to do that in the same state flow.

Before that, the StateFlow would just act as an initial state holder

If you don't need it to be mutable, then using the built-in stateIn sounds like a better option.

[Mark]

I see the confusion. I meant more like “immutable initial-state holder”. I don’t like the naming of state for both UiState and compose State. For example, how naming a compose State property (when not using by) can lead to uiStateState . Or is that just me?

[uli]

Not tested, but this might work (2nd update)

class FlowStateFlow<T> private constructor(
    private val flow: Flow<T>,
    private val stateFlow: MutableStateFlow<T>,
) : StateFlow<T> by stateFlow {

    constructor(
        flow: Flow<T>,
        initial: T,
    ) : this(flow, MutableStateFlow(initial))

    override suspend fun collect(collector: FlowCollector<T>): Nothing {
       coroutineScope {
            launch {
                flow.collect {
                    stateFlow.value = it
                }
            }
            stateFlow.collect {
                collector.emit(it)
            }
        }
    }
}

You might want to launch on Dispatchers.Default to collect the original flow in the background. I am just not sure if it is worth it. Would be interesting to see the behaviour under load.