Is it bad practice to use

MutableStateFlow

to replace

@Volatile var foo: Foo = defaultFoo()

? My main reasoning is that I don't like

var

and it would be easier for multiplatform etc...

With Flow what is the "nicest" way to chain two network requests where the second depend on data from the first?

We're thinking about adding guidance in Android about how to expose events/effects from a viewmodel and we need a flow that : • has a way to limit its capacity (usually 2-5) • replays everything in its capacity when going to 0 to 1 collectors but does not replay anything from 1 to 2 and beyond (as they're already consumed) We looked at

SharedFlow

(but it replays everything to a second observer),

Channel(CONFLATED).*receiveAsFlow*

but only one observer gets updates (and we discarded

StateFlow

,

consumeAsFlow()

, and

.broadcast()

/cc @elizarov @Manuel Vivo

I have two data sources which I want to combine:

private val workouts = flow { emit(userRepository.getWorkouts()) }
private val exercises = flow { emit(userRepository.getExercises()) }

val workoutsByExerciseId: Flow<Map<String, Workout>> = ? // do some transformation which combines both sources

I want to create

workoutsByExerciseId

which should depend on both

workouts

and

exercises

. If either source changes,

workoutsByExerciseId

should always re-emit with the latest data. Which flow operator do I need here?

I’m looking to convert some

LiveData

to

StateFlow

. What would be the equivalent flow construct for the

CoroutineLiveData

builder function?

val state = liveData {
    emit(loadState())
}

suspend fun loadState(): State = withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
    ...
}

The closest I can determine is:

val state = flow {
    emit(loadState())
}.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), null)

Except that

stateIn()

requires an initial value, which doesn’t make this quite the same, as I don’t want any value collected until the state is loaded. Another possibility might be to use the

shareIn()

operator to create a

SharedFlow

with replay set to 1. This might be a closer equivalent in some ways:

val state = flow {
    emit(loadState())
}.shareIn(viewModelScope, SharingStarted.WhileSubscribed(5000), replay = 1)

Except

SharedFlow

doesn’t provide a

value

property to get the current state. Which makes the

StateFlow

construct with a nullable type a closer equivalent in this regard.

I’m using

StateFlow

in a Kotlin/React project and noticed an interesting nuance that may make it unsuitable for providing state in React-like projects - and maybe others. Am I missing something or do I have an error in my thinking here? In my case the problem crops up in the sign-in process. To simplify, there are just two pages: - The

MembershipPage

shows information about the signed-in user. If no user is signed in, redirect to

SignInPage

. - The

SignInPage

allows the user to sign in. If a user is already signed in, redirect to

MembershipPage

. A single

UserController

component wraps these (and other) user-related pages, keeps track of the signed-in user and provides that value to pages, and performs the redirects if needed. Various React components keep track of the signed-in user by accessing

StateFlow.value

when first rendering the component, and collect

StateFlow

emissions of new values to account for changes. Now when the user signs in the following happens: 1. Send credentials to server. 2. Server responds with OK and user object. 3.

StateFlow

value is set to signed-in user (

flow.value = user

). 4.

SignInPage

redirects to

MembershipPage

. (user has signed in successfully) 5.

UserController

redirects to

SignInPage

. (not signed in) 6.

SignInPage

redirects to

MembershipPage

. (already signed in) 7.

UserController

redirects to

SignInPage

. (not signed in) 8. <repeats a few times> 9.

SignInPage

redirects to

MembershipPage

. (already signed in) 10.

MembershipPage

is presented. There’s a redirect cycle that gets broken automatically after a short time. What’s happening here? After a successful sign-in the `StateFlow`’s

value

is set to the signed-in user. From that moment on

flow.value

returns the signed-in user that was just set. The `StateFlow`’s emission of that new value however is not immediate and takes a while. During steps 4 through 8 above there is an inconsistent state in the React components. Those that happen to use

StateFlow.value

acknowledge that a user is signed in. Those that wait for the `StateFlow`’s emission of a new value still consider no user to be signed in. Depending on whether a component is first rendered (accessing

.value

) or rerendered (value depends on

.onEach/.collect

) the behavior is different. So by using

StateFlow

the system inadverently depends on two different sources of truth for the signed-in state. That’s because the return value of

.value

and

StateFlow

emissions of new value are not synchronized. I’m not sure if that’s intentional behavior of

StateFlow

. State should have a single source of truth and

StateFlow

itself is inconsistent here. One potential solution would be to change how

StateFlow

works. For example by having

.value

to not return a new value until the emission of the new value has begun. There’s still a potential for race between different contexts but at least in synchronous environments like JS that should be irrelevant. There are two possible approaches to use

StateFlow

as-is as a single source of truth: (1) Rely on

.value

only. (2) Rely on value emissions only. (1) doesn’t work because components need to know when the value changes in order to re-render. Otherwise the UI would get out of sync with the state. (2) doesn’t work because components need to access the current value synchronously at the time of initial render. Otherwise the UI would always have to briefly display a loading state while waiting for the initial emission even though that value is already available synchronously. As a compromise I could always rely on

.value

for accessing the current value and use emissions only for triggering a re-render. An inconsistency still remains however. Components aren’t notified immediately when the state they depend on changes. They either render a state that’s potentially newer than the state they were last notified about, or they render a state change too late, and they may render more often than needed. If the

SignInPage

executes some logic as a result of a successful sign-in it does so under the assumption that the user is signed in now. Other components however still assume that no user is signed in. No re-render is pending for them because there’s no notification about a change in state yet. Hence the UI is temporarily inconsistent. I don’t think that this subtle inconsistency between

.value

and emissions only affects Kotlin/React projects. It may also affects other projects which rely on a mix of accessing

.value

and emitted values. If it’s intentional then it should be communicated prominently in the documentation. However for me it would defeat the purpose of a

StateFlow

for managing state because it’s not a single source of truth. I could also write a wrapper that may use a

StateFlow

under the hood but synchronizes the current value with the emissions. But that’s just reinventing what I assume

StateFlow

is supposed to provide in the first place. What are your thoughts on that?

Hi everyone. I have a

MutableStateFlow

and I need to know when collectors subscribe/unsubscribe to/from this flow. I see the

fun SharedFlow<T>.onSubscription

extension which invokes the callback when the flow starts to be collected but I can’t find any extension function to know when a collector cancels the subscription. I’m basically looking for the equivalent of RxJava doOnSubscribe and doOnDispose. Does anyone know if there is a way to know when a collector cancels its subscription?

Can I receive most recently two value when a new value emit? for example: emit: 1, 2, 3, 4, 5, 6 downstream receive: [initValue, 1] [1, 2] [2, 3] [3, 4] [4, 5] [5, 6]

How can I propagate an upstream exception in a SharedFlow? For example, I have an Android screen which combines multiple data sources:

val uiState = combine(sharedFlow1, sharedFlow2, sharedFlow2) { … }

and I would like to catch anything which goes wrong in any of the flows. How could I achieve that?

.catch

doesnt work because shared flows never complete

I searched for this everywhere and I'm sure the answer is simple but... how do I check that a flow is empty inside

runBlockingTest

?

firstOrNull()

shows

IllegalStateException: This job has not completed yet

Hi everyone, I’m trying to find the best way to convert a

LiveData

that represents a UI state into a

StateFlow

. My initial live data code looked like this:

private val _state = MediatorLiveData<State>().apply {
    addSource(livedata2) { value2 -> value?.let { it.copy(someVariable = value2) } }
    value = State(someVariable = "initial value", someOtherVariable = "another initial value")
}
val state: LiveData<State> = _state

The idea is that I have an initial value and also part of the state gets updated whenever a new value is triggered on

livedata2

To reach the same behavior using flows, I modified

livedata2

to be a SharedFlow (I’ll call it

flow2

) and my

_state

became a StateFlow. However I have to

collect

the flow2 instead of just applying an operator.

private val _state = MutableStateFlow(
        State(
            someVariable = "initial value",
            someOtherVariable = "another intial value"
        )
    ).apply {
        viewModelScope.launch {
            flow2.collect { value2 -> emit(value.copy(someVariable = value2)) }
        }
    }

val state: StateFlow<State> = _state

While it does work, I feel like this is probably not the way to go? Maybe there’s an operator for that. (I tried to use

_state.combine(flow2)

but my state was not being updated for new events on

flow2

) I there a better way to achieve the same result?

Hello! I have MutableStateFlow (with string data as state) in PhoneStateListener class. I want to collect this flow in another class, where listener provides in constructor (repository). But this works only first time.

Hi everyone, is there a way to get the the previously emmited value of a flow in the onEach function, or something like this?

private val refreshJob = filterFlow
    .onEach { prev, new ->
        if(prev.x != new.x) {
            updateX()
        }
    }
    .launchIn(viewModelScope)

I know I can separate the filterFlow in multiple flows, which is probably the better way of solving my use case, but I'm still curious

What is the right way to testing a

MutableSharedFlow

?

@Test
    fun testIntSharedFlow() = runBlockingTest {
        val intSharedFlow = MutableSharedFlow<Int>()

        intSharedFlow.emit(5)

        assertEquals(5, intSharedFlow.single())
    }

This code results

java.lang.IllegalStateException: This job has not completed yet

If I update a

StateFlow

value from

Dispatchers.Main

, and have some separate code collecting the flow also on

Dispatchers.Main

, can I guarantee that the collector receives the value immediately (synchronously) after the state flow emits?

Hi everyone! Huge respect to JB! I have some question about StateFlow. What if I have 2 stateFlows and non-suspend transformer function, and I’d like to create a new StateFlow like a combination of theese two, transformed by the transformer function? It seems like current api does not allow to create such a StateFlow<>.combine operator? (sorry for bad English)

Say I have a Flow of Booleans and only want to emit when they flip, what is the proper way to do that? What am I looking for?

suspend fun Flow<Boolean>.flippedTo(value: Boolean, block: () -> Unit) {
    var previous: Boolean? = null
    collect { current ->
        if (previous == !value && current == value) {
            block()
        }
        previous = current
    }
}
boolFlow.flippedTo(false) {
    println("Flipped from true to false")
}
boolFlow.flippedTo(true) {
    println("Flipped from false to true")
}

Hi Everyone ! We have a legacy Android app in Java. Planning to rewrite some parts (DB and repository layer) in Kotlin. At one place , we have to access Kotlin Flow from java code. can you guys please direct me to any resource/link which talks about accessing Kotlin Flow from Java code ?

One of my teammates ran into the following problem, which confused me at first until I realized what he did When you type out the code to collect on a flow (fully type out

collect {}

without auto-importing it), it'll default to the

Flow.collect()

method, which should be expected since the extension function

collect {}

was not imported. The problem is that when you do this, there's no option to automatically fix the issue unless you manually add the import statement, or re-type in "collect" but allowing auto-import to do its job. Now that I have seen this issue, I know how I can help my teammates next time, as we continue working with Coroutines/Flow. However, this was a minor point of frustration... So my question is: Is there any plan to rename or remove the

Flow.collect

function so this doesn't happen again? Screenshot is an example of how unhelpful lint is when you find yourself using

Flow.collect()

by accident

Hi 👋🏼 How can I replace Rx operator

repeat

with some Flow operator?

Not entirely clear what you want but here's a

Flow

that emits something

n

times.

flow { repeat(n) { emit(theThing) } }

Hey folks 👋. I have a use-case where I construct a Flow that uses resources that can be expensive to initiate (e.g. queries from a database) but eventually completes (is not a continuous stream of data). I was hoping that I could use the

shareIn

operator to ensure the flow is only actually collected once, and subsequently cached. Based on what I've read, due to the fact that

collect

on a

SharedFlow

never actually returns, the recommended pattern is to use some kind of sentinel value to denote the completion of data and combine that with the

takeWhile

operator. Just wanted to check this is indeed the recommended solution.

Im trying to switch between flows depending on another flow 🙂

val result = state.transformLatest {
            when (it) {
                "a" -> emitAll(emitter("A"))
                "c" -> emitAll(emitter("C"))
            }
        }

        result.collect {
            println(it)
        }

this will print a flow of A+<counter> or “C” +<counter> depending on the last value from “state”. I would like to ignore “unknown” states, but since transformLatest cancels the previous flow even if i don’t emit anything new, setting the state to “B” will just cancel the output, not continue on the previous selected output. Is there a nice way to achieve what I want?

I want MutableSharedFlow to keep the values 1, 2 in it’s buffer if no one is collecting it.

Hi 👋 I'm learning flows, and I have the following question. Is it possible for a flow to be collected by two different coroutines concurrently? I want the following properties • a collector shouldn't affect the other one • I want the flow to be able to use

Conflation

in case one of the collector is being to slow • flow is generated once (meaning I don't want to compute each item in the flow twice) I was able to do this using broadcast channel and consuming it as channel like

openSubscription().consumeAsFlow().conflate()

. Is is possible to do something similar to this using flows? Thanks!

Hi, what would be the best approach to unsubscribe/cancel a flow once a certain item/condition is reached, but returning that last item: transformWhile?

/**
 * Implements function from
 * [RxJava](<http://reactivex.io/RxJava/3.x/javadoc/io/reactivex/rxjava3/core/Observable.html#throttleLatest-long-java.util.concurrent.TimeUnit->)
 */
internal fun <T> Flow<T>.throttleLatest(timeout: Duration) =
    conflate().transform {
        emit(it);
        delay(timeout);
    }

Hey, am I missing any flaws in this implementation? Thx 👋

Hi guys 🙂 I just had a small argument with myself regarding how to expose a flow in an interface. The implementation is backed by a shared flow. Obviously, no

MutableSharedFlow

exposed. But would you have a

SharedFlow

exposed? Since the consumer may need to know that it’s hot? Or a regular

Flow

? Since the consumer is just going to ‘.collect()’ anyway? Or is it a case by case discussion?

Hi 👋 does anyone know why asSharedFlow function may not be found during compilation? kotlin version 1.5.20, kotlinx-coroutines-core tried 1.4.1 and 1.5.0