Does Flow have something like RxJava's

compose

operator? (Rather than operating individual items emitted by a Flow, it would operate on the Flow itself, returning a new composed Flow.)

We are integrating flow in out Kotlin Native project to implement viewmodels in the common module so those viewmodels can be shared between iOS and Android. At the moment we have started using flow for this which seems to fit nicely in the overall architecture. A viewmodel can expose a flow to which the UI can collect to. The UI can also indicate it wants new data (refresh). I have implemented this in the following way, by using a MutableStateFlow indicating the refresh status. I have a simple example implementation ((with documentation)) below which seems to work quite well. Is this the correct approach to this problem with Kotlin flows?

@ExperimentalCoroutinesApi
class PersonsViewModel {
    private val realApi = RealApi()

    // cached list of persons
    private var cachedPersons: List<Person>? = null

    // indicates of the data should be retrieved from network or cache. This is a MutableStateFlow since its value
    // can be updated by the UI (for example uses refreshes the page).
    private val refreshTrigger: MutableStateFlow<RefreshTrigger> = MutableStateFlow(RefreshTrigger.FromCache())

    // flatMapLatest on refreshTrigger because we are interested in the latest value only.
    var persons: Flow<CommonDataContainer<List<Person>>> = refreshTrigger.flatMapLatest { trigger ->
        flow<CommonDataContainer<List<Person>>> {
            // Notify the UI we are loading data
            emit(CommonDataContainer.Loading())
            val refresh = trigger is RefreshTrigger.Refresh
            val cached = cachedPersons
            if (!refresh && cached != null) {
//                Return the cached data if not explicitly refreshed and data is in cache.
                emit(CommonDataContainer.Success(cached))
            } else {
                // Refresh or data is not in cache, retrieve from network.
                retrievePersonsFromNetwork()
            }
        }
    }

    private suspend fun FlowCollector<CommonDataContainer<List<Person>>>.retrievePersonsFromNetwork() {
        val response = realApi.retrievePersons()
        when (response) {
            is Success -> {
                cachedPersons = response.data
                // Emit the data.
                emit(CommonDataContainer.Success(response.data))
            }
            // Something went wrong, emit a failure.
            is Failure -> emit(CommonDataContainer.Failure())
        }
    }

    suspend fun refresh() {
        yield()
        refreshTrigger.value = RefreshTrigger.Refresh()
    }
}

// CommonDataContainer can represent various states which are interesting to collectors.
sealed class CommonDataContainer<out T> {
    // Indicates the data is loading
    class Loading<T> : CommonDataContainer<T>()
    // Indicates an error occurred retrieving the data
    class Failure<T> : CommonDataContainer<T>()
    // Indicates success. Holds the data in the data property.
    class Success<out T>(val data: T) : CommonDataContainer<T>()
}

// Indicates where the data should com from. Both classes  have a unique id to make sure the Refresh trigger is always emitted by MutableStateFlow
// since MutableStateFLow does not emit the value if the new value is the same as the current value.
sealed class RefreshTrigger {
    data class FromCache(private val id: String = Random.nextInt().toString()) : RefreshTrigger()

    /**
     * Every instance has a unique id so the refresh StateFlow sees it as a new value. StateFlow does not emit the same value twice if
     * the current value is the same as the new value.
     */
    data class Refresh(private val id: String = Random.nextInt().toString()) : RefreshTrigger()
}

Got a question: - I have three

suspend

operations, each can be called independently:

foo1(list)

,

foo2(list)

and

foo3(list)

- each function changes state of items in list -

foo1

just makes request to the server and returns updated list (updates 1 step) -

foo2

calls

foo1

and then with combined result makes a request (updates 2 steps) -

foo3

calls

foo2

, combines result and makes a request (updates 3 steps) Problem is,

foo2

needs to gather data (SMS-code) via DialogFragment. Is there a way to build a continuous pipeline, so the

foo3

can continue its flow after

foo2

called a dialog, gathered input and returned?

what's a good way to do things concurrently in

init {}

?

Quick question, is there any channel to share new articles we published recently? Don't want to create any spam here

I’m having issues in my unit test when I try to use AssertFlow extension on a StateFlow instance. If I use

expectItem()

I get a timeout exception, but if I check for no more items, it tells me something was received. Any hints to fix this?

Hi guys, would you say it’s necessary or unnecessary to change the dispatcher upon flatMapLatest’s transform?

fun getBooksFlow: Flow<List<Book>> = 
     getUserFlow().flatMapLatest { user -> getUserBooksFlow(user.userId) }.flowOn(Dispatchers.Default)

Hello, I have two `Channel`s, and I want to

.receive()

on both of them, so if there is anything in channel 1 or anything in channel 2 it resumes, but of course

.receive()

doesn't actually work since I can't call it on both. What should I be using here? The point here is that channel 1 is higher priority, so I want to get the next item, from channel 1 if there are any, and from channel 2 if channel 1 is empty.

did I miss something? are there any gotchas?

I’m using

StateFlow

to sort of cache and share the result of a coroutine. The

StateFlow

is held in a singleton, and accessed (collected) from multiple different places. When the scope that launches one of these collectors is cancelled (via

scope.cancel()

, it seems that the

StateFlow

is cancelled and no longer emits to any collectors. Can I cancel the collectors launched in a scope, without cancelling the parent producer?

What's the purpose of the

Flow.cancellable

operator ? What kind of use case does it solve ?

Hey, is there any way to pass e.g coroutine scope from suspended function to normal function and then invoke another suspend function on this scope? For example:

suspend fun fooA(){
  coroutineScope {
    fooB(this)
  }
}

fun fooB(coroutineScope: CoroutineScope){
  continueWithSuspend(coroutineScope){
    continueSuspendingFooA()
  }
}

private suspend fun continueSuspendingFooA(){}

I have noticed the zip operator executes the coroutines sequentially, see the following example:

flowOf(getWeather(loc)).zip(flowOf(reverseGeocode(loc))) { weather, address ->
            weather to address
        }
            .flowOn(Dispatchers.Default)
            .collect { println(it) }

The time it takes to execute this is the total time of the getWeather and reverseGeocode calls combined. I know I can use a map operator and two async/wait blocks to execute the two calls asynchronously. Is it possible to execute multiple coroutines like this with flow operators? I tried a couple of different things but without success? I saw https://github.com/Kotlin/kotlinx.coroutines/issues/1147 so perhaps all operators are sequential for now.

I need to unit-test the code involving a

scan

operator and hot event stream. Basically I want to configure the flow, send events and verify they are received:

runBlocking(Dispatchers.Default) {
    val events = BroadcastChannel<Int>(capacity = Channel.BUFFERED)
    launch {
      events.asFlow().scan(1) { accumulator, value -> value }
        .take(2)
        .collect { println("received $it") }
    }
    launch {
      events.send(42)
    }
  }

The above doesn't work, it receives 1(inital accum) item and then hangs indefinitely, because

send

is done before collect begins. I tried moving

send

inside the

collect

, after initial element is received, but this doesn't work, because

scan

is implemented so that its internal

collect

also starts after my

send

is called (it starts after first collection completes). Any advice on how to do that?

I'm building a massively concurrent system with coroutines, have thousands of coroutines running at a time, and have 100% CPU usage most of the time. It all works great, but how can I give some coroutines higher priority? Sometimes I have work incoming that should be completed first, but because the application is already busy with other stuff the priority work has to wait for other work to finish before being processed. Alternatively, how can I start some coroutines on a low-priority thread, which should let me achieve the same goal?

When querying an Android

ContentResolver

you can optionally pass a

CancellationSignal

argument. So I would like to call

cancel()

on that object when the coroutine is cancelled. Is there a way to do this without using

suspendCancellableCoroutine

(since I don’t need the whole

resume()

functionality)?

suspendCancellableCoroutine<Result> { continuation ->
    val cancellationSignal = CancellationSignal()
	continuation.invokeOnCancellation { 
    	cancellationSignal.cancel()
	}
	val result: Result = resolver.query(args, cancellationSignal)
	continuation.resume(result)
}

I've got some code that is basically a hierarchy of sub-tasks that I think would be a good fit to migrate to coroutines, but I've got some dumb questions trying to get a few things straight. 1. Is the proper way to propagate a failure up the chain (from, say any child task in that hierarchy) to define a set of custom exceptions to encode different events and use those? If so, is an exception typically used even for a non-error "finished" case? (one of the sub-tasks can finish on its own for a non-error reason, so I could define something like a "FinishedSuccessfully" exception for that? Or is there a better pattern to use there?) 2. If object A has a coroutinescope defined and creates object B--which will launch its own child coroutines--and I want object B's coroutines to be launched within object A's coroutinescope, is the best way to just have object A pass its scope to object B in the ctor (or in function calls)? Or is there a better way to (implicitly?) have B 'inherit' A's coroutine scope?

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?

This might be a trivial questions for most of you. But can anyone please point me to an article/ list of articles which dig deep in inner workings of coroutines. I mean elaborate details of thread vs coroutines... Coroutine contexts ... Etc

I’m not usually one to post an SO question, but i feel like it’s relevant here, and it could use some more attention: https://stackoverflow.com/questions/62747825/coroutine-stateflow-stops-emitting-when-coroutinescope-is-cancelled

when using callbackFlow I need to clean up resources but when the job is canceled awaitClose isn't being called. How do I invoke awaitClose on a cancellation of the job

fun listen(port: Int = 1337, nThreads: Int = 2) =  callbackFlow<String> {

                val ch = b.bind(port).coAwait { }
                for (obj in coReceiveChannel) {

                    send(obj)
                }

                  awaitClose {
                    coReceiveChannel.cancel()
                    ch.close()
                   }
            
      } // end of listen
      
job = listen().onEach { println(it) }.launchIn(scope)
        
job.cancel()   // doesn't call awaitClose

How can you emulate `ContentLoadingProgressBar`’s behavior in a

Flow<T>

? Specifically, if I begin observing a

Flow<ViewState>

and it emits no

ViewState.Loaded

in the first 300ms, I want to emit a

ViewState.Loading

while ensuring that this

ViewState.Loading

emits before any

ViewState.Loaded

from the source

Flow<ViewState>

? Kind of equivalent to:

val sourceFlow: Flow<ViewState> = ..

sourceFlow
  .onStart {
    delay(300)
    // if `sourceFlow` has not emitted any items yet
    emit(ViewState.Loading)
  }

https://developer.android.com/reference/androidx/core/widget/ContentLoadingProgressBar

how do I map from a

(Mutable)StateFlow<T>

to another

(Mutable)StateFlow<T>

? when I call the

map

method I map the

StateFlow<T>

to a

Flow<T>

@elizarov it seems I can't find native-mt version for Kotlin 1.4-M3 in Maven (only M2 is there). Is it going to be released soon?

How can I somehow suspend until a flow starts to be collected to make sure that a collector will receive the first values that will be emitted in the flow ? Example of code to fix (https://pl.kotl.in/-sNdWODnA):

package com.example.script

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() {
    val count = MutableStateFlow(0)
    
    runBlocking {
        // Collect the values in a new coroutine.
        val job = launch {
            // Prints "3 4 5".
            count.take(3).collect {
                println("$it ")
            }
        }
        
        // TODO: Somehow synchronize and suspend here until the coroutine starts collecting
        // so that we print "0 1 2" instead.
        
        repeat(3) {
            count.value++
        }
        
        repeat(3) {
            delay(500)
            count.value++
        }
    }
}

Hi there, I’m facing an issue when declaring an suspended overload of a function

interface ASyntax {}
interface BSyntax : ASyntax

fun hello(f: ASyntax.() -> Unit) = println("A")
suspend fun hello(f: BSyntax.() -> Unit) = println("B")

fun f1() = hello {  } // Might work or cause a compiler error because it resolves to the suspended version
suspend fun f2() = hello {  } // Will call any hello version, but which one is unknown at first place

suspend fun main() {
    f1()
    f2()
}

The problem is that depending on the case, either f1 will cause a compilation error because it resolves to the suspended version or f2 will resolve to the non-suspended version. So I’m wondering if someone knows if this is doable and it might be a bug (that I can report) or this is not really intended to be a thing in the first place (and I should rename one of the hello functions instead)

if i'm using a JNI/native library, and the jvm crashes (

SIGSEGV (0xb)

,

C  [libsystem_pthread.dylib+0x15ad]  pthread_mutex_lock+0x0

) when the value is created outside of a

callbackFlow

but not when it's inside, what would that mean? in other words,

val thing = createNativeThing()
callbackFlow {
  thing.onCallback = {
    offer(it)
  }
  thing.start()
}

crashes but

callbackFlow {
  val thing = createNativeThing()
  thing.onCallback = {
    offer(it)
  }
  thing.start()
}

does not

Hello everyone, I have problem with Coroutine - Flow and retrofit Does anyone know why Coroutine doesn’t work with QueryMap? Thanks.

Not Work:

@GET("/users")
    suspend fun getUsers(
            @QueryMap params:Map<String,Any>): BaseResponse<List<UserDto>>


Worked: 
@GET("/users")
    suspend fun getUsers(
            @Query("lat") lat:Double,
            @Query("long") lon:Double,
            @Query("genders")genders: String,
            @Query("age_max") ageMax:Int,
            @Query("age_min") ageMin:Int
    ): BaseResponse<List<UserDto>>

My Repository implement:

fun search(params:Map<String,Any>) = flow {
        emit(service.getMatchedUsers(params
        ).data.map {
            it.mapToDomainModel() }.toList())
    }

Hello, is there a way to declare a stateFlow without an initial value ?