Hi everyone, I have a doubt regarding the flatMapLatest and the UDF. I have pasted a simplified code version with more details. I would like to create a UIState in my viewmodel, with several properties (example like HomeUiState in JetNews). One of these properties is a list of users that might be filtered by the user. 1 - The first approach would be exposing the list, which would be filtered every time the userId changes. In this case I would not be able to add the list into my UIState. 2 - The second one would be collecting the flow into the init and update the UIState in the onEach. In this case though, I would need to trigger the userId from the UI in order to start it the first time. Any suggestions for situations like this? Am I doing anything wrong? Thanks!

data class HomeUiState(
    val users: MutableList<User>
)

class HomeViewModel(
) : ViewModel(){

    private val repository: Repository = Repository()

    private val viewModelState = MutableStateFlow(HomeUiState(mutableListOf()))
    val uiState: StateFlow<HomeUiState> = viewModelState
        .stateIn(
            viewModelScope,
            SharingStarted.Eagerly,
            viewModelState.value
        )

    val userId = MutableStateFlow(1)

    //  1 solutions: Keep the list out the UIState and collect it in compose with collectAsState 
    val userList = userId.flatMapLatest {
        repository.getUsersList()
    }.stateIn(
        viewModelScope,
        SharingStarted.WhileSubscribed(5000),
        emptySet()
    )

    // 2 solution, flatMapLatest for the filtered list
    init {
        userId.flatMapLatest {
            repository.getUsersList(it)
        }.onEach { users ->
            viewModelState.update {
                it.copy(users = users) }
        }.launchIn(viewModelScope)
    }
}

Any guide on how to migrate

@ObsoleteCoroutinesApi
public fun <E> CoroutineScope.broadcast(
    context: CoroutineContext = EmptyCoroutineContext,
    capacity: Int = 1,
    start: CoroutineStart = CoroutineStart.LAZY,
    onCompletion: CompletionHandler? = null,
    @BuilderInference block: suspend ProducerScope<E>.() -> Unit
): BroadcastChannel<E>

to

SharedFlow

?

Is there a function that splits a

Flow<UnmappedValue>

into a

Map<SomeKey, Flow<MappedValue>>

? Like the opposite of

merge

? Maybe

explode

or

subFlows

or something like that?

Still not super comfortable with flows, but Im getting a laggy UI in my android app with the following, and was wondering if anyone had any pointers. Basically I just have a search field and as a user types, I want to search against my api, but I want to make sure to wait 250ms before hitting a network call.

var queryText = MutableStateFlow("")

init {
  viewModelScope.launch {
    myState.queryText.debounce(250).collect {
        search(it)
    }
  }
}

private fun search(term: String) {
  viewModelScope.launch {
    val result = service.searchApiCall(term)
...

Now that I think about it. I probably want my old network calls to be cancelled too. Can anyone point me in the right direction about how I should rethink this?

Does it Flow extensions

single

is equivalent to

first

? There’s seems a weird issue in my part. When using

single

it returns

null

, but exist otherwise with

first

Hi, I'm pretty new to using Flows in Kotlin so I'm having an issue I hope one of you can provide me some help to. I'm trying to chain some requests and if they fail I want them to stop the chaining and just emit an error. Preferably it would be one size fits all so I don't have to check for the Error for each response. Right now I just emit emptyFlow as the IDE will complain if I return nothing Any idea what I can do? (also any other inputs are more than welcome)

signupUseCase.signUp(createPostRegister(email, password))
     .flatMapConcat {
         when (it) {
             is NetworkResult.Success -> loginUseCase.login(
                 createPostToken(
                     email,
                     password
                 )
             )
             is NetworkResult.Error -> {
                 it.rbError?.let { error -> handleErrors(error) }
                 emptyFlow()
             }
             else -> emptyFlow()
         }
     }
     .flatMapConcat {
         when (it) {
             is NetworkResult.Success -> {
                 prefManager.saveString(AppConstants.TOKEN_STORAGE, it.data)
                 prefManager.saveString(AppConstants.EMAIL_STORAGE, email)
                 prefManager.savePasswordEncrypted(password)
                 updateCountryUseCase.updateCountry()
                 return@flatMapConcat patchSelfUseCase.patchSelf(name, ApiConstants.fullName)
             }
             else -> emptyFlow()
         }
     }.flatMapConcat {
         when (it) {
             is NetworkResult.Success -> getSelfUseCase.getSelf()
             is NetworkResult.Error -> {
                 it.rbError?.let { error -> handleErrors(error) }
                 emptyFlow()
             }
             else -> emptyFlow()
         }
     }.flatMapConcat {
         when (it) {
             is NetworkResult.Success -> {
                 it.data?.let { self -> instance.setSelf(self) }
                 legalRevisions.value?.let { revisions ->
                     return@flatMapConcat putLegalAcceptUseCase.putLegalAccept(revisions)
                 } ?: run {
                     return@flatMapConcat emptyFlow()
                 }
             }
             is NetworkResult.Error -> {
                 it.rbError?.let { error -> handleErrors(error) }
                 emptyFlow()
             }
             else -> emptyFlow()
         }
     }.collect { signupCompleted.postValue(true) }

can anyone provide insight into what the best way to combine two flows is? I’m currentlly using the actual

combine

operator, but it’s not getting me quite the result I’m after. Here’s an example of what I’ve got now

flowOne()
.combine(flowTwo, { a, b -> Pair(a,b) })
.onEach { ... }

While the actual values of the

Pair

i’m creating are correct, as far as I can tell, the values are only updated when

flowOne

emits a value, only then combining with the latest value from

flowTwo

. Is there an elegant operator (or approach) I can use to propagate my

Pair

immediately when either

flowOne

or

flowTwo

update, not just

flowOne

? update: turns out I

combine

is what I want! except, rather than

flowOne.combine(flowTwo)

, I need

combine(flowOne, flowTwo)

:)

Is there a way to explicitly create a child coroutine scope? My use case is that I want to create a bunch of `SharedFlow`s and occasionally close them all. The way I’m trying to do that is by creating them with

shareIn(scope.newChildScope())

and closing with

scope.coroutineContext.cancelChildren()

Hi, I'm new to use Flow so I confused when to use regular

Flow

or

StateFlow

. Many codes expose

Flow

in repository, and convert to

StateFlow

in ViewModels like this.

val userState: StateFlow<User?> = userRepository.userFlow.stateIn(viewModelScope, Eagerly, null)

This case always show null state on UI started, even if repository has initial value. This problem may be solved by exposing

StateFlow

instead of regular

Flow

in repository, is it bad practice?

Hi, I want callbackflow to be shared between subscribers. I see solutions using shareIn for example - https://barbeau.medium.com/kotlin-callbackflow-a-lightweight-architecture-for-location-aware-android-apps-f97b5e66aaa2. But I would like to avoid an external coroutinescope for shareIn. Is there any solutions which relies on a combination of scopes of the subscribers?

According to the docs, when the upstream of a SharedFlow completes, the downstream just remains stuck collecting forever. What do you guys think of this utility to work around this? Code in thread.

Greetings, seems like I misunderstand

shareIn

functionality. Trying to share the flow emitions betwenn 3 subscribers, but only one of them (first) is collecting data:

@Test
  fun testSharedFlowEmitions() = runBlocking {
    var tick = 0
    val f: Flow<Int> = flow {
      while (tick < 10) {
        emit(tick)
        tick++
      }
    }

    val sharedFlow: SharedFlow<Int> = f.shareIn(this, SharingStarted.Eagerly, replay = 1)

    val firstSubscriber = sharedFlow.collect { Truth.assertThat(it).isGreaterThan(0) }
    val secondSubscriber = sharedFlow.collect { Truth.assertThat(it).isGreaterThan(0) }
    val thirdSubscriber = sharedFlow.collect { Truth.assertThat(it).isGreaterThan(0) }
  }

Need help so that all three get ticks.

Is there an easy way convert a

SharedFlow

only to emit to its first subscriber?

Is it possible to transform from SharedFlow<typeA> to SharedFlow<typeB> without introducing a new scope? A standard transform() converts it to a Flow<typeB>

Equivalent of RxJava's

Observable.combineLatest(...)

with n Flows to combine? There has to be a better way than this 👉 🧵

As part of practicing coroutines and flow, I tried an implementation of

amb/race

where I don't allow myself to use channels directly. Or, to put it differently, I use

SharedFlow

instead of channels. It runs fine except that final flow collection suspends indefinitely, even after the jobs used to share the losing flows have completed and the winning flow has emitted all its items. Why? I spent a couple days trying to figure out what might be going on, without success. Here's a gist of my work, including the code I'm using to test it, and the output generated. You'll see from the output that the flow correctly emits the exception thrown by the first flow, and cancels the third flow because it emits last (so it loses the race), and all the items in the winning flow are emitted. This is a learning exercise, so I'm interested in the shortest path to getting this implementation to work. I'm not looking for a completely different implementation that works. Also happy to create a project on Github with dependencies (I use Arrow's Either data-type) if anyone's interested in running the code directly. Let me know. Thank you!

Is there an equivalent for

Observable.compose()

in Kotlin Flow, so we don’t break the chain, the idea is to offer only callbacks to consumer code, and internally I’m using

Stateflow

and

SharedFlow

. in Rxjava I was using

Observable.compose()

with Rx transformers.

Hi all, is there any flow operator similar pairwise: https://rxjs.dev/api/operators/pairwise ?

Is there a flow operator which allows me to to take the last two events and map them? or do I need to use a SharedFlow/StateFlow for that?

Why do the docs discourage the use of

flatMapMerge{}

? Android SQLite has a limit of 999 params per query, so I

window

my params by 999, call

asFlow

, then

flatMapMerge{ids->db.flowOfEntities(ids)}

. Is that wrong?

Hello! Is there a way to wait for an update on a list of flows to generate another flow? for example, these could be stateflows of integers, one per resource, and I'd like to generate a flow which is the resource associated with the current maximum integer

Any idea why the require in awaitClose would fail? According to docs the channel is closed before the block passed into awaitClose is invoked

fun flowFrom(api: CallbackBasedApi): Flow<T> = callbackFlow {
    val callback = object : Callback { // Implementation of some callback interface
        override fun onNextValue(value: T) {
            // To avoid blocking you can configure channel capacity using
            // either buffer(Channel.CONFLATED) or buffer(Channel.UNLIMITED) to avoid overfill
            trySendBlocking(value)
                .onFailure { throwable ->
                    // Downstream has been cancelled or failed, can log here
                }
        }
        override fun onApiError(cause: Throwable) {
            cancel(CancellationException("API Error", cause))
        }
        override fun onCompleted() = channel.close()
    }
    api.register(callback)
    /*
     * Suspends until either 'onCompleted'/'onApiError' from the callback is invoked
     * or flow collector is cancelled (e.g. by 'take(1)' or because a collector's coroutine was cancelled).
     * In both cases, callback will be properly unregistered.
     */
        awaitClose { 
            require(isClosedForSend) // <--- THIS FAILS
            api.unregister(callback) 
        }
    }

What is the most efficient way to add a List<X> to a Flow of

List<Y>.flatMapMerge { channelFlow { send(it.toX()) } }.buffer()

while preserving concurrency?

Why is only

flow2

being called in

flow1.map{ flow2(it) }.flattenConcat()

?

flow1

is a

MutableStateFlow<MutableSet<Int>>

of db entity IDs,

flow2

is a Flow of lists of entities, which is updated when any of the flow's entities' properties change. I am updating the properties and then calling

flow1.emit(set)

, but only

flow2.onEach

is being called, not

flow1.map{}

. Any suggestions?

How to convert pass Observable emissions to

MutableSharedFlow<T>

?

I have an interesting use case I’m looking to support. I’m currently using a MutableStateFlow of a single data class to represent the UI state. I have several video controllers that control several videos. Each one exposes the current timestamp of the video as a Flow<Double>. The user can select one video, and the seekbar should set its position based on the timestamp of the selected video. My current solution is to collect the state flow, get the current controller, and collect the position inside of that. I can then update the state based on the collected value. Is there a better way to handle this?

Why does

combine(flows, transform)

not call

transform

if

flows

is empty? Is that intended? Seems like it would be more natural to call it with an empty array.

Whoever decided to have both

Flow{ fun collect() }

and

Flow.collect

(an extension function) needs to have a stern talking to.

Can I save the last n values of

sharedFlow

and later return the value that I want, Do

Reply

help me in this task?