Hey - I'm having some struggles thinking through flowOn() and the final collect operation. flowOn(SomeDispatcher) seems to configure the up-stream pipeline up to the point where teh pipeline element has a coroutine context already. Cool. But how do I configure

collect{}

to run in a particular context? Should I move what was done in a collect() to an onEach{} and then let collect{} just trigger final logging?

Hello, I am trying to concat two flows in a way that I get results sequentially when I collect on it

flowOf(
 remoteDataSource.getDataFromCache() // suspending function returning Flow<Data>
  .catch { error -> Timber.e(error) },
 remoteDataSource.getDataFromServer() // suspending function returning Flow<Data>
).flattenConcat().collect {
Timber.i("Response Received")
}

my expectation here was that I will get first result first and then second after sometime as response comes from server. But the problem here is collect only gets called after server returns result.

Has anyone cooked up a solution yet for

publish(selector: (Flow<T> -> Flow<R>)

? I’m looking at this issue and I have the exact same use-case https://github.com/Kotlin/kotlinx.coroutines/issues/1086#issuecomment-585174549

Hey all, I’m having some trouble understanding the downside of using buffer() on a flow before collecting it? Is there some reason not to use it? At first I thought it could change the order of values treatment but it doesn’t seem like it

ReceiveChannel#consumeAsFlow() never completes? I wrote a short program to test this

fun main() = runBlocking<Unit> {
    val test = Test()

    launch(<http://Dispatchers.IO|Dispatchers.IO>) {
        println("Receive 1")
        test.getResponseChannel()
            .collect { println("Collect 1 $it") }
    }

    launch(<http://Dispatchers.IO|Dispatchers.IO>) {
        println("Receive 2")
        test.getResponseChannel()
            .collect {
                println("Collect 2 $it")
            }
    }

    launch(<http://Dispatchers.IO|Dispatchers.IO>) {
        println("Receive 3")
        test.getResponseChannel()
            .collect {
                println("Collect 3 $it")
            }
    }

    println("Sending A")
    test.responseChannel.send("A")
    println("Sending B")
    test.responseChannel.send("B")
    println("Sending C")
    test.responseChannel.send("C")
}

class Test {
    val responseChannel = BroadcastChannel<String>(BUFFERED)

    fun getResponseChannel(): Flow<String> {
        val channel = responseChannel.openSubscription()
        return channel
            .consumeAsFlow()
            .onCompletion {
                println("Cancelled")
                channel.cancel()
            }
    }
}

Here you will see that

onCompletion()

and hence “Cancelled” never prints. My requirement is to be able to add and detach observers to a broadcast channel. It that achievable?

Hey guys, I want to send stream of data from one class to the other (adapter/viewholder to activity). Previously I'd use RxJava's Subject for it, what would be recommended way of achieving this with flow? I see how I could do it with channels but it's hard for me to wrap my head around the flow for such use cases

I'm starting with flows (and coroutines). I'm trying to create a consumer of a flow that asynchronously executes a function. Would this approach work? https://pastebin.com/k00hWppm I still have some questions like: 1. Should the map of flows and the executor be in the same scope? 2. Would this scale? For example if i publish hundreds of thousands of "events" in the flows, would this become a problem as I'm launching many coroutines? As context this is for a load testing library that I'm playing around with

@Jacob Applin You can add

"-Xopt-in=kotlinx.coroutines.ExperimentalCoroutinesApi"

to compiler args. Like this:

compileKotlin {
    kotlinOptions {
        jvmTarget = "1.8"
        apiVersion = "1.3"
        languageVersion = "1.3"
        freeCompilerArgs = ["-progressive", "-Xopt-in=kotlinx.coroutines.ExperimentalCoroutinesApi"]
    }
}

Is intermediate flow operators are cancellable? Are they cancelled when the calling scope is cancelled? For instance:

// If the collecting scope is cancelled before this map operation,
// seems like map operation is not cancelled and throws an exception
numbersFlow.map {
    // There is an operation which throws exception
}
.flowOn(<http://Dispatchers.IO|Dispatchers.IO>)

I have to use

ensureActive()

before mapping but is it the expected behaviour?

Is there some nifty way to get the last emitted value of a flow, without having to cache it my self?

Hello, I currently have a

Channel

representing a job queue, and a method that `send`s a new job to the queue. A separate coroutine is an infinite

receive

loop that processes the jobs. How would I represent that with Flow? I'm not sure how to use the

flow {}

builder since I don't know what to

emit

upfront, the jobs come from outside.

Does anyone have an elegant way to gracefully stop processing a flow, other than cancelling at source? Some sources are not easily cancellable, like (e.g.) StateFlow. I may also have several flows collecting from a StateFlow, but only want to cancel one of them. [Conflated]BroadcastChannel will let me cancel my subscription to it, but there seems no obvious equivalent for flows?

I am new to kotlin flows and trying to add Flows to one of my project and facing an issue while converting list of objects of one type to another as shown below. Basically, am not sure which operator to be used for this.

//Repository - method return type is  Flow<List<HistoryEvent>>
fun getHistoryEvents(): Flow<List<HistoryEvent>>

//ViewModel - calling repository method and trying to convert data to List<AccessoryEvent>

val accessoryEvents : LiveData<List<AccessoryEvent>>=  liveData{
                                                      repository.getHistoryEvents()

                                                  // what operator to be used to convert from List<HistoryEvent> to List<AccessoryEvent>

                                                        .collect()
                                                      }

I tried like below

val eventHistory: LiveData<List<AccessoryEventType>> = 
        liveData {
            deviceRepository.getDeviceEventHistoryFor("abcd")
                .map { historyEvents ->
                    emit((historyEvents.map { AccessoryEventType.getEventType(it) }))
                }
                .collect()
        }

after this, all elements in the returned list are same and equals to the last emitted element. Am not sure what I am doing wrong

I'm wondering how to structure the code inside the class where I need something like an interval from RxJava, that'll do heavy operation every 50ms. The problem is that I'd want the channel producer to execute heavy operation only when someone is collecting the flow. I don't want to waste resources. Is there some easy way to achieve it?

Hi, I have a flow of integers. I'd like to keep track of the highest value received and perform an action if a received value is greater than the highest. I'm struggling to figure out which operation I can use to do that https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/ - any ideas? Equivalent without flows might be something like:

var highest = 0
for (i in myIntegers) {
   if (i > highest) {
      highest = i
      doSomethingElse()
   }
}

Hi! Is there something similar to MutableStateFlow, but without “state”? I just want simple entity that allows to pass items from one place and listen for changes at another. So, I don’t need getting last sent item after calling collect(), only new items

@Test
    fun testFlow1() {
        (1..5)
            .asFlow()
            .onEach { println(it) } // prints: 1, 2, 3, 4, 5
    }

    @Test
    fun testFlow2() {
        data class Abc(val id: Int)

        val list: Flow<List<Abc>> = flow { emit(listOf(Abc(0), Abc(1), Abc(2))) }
        list
            .onEach { println(it) } // prints: list of object instead of each single abc object
    }

Does currently possible to iterate each index for

testFlow2

without using it like this:

list
   .onEach { 
      it.forEach { abc -> println(abc.id) }
   }

My goals:

list
   .onEach { abc -> 
       println(abc.id) // prints: [0, 1, 2]
   }

What are the functional differences between a

SharedFlow

and a

StateFlow.buffer(_UNLIMITED_)

? Buffering a StateFlow seems to mimic SharedFlow for a simple test I created. Is there a fundamental difference between the way those two flows behave?

Hey people, good day =D I’m having an issue of founding the best usage of Kotlin Flow in a normal API request. Basically I’m doing a small sample Android app with Flow usage for my team. I will retrieve the data from a API endpoint and want to show it in a List. I’m using MVP arch and no LiveData.

Does any one can guide me to an article or sample app where this works?

Not sure what is the best single responsibility for this case. • Do I retrieve all information from the Endpoint and emit after each map? • Can I use Flow for pagination ? • Maybe this is not a good case for Flow? Thanks ;)

Yes. Probably that’s the problem. I don’t see any fast solution for solving this with MutableStateFlow. Probably simply MutableStateFlow is not the correct class to use in this case

Is there an equivalent of

transformLatest

but for

combineTransform

? I'm looking specifically for something that cancels the work if a new item comes in. For now I can just zip up the items in

combine

to a pair and use

transformLatest

I'm using

StateFlow

to emit some states and do some work based on that. My code looks like this

state.onEach(this::handleState)
     .launchIn(scope)

Now since the

onEach

block is a suspending function, if another state is emitted while it is executing, it doesn't cancel the current suspending function. Instead it waits for the

handleState

function to finish it's execution and then next state's handling is being done. I realise this is actually done intentionally to handle back pressure and all, but what I want is to cancel

onEach

execution when new state is received and run

onEach

for the new state. Currently I'm running the

handleState

block in another coroutine using

scope.launch { }

, store that job as member variable, then cancel that in

onEach

before launching new coroutine. Is there any other way to achieve this?

Hey! I’m new with flow and I’ve been pondering how to implement the following data flow. My use case is that I’m using TMDB API with two endpoints: one to fetch a list of genres and second to fetch a list of movies per genre. This means that if the API returns N genres I have to make N requests to fetch the list of movies. My repository returns a flow of these objects, for example Flow<List<Genre> and Flow<List<Movie>>. My question is, what would be a flow way of “combining” these two streams, so that once the genres stream emits a value (it will only emit one value, all the genres), the genres would be looped over and movies would be queried for each, then those two combined. If I were to do this without flow I’d do something like:

fun fetchData() {
   val genresWithMovies: List<GenreWithMovies> = dataSource.queryGenres().map { genre ->
        val movies = dataSource2.queryMovies(genre.id)
        genre.movies = movies
        genre
   }
}

Any tips?

If you have a buffered SharedFlow with a slow and fast collector, and the fast collector finishes collecting the most recent value, does the fast collector suspend until the slow collector finishes collecting that same most recent value? Or does the fast collector pull the next value from the buffer and get ahead of the slow collector? If the latter, is the total buffer space (replay + extraBuffer) equal to the amount of values a fast collector can get ahead of a slow collector?

The shared model I am working on (in

commonMain

) looks like this:

class CoreViewModel {
    internal val mutableStateFlow = MutableStateFlow(AppState())
    val stateFlow: StateFlow<AppState>
        get() = mutableStateFlow
}

Can anyone show how it’s possible to collect the

StateFlow

value in Swift, on the iosApp side? I am looking for a way to connect the StateFlow value to an ObservableObject.

Hello everyone, I am a Kotlin newbie and are looking for a way to resume a Flow after an exception has been thrown by the emitter. I want to adhere to Exception Transparency. I have seen a workaround on SO (https://stackoverflow.com/a/63886110) but wanted to ask if you know an easier / less custom solution. Thanks!

Is there an example how to implement rate limiting with Flow? I have an API client that is supposed to make hundreds of requests. X requests should run in parallel and as soon as one of the request receives a rate limiting error from the server, all request processing should pause until the waiting period is over.

Is it possible to create a flow operator similar to

collect

with its own coroutine scope for each item it receives? E.g lets say I have an updating list of bicycles, a

Flow<List<Bicycle>>

. For each bicycle I can request a price, unfortunately this API requires me to request the prices for a single bicycle at a time. This is all fine i can spin up some requests using a couple of

async

builders, no problem so far. The API for requesting prices might be slow at times, slower than the stream of new bicycles that becomes available or taken. This however creates the issue of still be working on fetching prices for bikes that are no longer available or taken. Is there any way to create a coroutine scope for each item collected from my

Flow<List<Bicycle>>

and then cancel it when a new item is collected, If I understand correctly due to the sequential nature of flows, the next item is not requested until the previous is finished/completed?

Hey is there a non-state version of

MutableStateFlow

? I have been using

ConflatedBroadcastChannel

, but would like to move to flow completely. I know there is

Flow

itself, but I have not found a way to send values to it from outside the creation lambda.