Request: Map MutableStateFlow to StateFlow

[andreimesina]

Hi, I am trying to convert the generic type of a MutableStateFlow and expose it as a StateFlow using the map operator.

Unfortunately, map can only return a Flow<R>:

I also imagine that adding as StateFlow<Group after the map isn't really a good idea.
So, is there any solution for this case right now, or can we have a map that returns StateFlow?

[psteiger]

This has been discussed in #2008. Looks like the current way to deal with this is by using stateIn after map.

[andreimesina]

What if for instance we need to expose the StateFlow from a Repository pattern component and consume it upper in the hierarchy, in a ViewModel?

We can't hold a CoroutineScope in the Repository to pass it to the stateIn operator, as the scope belongs to the ViewModel instead... is my point of view wrong?

[psteiger]

@andreimesina you can use a scope that matches the lifecycle of your repositories. In my case, the repositories are singletons, scoped to the whole application, and thus I use ProcessLifecycleOwner.get().lifecycleScope as the CoroutineScope for shareIn and stateIn.

Alternatively you can just construct your own CoroutineScope objects and use them. However, you might miss Lifecycle callbacks in your Repositories, either for canceling the scope, such as you would in onDestroy or onCleared, or even pausing/resuming the coroutines when your app is in background/foreground, such as onStart and onStop. Using scopes from Lifecycles (LifecycleCoroutineScope) handle that automatically.

If ProcessLifecycleOwner does not fit your repositories for some reason (e.g. they are not scoped to the whole app), you can still have a LifecycleCoroutineScope if you are able to make your repositories implement the LifecycleOwner interface, return a LifecycleRegistry in the getLifecycle() override, and somehow trigger the Lifecycle callbacks at the right moments (e.g trigger onStart when repo is in use and onStop when it isn't).

(Every Lifecycle object, such as LifecycleRegistry has a LifecycleScope)

My personal suggestion would be to build your architecture around the scopes automatically provided by LifecycleOwner components as it makes managing scopes easier.

[psteiger]

@andreimesina by the way, why not just using asStateFlow()?

[andreimesina]

@psteiger asStateFlow() is an extension on MutableStateFlow, whereas map and other operators return a Flow.

So we cannot do

val mutableStateFlow = ...

val stateFlow = mutableStateFlow.map {
    ...
}.asStateFlow()

Because there is no asStateFlow() for a simple Flow.

[qwwdfsad]

It is indeed not there because it has a lot of open design questions.

E.g. who should invoke the map operator, each collector or the upstream? Should it depend on whether the upstream state flow is a standalone object (just created via StateFlow(...) and emitted via flow.value = 42) or shared via stateIn?
What if mapping lambda suspends in the middle, how should the system behave?
All these questions have controversial answers that conflict with each other.

As the potential solution, I'd recommend either using stateIn operator or basic delegation mechanism:

class MappedStateFlow<T, R>(private val source: StateFlow<T>, private val mapper: (T) -> R) : StateFlow<R> {

    override val value: R
        get() = mapper(source.value)

    override val replayCache: List<R>
        get() = source.replayCache.map(mapper)

    override suspend fun collect(collector: FlowCollector<R>) {
        source.collect(object : FlowCollector<T> {
            override suspend fun emit(value: T) {
                collector.emit(mapper(value))
            }
        })
    }
}

[pacher]

@qwwdfsad Basic delegation is not very good solution since StateFlow is not stable for inheritance.

[GrazianoRizzi]

something like that?

fun <T, M> StateFlow<T>.map(
    coroutineScope : CoroutineScope,
    mapper : (value : T) -> M
) : StateFlow<M> {
    val mFlow = MutableStateFlow(
        mapper(value)
    )
    coroutineScope.launch {
        collect {
            mFlow.emit(mapper(it))
        }
    }
    return mFlow
}

class NavigationViewModel(
    private val archipeppeUseCase : ArchipeppeUseCase
) : ViewModel() {

    val state : StateFlow<NavigationState> = archipeppeUseCase.state.map(viewModelScope, ::reduce)

    private fun reduce(
        archipeppeState : ArchipeppeState
    ) : NavigationState = when (archipeppeState.actualView) {
        Views.FIRST_VIEW -> NavigationState.FirstView
        Views.SECOND_VIEW -> NavigationState.SecondView
    }
}

[psteiger]

something like that?

fun <T, M> StateFlow<T>.map(

    coroutineScope : CoroutineScope,

    mapper : (value : T) -> M

) : StateFlow<M> {

    val mFlow = MutableStateFlow(

        mapper(value)

    )

    coroutineScope.launch {

        collect {

            mFlow.emit(mapper(it))

        }

    }

    return mFlow

}

Looks to me this is roughly (if not exactly) equivalent to:

val stateFlow = originalStateFlow
    .map { mapper(it) }
    .stateIn(
        coroutineScope,
        SharingStarted.Eagerly,
        mapper(originalStateFlow.value)
    )

[GrazianoRizzi]

Yes it is the same, I did not know this version of stateIn, the others are suspend fun and these are not suitable for use with Compose and the collectAsState fun. So we could opt for a solution like this to compact the expression:

fun <T, M> StateFlow<T>.map(
    coroutineScope : CoroutineScope,
    mapper : (value : T) -> M
) : StateFlow<M> = map { mapper(it) }.stateIn(
    coroutineScope,
    SharingStarted.Eagerly,
    mapper(value)
)

[ansman]

Please note that although @qwwdfsad's mapped flow appears to work fine, it has a critical flaw that makes it behave differently than a real state flow, namely that only distinct items are emitted. So for example, if you have an upstream that emits a number and a downstream that emits if it's even or not, changing the upstream from 1 to 3 would emit false twice in a row. This might not matter for most people but might cause subtle issues.

An improved version might be:

private class MappedStateFlow<T, R>(
    private val upstream: StateFlow<T>,
    private val mapper: (T) -> R,
) : StateFlow<R> {
    override val replayCache: List<R>
        get() = listOf(value)

    override val value: R
        get() = mapper(upstream.value)

    override suspend fun collect(collector: FlowCollector<R>): Nothing {
        var previous: Any? = Unset
        upstream.collect {
            val value = mapper(it)
            if (value != previous) {
                previous = value
                collector.emit(value)
            }
        }
    }

    private object Unset
}

[silverhammermba]

@ansman I really like that solution. I recommend pairing it with a public extension function so it can be used everywhere without knowing about the wrapper class

fun <T, R> StateFlow<T>.stateMap(transform: (T) -> R): StateFlow<R> {
    return MappedStateFlow(this, transform)
}

A couple differences from the built-in map is that the transform cannot be inlined due to the wrapper and the transform cannot suspend due to the need to get a non-suspending value from the wrapper.