Does anyone know any any example code for hooking up Channels/Flows to a socket? I.e., I have a socket in a WearOS 3 app I want to wrap in a receive Channel that is a producer for various flows in viewmodels, and a send channel that the viewmodels can use in callbacks.

Hey guys, I am using

MutableStateFlow

in my project. When we initialise the

MutableStateFlow

object we need to give default value.

val topics = MutableStateFlow<List<String>>(emptyList())

when I collect this value

[null, "Hello", "world"]

I want to pass this list in

adapter

. So is there a way we can remove the null object before passing in adapter or Is there any better way ?

viewModel.topics.collect { topicsList ->
    println(topicsList)         // [null, "Hello", "world"]
    adapter.submitList(topicsList)
}

Is it required to use the repeatOnLifeycle with state flow collecting in the UI ? As I am updating the UI only when it is visible on the screen.

In my App, I have an AppState. AppState contains

currentUser = Flow<User>

Would this be better represented as a StateFlow?

hey, I would need help with custom flow transformation. I would basically need something like

distinctUntilChanged()

, with the exception, to re-emit same value, after certain amount of time has passed. How would I achieve that?

time [sec], input values, emitted value
0, "A", "A"
1, "A", <no emission>
2, "A", <no emission>
3, "A", "A"
4, "B", "B"

If I turn regular Flows into StateFlows

stateFlow1

and

stateFlow2

via

.stateIn(started = SharingStarted.WhileSubscribed)

and then create

stateFlowCombined = stateFlow1.combine(stateFlow2).startIn(started = SharingStarted.WhileSubscribed)

then will subscribing to

stateFlowCombined

transitively subscribe to

stateFlow1

and

stateFlow2

? I'm struggling with writing unit tests to verify behavior that works fine in our production app.

How to reflect a flow value when it gets ready after collecting?

var property: String? = null
.......

propertyFlow.collect{
    property = it
}

.......

textView.text = property // reflect this value when its get ready

Something like that, any idea?

I have a case where I am assigning the stateflow in viewmodel to the usecase stateflow like below class TestModel: ViewModel() { private var discoverTag = MutableStateFlow(false)

fun initiate(){
viewModelScope.launch {

discoverTag = usecase.tag
}
}

Where

tag

is stateflow in usecase which emits the value The problem here is that the discover tag is only collected for the first time but not when emitted from usecase

This question didn't get much traction in #coroutines so i figure that id ask here. I'm joining a project that has a bit of a flow hell going on. Bunch of flows started off of other flows. collect instead of collectLatest being used in some places. I was just wondering if anyone had any debugging tips or something to be able to tell how many flowables are still being collected. what hot flowables are still emitting without their being an observer. etc. Are logs my only hope here?

Is there a real way to recollect the

Flow

each x seconds? like:

val flow = flow {
    emit(doApiCall())
}
.onEach{ delay(1000) }// something like that
........
flow.collect{
  //get flow value
}

I searched a lot but it seems that I need to put the code in fun and call it each time, what do you think?

I have a flow that emits when value changes. That value is mapped to a value with optional expiration date. I want to emit recalculated value when the last value expires. My current solution is recursive and I don’t really like that.

fun myFlow(): Flow<B> =
  valueFlow()
    .map { B(it) } // B is the type with expiration date
    .flatMapLatest { value ->
      flow {
        emit(value)
        value.timeToExpiration?.let {
          delay(it)
          emitAll(myFlow()) // !recursion
        }
      }
    }

Any hints how to remove recursion? Thx.

This may be a stupid question, but is there a concise way to convert a List<Flow<T>> to a Flow<List<T>> that emits a value when any flow in the list emits a value?

When I call MutableStateFlow.update {}, isn’t function() called on the same thread? I’m running into an issue when my code looks like the pseudocode in the 🧵 [Solved] function() is called on the same thread, some other code changed my variable that I missed before calling state.update

hi , It ok to do this in a single launch ? or do we have to use separate launch for repeat on life cylce ?

lifecycleScope.launch {

    lifecycle.repeatOnLifecycle(Lifecycle.State.STARTED) {
        loginAuthorizationViewModel.onLoginSuccessful.collect {
            startActivity(HomeActivity.getDefaultIntent(this@LoginActivity))
            finish()
        }
    }

    lifecycle.repeatOnLifecycle(Lifecycle.State.STARTED) {
        loginInputValidationViewModel.onPasswordInvalid.collect {
            enableViews(true)
            et_password.error = it
        }
    }

    lifecycle.repeatOnLifecycle(Lifecycle.State.STARTED) {
        loginAuthorizationViewModel.onLoginFailure.collect {
            enableViews(true)
            showMessage(it.toString())
        }
    }

    lifecycle.repeatOnLifecycle(Lifecycle.State.STARTED) {
        loginInputValidationViewModel.onUserNameInvalid.collect {
            enableViews(true)
            tv_user_name.error = it
        }
    }

    lifecycle.repeatOnLifecycle(Lifecycle.State.STARTED) {
        loginInputValidationViewModel.onValidationSuccessful.collect {
            loginAuthorizationViewModel.requestLogin(
                userName = it.first,
                password = it.second
            )
        }
    }

Hi there! I'm having issues trying to implement caching with Flow. Requirements: return a cached value and then follow up with network request result. The cached value must be observed regardless of network call success or failure. Implementation looks like

fun data(): Flow = flowOf(
    flowOf(cachedValue),
    networkCallFlow
). flattenConcat()

which is then observed in a ViewModel and mapped to UI state

data()
.onEach { .. }
.map { uiState(data) }
.catch { emit(uiState(error)) }

It works fine when the network call succeeds. The problem is, when

networkCallFlow

throws an exception (no network), the

cachedValue

is never mapped, however it triggers the

onEach

handler. My guess is that the emission of the

cachedValue

does not guarantee that it will be observed, and the exception halts the Flow before the consumption of the

cachedValue

happens or while it is in progress. Is my understanding correct? Is there any way to adjust the Flow to fit the requirements? The code: https://pl.kotl.in/3U8jVprL-

Hello All, I might be missing something but what is the difference between a flow generated from flow { … } and a ReceiveChannel.receiveAsFlow() im seeing some behavour i dont understand, namely if i create multiple new flows based on filters, the regular flow one receives all the messages, this isnt the case when they come from a receive channel. I have a unit test to replicate what im seeing if it helps?

I"m not able to use coroutine Context, in the flowOn(....), app is crashing with

java.lang.IllegalArgumentException: Flow context cannot contain job in it. Had [SupervisorJobImpl{Active}@7b4ca1f, <http://Dispatchers.IO|Dispatchers.IO>]

interface UseCaseScope : CoroutineScope {
    override val coroutineContext: CoroutineContext
        get() = SupervisorJob() + <http://Dispatchers.IO|Dispatchers.IO>
}

The intention here is to cancel the coroutine context when we business logic need to cancel it. So that i can call coroutineContext.cancel()

class LoginUseCase(private val authRepository: AuthRepository) : FlowUseCase<LoginRequest, Boolean>() {

    override fun execute(parameter: LoginRequest): Flow<Result<Boolean>> =
        flow {
            authRepository.login(parameter.username, parameter.password).collect { response ->
                this.defaultResultHandler(onSuccess = {
                    emit(Result.success(true))
                }, response)
            }
        }.flowOn(coroutineContext)

    override fun cancel() {
        coroutineContext.cancel()
    }
}

^^^ will cause app the crash Or is this is the way to execute my intention, and ignore flowOn(.....) , i'm really confused here

class LoginUseCase(private val authRepository: AuthRepository) :
    FlowUseCase<LoginRequest, Boolean>() {

    override fun execute(parameter: LoginRequest): Flow<Result<Boolean>> =
        flow {
>>>>         launch(coroutineContext) {
                authRepository.login(parameter.username, parameter.password).collect { response ->
                    this@flow.defaultResultHandler(onSuccess = {
                        emit(Result.success(true))
                    }, response)
                }
            }
        }

    override fun cancel() {
>>>    coroutineContext.cancel()
    }
}

Flow is the mental state

How do you collect a flow once in a non-blocking way, so it doesn't update when the flow updates?

I have a

stateFlow

that combines two

snapshotFlow

that works when running on device, but when I attempt to unit test with turbine, it never emits more than once. Am I doing something wrong or am I going down the wrong path?

fun validInput() = runTest {
    viewModel.screenState.test {
        assertEquals(awaitItem(), EnterNameScreenState.EnterNameLoadingResults) // passes
        viewModel.firstNameInputValidation("test")
        // fails, waited too long
        assertEquals(awaitItem(), EnterNameScreenState.EnterNameLoadedResults(InputWrapper(value = "test"), InputWrapper(), false)). 
    }
}

Anyone can help with this question related to Flow

retry

mechanism?

Hello 🙂 I'm trying to use MutableStateFlow to log state updates but I can't quite pull it off: • I'm starting a collect job with launch, but I can't figure out how to be sure the collect job will be started before the first update comes (I could use a condition variable but it feels pretty overkill). I'm using yield() to help make it happen, but it feels very hacky • how can I make sure the updates get delivered to the collector once it is subscribed?

Would running all of it unconfined be appropriate?

Not sure if I should ask in compose or here, but cross share it https://kotlinlang.slack.com/archives/CJLTWPH7S/p1667198842541229

Is there a better way to “launch flow collection but suspend until the collection is ready”? Without this workaround (i.e.

.onEach { … }.launchIn(…)

) elements can be missed if they’re emitted shortly after this.

Hi! I have an issue with

StateFlow

-> when I open the screen for the first time and do actions to emit new values then value is updated, but when I re-open the view then value is not updated to initial value, but old value is still there. This is the way I try to keep the state up to date:

private val valueChanges: MutableSharedFlow<Int> = MutableSharedFlow(
  extraBufferCapacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST
)

private val valueState: StateFlow<Int> = valueChanges
  .stateIn(scope, SharingStarted.Eagerly, 0)

Then at the very beginning of the whole flow I call this function where

changes

is

valueChanges

flow:

updateChangesOnStart(changes: Flow<Int>) = changes
  .onStart { emit(0) }

To update the state later I use:

scope.launch{
  valueChanges.emit(${passed number})
}

Is the scope the one that does not allow me to update the value when function is reused?

So i've built an EventBus around flows and I'm trying to implement a priority of subscriptions would there be a good way to implement this?

Hi! I created a new video about “What is a Flow in Kotlin?”

Hi, how would I emit an error into my MutableStateFlow for testing?

When dealing with a StateFlow in ViewModel, stateIn vs collect: the benefit of using stateIn is the “started” param which lets you resubscribe to keep the stream alive on configuration changes. Correct? Any other considerations?