Hi all, I was trying to find an easy implementation to launch multiple api calls in an async block but instead of waiting for all of them, just wait for the first n that return, cancelling the others. I know I can do something like

val results = endpoints.map {
            async {
                client.call(it)
            }
        }
results.awaitAll()

but of course that will wait for all the calls to come back, not just the first N. I know I could solve this problem by using channels and subscribe n topics waiting for them, but I was wondering if there is a counterpart with async/await as well? Also, as a side note, one thing I’ve noticed is that if the

client

is blocking (imagine a Thread.sleep) the async is called sequentially and I lose completely the parallelism..what am I doing wrong?

Try select

The blocking issue is probably caused by a single threaded dispatcher

are there example impls of recommended patterns for achieving something similar to this intra-process EventBus but with coroutines (or project reactor)? #events https://guava.dev/releases/snapshot-jre/api/docs/com/google/common/eventbus/EventBus.html

Is there a way for handling loss of items If a MutableStateFlow which is emitting from background thread faster than the subscriber can collect in main thread?. Using .value to set the value and copying the value from previous state value.

I have a question regarding if I can get into any “trouble” using async in this context (Using async to do 2 stuff in parallel to not unnecessarily wait for some network request). I am always unsure about how cancellations etc. are handled so I am posting this question here, more in thread🧵

A few years into Kotlin and Coroutines -- starting in backend/server now also in android -- I still struggle with a fundamental design problem - how to 'cross the coroutine divide' ... Its a simple and complex problem : Given: one shouldn't make all functions suspend - therefore you will have a mix of suspend and non-suspend functions calling *each other * Given: <in my current domians> -- the 'majority/main' application is not 'coroutine based' -- its "traditional". Coroutines are added as needed, but the scale of existing code (custom and 3rd party, java/JVM and kotlin) is such that its not practical (yet) to have a coherant coroutine architecture and convention everything follows: Given: one should 'Avoid GlobalScope' -- and given its not trivial to create app coroutine scopes such that they are all self-contained -- real world code is not that pretty. Given: It is not obvious or easy to to tell if a non-suspend function was called by a suspend function, or is 'running in a coroutine' without passing down a CoroutineScope or context to every function (see 2.) Given: 'traditional' non-coroutine aware functions expect functions to only return when complete -- quite often you have to (and should) scope coroutine calls so they are blocking from the perspective of the parent non-coroutine-aware caller. This all leads to the need to have a good convention to follow for: How to create blocking scopes from within non-coroutine-aware functions. (without GlobalScope?) How to know if a non-suspend function is in fact in the call-chain of a suspend function or coroutine, so for example, you could call a suspend function without having to launch a new coroutine -- in what scope ? ( presuming the the caller is non-coroutine-aware it will not generally pass in a coroutine scope for you to use --) A few examples: Android: in onCreate() {} I need to call a suspend function (maybe it pre-existed as a suspend even if I dont need suspension for this call, 'suspend' is in it signature) -- But -- I should not return from onCreate until all its work is done. How ? <what scope>?.<what builder> { call suspend function } return // Do NOT return until above is complete If I have a scope (like lifecycleScope) life is easier -- Much, but if not, Then I have to make or borrow one, and then arrange somehow so its scope/lifecycle is well behaved. Could use GlobalScope. shouldnt probably but maybe 'this time' ? What builder? both launch and async -- are async so I have to wait for them. I cant do that outside the scope (the return statement) -- so where? Punt to yet another scope (as seen in sample code) val job = <scope>.launch { .. } <????> { job.join() } // This solves nothing -- Now I need to know how to wait for ??? Is 'runBlocking' the answer ? Its 'to be avoided' -- but precisely under which conditions? ------ Server side: A web request is processed and calls some "doGET" method -- from legacy java code expecting blocking behaviour. fun doGET() { // same question .. what scope ? launch/async ? runBlocking? } ----- The BIGGER problem In comes 'typical mature application' of some thousands of functions written over a decade by different people long gone .. Kotlin/Coroutines was introduced by some a while ago, haphazardly, Now we have a mess because its not at all clear which non-suspend functions are called by suspend functions upstream top() { module() library(callback) callback() --->| back to module onCallback() -> library() subsystem() -> Fancy Coroutine Stuff over blocking IO --> callback() // Developer comment about #@#%% old code and so on -- at any given point maybe a suspend fun maybe not, with no easy/quick way of telling if it was or could be called* by a coroutine or suspend func, or downstream, is there some 5 level deep suspend func calling back into the module without its context to hang on to ? Can one block the thread ? what other choice is there ? I end up having to solve these one by one ad-hoc with no real global architecture or understanding of what could go wrong. e.g. how bad is this in a library function: fun someFunHasToCallASuspendFun() { GlobalScope.runBlocking{ aQuickSuspendFunc() } } To avoid that Ive had to make massive invasive changes to pass in the appropriate scope from 5 levels up and then that can have a huge blast radius (let alone the ire of my teammates for such a huge change for a 'simple fix') Alternatively .. I've added a few 'suspend' modifiers to the fun and its parents etc. But the effect is very much like C++ "const rot" Once you add one suspend, then its parent has to be suspend, then everything that calls it has to also, then their parents -- it can and does explode quickly until often ALL methods have to be suspend just to call 1 10 levels deep. Another idiom: try to establish some well named 'global scopes' that are not 'GlobalScope' -- at some point this just seems silly -- inventing all these scopes just to avoid 1 -- when they do nothing different. It also requires opening up all libraries and modules to more dependencies, to the point sometimes of having to combine libraries just to share a common scope. The closest 'to ideal' I have found is Android Only, and UI Thread Only, <scope>.launch( Dispatchers.Main.immediate ){} If Im really careful I don't have to wait for this -- it inherently blocks the caller -- with some care even child coroutines are managed. But it seems so wrong that this is subtle and complicated -- surely this is the key difficulty in integrating coroutines -- yet most of the docs are written in a way that presumes everything is already coroutine aware - leaving the 'corner cases' as minor issues someone else can work out. Any commentary welcome - am I missing something big here ? Surely theres a good strategy to manage this problem

🌊 How do I test a

viewModelScope

controlled

SharedFlow

? 🤔 😒tackoverflow: More details: https://stackoverflow.com/questions/69425075/how-do-i-test-sharedflow-controlled-by-a-viewmodelscope

Hey, I am testing the difference between the performance of different dispatchers. I got the same results before, but now I confirm it with a benchmark - CPU intensive operation is faster with more threads then Dispatchers.Main, but (on IO the fastest, on executor with 100 threads it is faster than on Default). My only explanation is that it is because those threads are fighting with other threads on my computer, like IntelliJ, browser etc. Is it so, or is there any other explanation?

This message was deleted.

Hi,

Activity {
	private val scope = MainScope()

	onCreate {
		scope.launch {
			userDao.user
				.collect {
					if (it == null) {
						resetBackstack()
					}
				}
		}
	}
}

SomeScreenViewModel {
	private val scope = MainScope()

	fun init() {
		scope.launch {
			userDao.user
				.collect {
					...
				}
		}
	}
}

UserDao.user

emits on its io dispatcher, both activity and some screen viewmodel are driven by main threaded scope (android) Occasionally in crashlytics I see crashes related, basically to the fact that the screen collects the emit first Is there a way I could enforce order of collections? (I presume if it were a replayed hot Flow, then order would be the order of subscriptions, however I presume that posting to main thread will always be race-y?)

Hello! I have a few depencies in my Android project which transitively depends on

kotlin-coroutines-core

different versions. I’ve used

resolutionStrategy

Gradle feature to force using the same 1.4.3 version of coroutines but during release build I get the R8 (ProGuard) error like:

ERROR:/Users/*****/***********/app/build/intermediates/module_and_runtime_deps_classes/preliveGooglePlayRelease/base.jar: R8: Type kotlinx.coroutines.debug.AgentPremain is defined multiple times: /Users/*****/***********/app/build/intermediates/module_and_runtime_deps_classes/preliveGooglePlayRelease/base.jar:kotlinx/coroutines/debug/AgentPremain.class, /Users/*****/***********/voip_ui/build/intermediates/module_and_runtime_deps_classes/preliveGooglePlayRelease/feature-voip_ui.jar:kotlinx/coroutines/debug/AgentPremain.class

As I see

AgentPremain

indeed contained in both

kotlin-coroutines-core

and

kotlin-coroutines-core-jvm

with sligtly different byte code. The interesting thing is that

kotlin-coroutines-core

introduces the dependency to

kotlin-coroutines-core-jvm

according to Gradle dependency graph:

+--- org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.7 -> 1.4.3
|    \--- org.jetbrains.kotlinx:kotlinx-coroutines-core-jvm:1.4.3

and the POM file says the same:

<dependency>
	<groupId>org.jetbrains.kotlinx</groupId>
	<artifactId>kotlinx-coroutines-core-jvm</artifactId>
	<version>1.4.3</version>
	<scope>compile</scope>
</dependency>

I wonder how two dependent libraries could have the same class but with different code inside them? Maybe I do something wrong?

Can some1 help me understand when the awaitClose gets called:

@Test
fun `test flow` (): Unit = runBlocking{
    val flow = callbackFlow {
        trySend(Unit)
        delay(5000)
        close()

        awaitClose {
            println("Closed")
        }
    }

    val job = launch {
        flow.collect {  }
    }
    job.cancel()
}

The doc says on close and cancellation. But in this case, it is not called. Why ? When it is cancelled, awaitClose not called.

Trying to use Flow for the first time for a piece of app state. my app state is basically whether or not the user is logged in. I'm creating my flow this way

@Singleton
class AppUserManager
@Inject
constructor(@ApplicationContext val context: Context) {
    var loggedInFlow = flow<Boolean> { false }
...

and then there is a login call in the appUserManager

fun login(token: String) {
    loggedInFlow = flow { true }
...
}

then in my apps ViewModel I listen to this flow via

@HiltViewModel
class HomeScreenViewModel
@Inject
constructor(
    var service: ApiService,
    private val appUserManager: AppUserManager
) : ViewModel() {

    init {
        reload()
        viewModelScope.launch { appUserManager.loggedInFlow.collect {
            reload()
        } }
    }
...

This doesn't work. I'm assuming I'm missing something fundamental here about how to observe a flow. Any tips for me?

What are the consequences for a Kotlin/JVM application when I use a library that depends on the -native-mt version of coroutines lib?

Is there an established convention on how to propagate structured concurrency in a hierarchy of classes (with progressively smaller lifecycle)? When using functions this comes for free. With objects, you have a handful of options, • pass parent job to the constructor, object can define its private scope using job as a parent • have object implement CoroutineScope, parent should cancel() it • have object expose a cancel/release function • probably more The first seems less error prone to me, but also something I have rarely seen in other people's code.

Hello everybody :) I wished to ask about flows behaviour, especially in context of

MutableSharedFlow

- what shared flow will do in case if nobody is listening for its stream? As I understand, currently all hot flows will throw out data when there are nobody to get it. Or I missed something? If not - is there some built-in flow which allow to prohibit trashing of data in case when nobody listetning that flow?

What is the difference between GlobalScope.launch(Dispatchers.IO) {} And CoroutineScope(Dispatchers.IO).launch{} I have read from this StackOverflow https://stackoverflow.com/questions/65008486/globalscope-vs-coroutinescope-vs-lifecyclescope that both are same except the syntax differs but there has to be some difference so that we can use each one accordingly.

Is there a certain trick to debugging (on android) coroutines that may be executing on different threads when using break points? I see the switch thread option, it opens to that executing line and i start stepping but then it hangs (probably switched back to the other coroutine would be my guess)

Hi anyone who would know ? I'm trying to do a flow of ints and trying to detect if the sequence is increasing or decreasing based on last 2 numbers...

Is there any dangers of using .stateIn Flow extension with custom background CoroutineScope inside your ViewModel?

Is it possible to transform a flow into a flow that emits a rolling window of N elements? As if:

flowOf<Int>(0,1,2,3,4,5,6,7,8,9).magicOperator(2).collect { values: List<Int> ->
 print("${values}; ")
}

Would print:

0,1; 1,2; 2,3; 3,4; 4,5; 5,6; 6,7; 7,8; 8,9

if proguard obfuscation is breaking coroutines

await()

(it doesn’t wait 😂 ), where should I start digging? Tried these in `proguard-rules.pro`:

-keepclassmembernames class kotlinx.** {
    volatile <fields>;
}
-keepclassmembers class io.ktor.** {
    volatile <fields>;
}
-keepclasseswithmembers class io.netty.** {
    *;
}
-keepnames class kotlinx.** { *; }
-keepnames class io.netty.** {
    *;
}
-keep class io.ktor.** { *; }
-keep class kotlinx.coroutines.** { *; }
-keep class kotlin.reflect.jvm.internal.** {*;}
-keep class kotlin.text.RegexOption {*;}

what is the correct way to use suspend functions inside

androidTest

source set Currently I'm testing Room DB Dao in Android and writing test like this

runBlocking {
    movieDao.insertCategory(popularMovies)
}

I think that there might be a better way to write test for such scenario because if I have 20 test than I have to write

runBlocking

20 times

How to get a boolean value from 

dataStore Preference

 *synchronously*(one-shot) without delay?

I want to downlaod list of assets in parallel, is this correct?

private suspend fun prefetchStories(stories: List<Story>) {
    coroutineScope {
        val fetches = mutableListOf<Deferred<Unit>>()
        for (story in stories) {
            fetches += async {
                prefetchStory(story)
            }
        }
        fetches.awaitAll()
    }
}

Unsure about the

coroutineScope

. Most example don't use it but I presume that means they're using GlobalScope, right?

Hi guys, is alteration to different fields of a same object thread - safe? lets say I have this dto class:

data class Example(
var fieldA: Int,
var fieldB: Int
)

and that i change fieldA in thread 1 and fieldB in thread 2 (of a same object of class Example). would this be thread-safe? and if not so, how would I achieve thread safety?

Is there a reason trySendBlocking() is JVM only? How can I achieve this behavior in Kotlin/Native? My use case is adapting a callback API to Flow, without potentially dropping emissions as

trySend()

could.

val flow = callbackFlow {
    val job = addListener {
        trySendBlocking(it) // unavailable in Kotlin/Native
    }
    awaitClose {
        removeListener(job)
    }
}

Hey! I am working with Redisson locks and Kotlin coroutines. The issue is that the locks are based and acquired by a thread (with its thread id), however considering different coroutines can actually be scheduled on the same thread the lock mechanism won’t work across different coroutines (cause they have the same thread id). Do you have any idea on how I could fix this ? Is there a way to have a coroutine id for example ? Thanks