I'm trying to use a websocket with ktor and I need to set up two while true loops. I've wrote this code which I thought would work but it only completes one iteration and then exits

val eventJob = launch {
    while (true) {
        val event = try {
            when (val frame = incoming.receive()) {
                is Frame.Text -> json.decodeFromString<Message.Event>(frame.readText())
                is Frame.Close -> {
                    println("Received close frame: ${frame.readReason()}")
                    continue
                }
                else -> continue
            }
        } catch (e: Exception) {
            println("Error while receiving: ${e.message}")
            break
        }
        println("Received event: $event")
    }
}

val heartbeatJob = launch {
    while (true) {
        delay(helloMessage.data.heartbeatInterval.milliseconds)
        println("Sending heartbeat")
        sendSerialized(Message.Heartbeat)
    }
}

joinAll(eventJob, heartbeatJob)

Why does the below logic crashes where as if i separate the top-level it doesn't. val toplevel = CoroutineScope(Dispatchers.IO) this crashes: toplevel.launch { val sync = async { print("this is sync 1") throw RuntimeException() } try { sync.await() } catch (e: Exception) { print("Exception handled") } } This does not: val sync = toplevel.async { print("this is sync 1") throw RuntimeException() } toplevel.launch try { sync.await() } catch (e: Exception) { print("Exception handled") } }

Are there situations in which you, instead of using a suspend function, use a regular function that returns a Flow that emits a single value? Maybe to use some flow operators that make the code a bit cleaner?

Hello is there any way to run two coroutines in parallel and then return the first result that finishes first? I've thought to use

Job

and then if the first finishes first cancel the second

Job

to stop doing the work, also I think the way to do it is using

launch{}

right? Instead of

async {}

The Coroutines docs discourage the use of

flatMapMerge

and

flatMapConcat

in regular application-specific flows, see attached image. What should we used instead when flatMapping two flows?

Is it possible to start observation when launching

StateFlow

? So the case is that I have

StateFlow

in my

ViewModel

and I would like to launch some observation(e.g. User status

Flow

) but I would like to stop this observation at the same moment that I will stop observing main

StateFlow

itself so they need to be launched in the same scope. Currently I’m using combine with status observation flow but I’m ignoring events emitted by this Flow, is there a better way to handle it? Side not: I'm not using values emitted by the second flow directly so that's why I need to ignore them.

What is fine common name for Iterator-like interface, which has suspend

hasNext

? ChannelIterator has similar contract

Is there something like

combine

but for when the flows (arguments) are calculated based on the item in the main flow? My use case is a flow of

Item

s where each

Item

has a number of properties whose values are provided by flows. Initially those properties are null and so the operator I’m trying to implement, would transform an

Item

such that the properties are populated and the resulting Flow will emit a new

Item

each time any of the properties updates (as well as when the original flow emits a new

Item

).

Hi all, I’ve got multi-platform Kotlin code which uses coroutines, and it is important that things really execute in parallel. For my specific use case there is no real problem except for when targeting JavaScript, as that’s single-threaded and a coroutine has to actually be suspended in order for other things to execute. The solution I’m currently using is to call

delay(1)

in several places, to achieve just that: when any coroutine hits the

delay(1)

its execution is suspended, allowing others to run (and complete or hitting

delay(1)

themselves, allowing suspended coroutines to resume). And it generally seems to work. But it’s not elegant, and the delay itself, while not that noticeable, is not necessary. Is there another way to do it? Something I can do at certain places in the code to allow a coroutine to suspend execution just in order to allow others to run until they complete or do the same?

what’s the equivalent of a Completable? like if I have a method that just deletes thing and I don’t care what ends up happening with it, I just want to know if it’s done or not

I’m running some tests that call

suspend

functions, so I’m using

kotlinx.coroutines.test.runTest

. This has the useful feature of skipping `delay`s and simulating actual scheduling with virtual time, which is super useful as it allows the tests to run fast, and one can also control the order of execution of various parts of different coroutines very delicately. However, I have a very specific use case where functions under test use `delay`s, and I actually want the `delay`s to execute for real, just like they would in production (this shouldn’t normally be required in pure unit tests, but is necessary in my case as the functions I’m testing interact with other systems). Is there a way to do that?

state.myList.forEachIndexed { index, it ->
  viewModelScope.launch {
    it.isDone.value = apiService.isDone(it.id)
  }
}

I thought I'd be able to add a

.await()

to the end of the launch block to know when all of those parrallel api calls are done, but that doesn't compile. Is there a more idiomatic way to do what I'm trying to do?

how do I turn this code into coroutines? https://hastebin.com/ukuxatodew.kotlin I figured out that I am able to do this

viewModelScope.launch(<http://dispatchers.io|dispatchers.io>) {
            val filterSources = try {
                listOf(
                    launch { getLocation.execute() },
                    launch { getCategory.execute() },
                    launch { getPeriod.execute() },
                    launch { getDateRange.execute() }
                ).joinAll().
            } catch (e: Throwable) {

            }

        }

but unable to figure out how to get that to a map where i can use the stuff inside

I have a really specific use case and I guess I need something to pause and resume the emission of a flow multiple times. To be more concrete: I need to pause a flow to wait for a confirmation from another flow before emitting the next item. Any ideas how to solve this?

how do I emulate the same behavior that Observable.combineLatest does? I have that inside a viewmodel and when I leave the screen and come back, that line is triggered again and fetches the values i’m requesting again if I just have viewModelSCope.Launch ( my stuff), it doesn’t get retriggered when coming back with new values

Hello ! Is it possible to unit test for job cancellation for a class like below ?

class Sample(private val scope: CoroutineScope) {
    
    private var currentJob: Job? = null
    
    fun doSomeAsyncWork() {
        currentJob?.cancel() // how to test this line ?
        currentJob = scope.launch { 
            delay(5000)
        }
    }
}

Is there an alternative to this https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-debug/ for android? We faced some issues with coroutines recently and the stack trace wasn't helpful so we were trying to explore to way to debug coroutines/ improve stack traces on android.

It seems to me that Im able to call a suspend function in a non-suspending block. Why is that? Details in thread 🧵

coroutines runs in main thread how can i execute coroutines in background thread. ( Dispatchers.IO ) ?

suspend fun getTopKeys(con: Context, rootObject: JSONObject): MutableLiveData<AuthResponse> {

    return withContext(<http://Dispatchers.IO|Dispatchers.IO>){
        val gson = Gson()

        val accessToken = MyPrefConstant.getAccessToken(con)

        val retrofit = RetrofitConfig.getRetrofitWithAccessToken(accessToken, con)

        val service = retrofit?.create(ApiInterface::class.java)

        var data = AesKotlin.encrypt(rootObject!!.toString(),AppConfig.ENCRYPTION_KEY)

        Log.e("App1>>>>", data.replace("[\n]", "")!!.toString())

        val call = service?.getTopKeys(data.replace("[\n]", "")!!.toString())

        call?.enqueue(object : Callback<AuthResponse> {
            override fun onResponse(call: Call<AuthResponse>, response: Response<AuthResponse>) {

                if (response.code() == 200 && response.body().status.equals("OK")) {
                    val responseData = response.body()!!
                    searchData.value = gson.fromJson(gson.toJson(responseData), AuthResponse::class.java)

                    val i = Intent("return.apiResponse")
                    i.putExtra("data",  gson.toJson(responseData).toString())
                    i.putExtra("method", "getTopN")
                    //con.sendBroadcast(i)
                    LocalBroadcastManager.getInstance(con).sendBroadcast(i)
                }else{
                    //  Toast.makeText(con, response.body().errorMessage, Toast.LENGTH_LONG).show()
                    val json = "{\"Status\": \"NOT OK\"," +
                            " \"ErrorMessage\": \"Something went wrong\" ," +
                            " \"Data\": \"NULL\", " +
                            " \"Timestamp\": \"NULL\" }"

                    searchData.value = gson.fromJson(json, AuthResponse::class.java)
                }
            }

            override fun onFailure(call: Call<AuthResponse>, t: Throwable) {
                Log.e("API","Error")
                val json = "{\"Status\": \"NOT OK\"," +
                         " \"ErrorMessage\": \"Something went wrong \"," +
                        " \"Data\": \"NULL\", " +
                        " \"Timestamp\": \"NULL\" }"
                searchData.value = gson.fromJson(json, AuthResponse::class.java)

            }
        })
        searchData
    }

}

In above method getTopKeys following methods are showing in mainthread in profiler

getJSON() - Related to gson

setValue - setting value to livedata

fromJSON - Related to gson

Does

runTest

not work with

@BeforeTest

?

I dont know why I’m not able to run this coroutines code in standalone IntelliJ, I have a clue as to the Looper not being there means something is trying to access something Android specific, but not much clue otherwise.. is it because of the

Main

Dispatchers called there? there’s no Main thread like UI dispatcher in kotlin itself https://hastebin.com/eyolacixis.properties

So about

runBlocking

, I read that it shouldn't be used from a coroutine due to the potential for deadlocks (and we do seem to be encountering deadlocks here) but is there any way to "detect" if you're inside a coroutine somehow and run suspending methods in that coroutine instead of using

runBlocking

? For reference, we're working inside a lot of pre-existing Java code that obviously can't be suspending, so I'm not sure how else to bridge it other than

runBlocking

, which obviously isn't great and doesn't seem to be working for us.

Pretty basic question, in cases like this:

val account = async { getAccountByUuid(accountUUID) }
val devices = async { deviceRepository.findDevicesByUuid(accountUUID) }
account.await()?.addDevices(devices.await()) ?: return@coroutineScope null

In case the account.await() is null, the devices.await() is still waiting or it is just canceled ? I believe it is canceled but just want to make it sure. Thanks in advance for any answers !

I'm converting some Java code that uses a BlockingQueue to use coroutines. I was starting to switch to using an actor, but I noticed that it is marked as being ObsoleteCoroutinesApi. What is the replacement? Also, why does the doc recommend this pattern if it is marked Obsolete? https://kotlinlang.org/docs/shared-mutable-state-and-concurrency.html#actors

Is there an interruptible callbackFlow (analogue to runInterruptible)? I.e., to timeout something that should get callbacks periodically that you’re trying to convert into a flow…

Something that came up yesterday when I was working with coroutines and thread locals: is there a way to use an inline higher-order function, but prevent its lambda parameter from suspending? Hypothetically something like:

inline fun <T> foo(f: nosuspend () -> T): T

Hello guys I am having problem with my synchronous call, my use case is I wanted to finish all collection before launching

doAnotherCall()

, the

viewModel.detailsState

collects in a loop. I mean its

emit and collect

in a loop adding delay in

doAnotherCall()

does not help. Best Regards.

private suspend fun collectProductList() = withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
    viewLifecycleOwner.lifecycleScope.launch {
        viewModel.detailsState.collectLatest { state ->
            when (state) {
                is ItemDetailState.ShowLoader -> binding.progressBar.visible()
                is ItemDetailState.HideLoader -> binding.progressBar.gone()
                is ItemDetailState.OnSuccess -> handleSubListSuccess(state.list, state.details)
                is ItemDetailState.OnFailed -> handleFailure(state.error)
            }
        }
    }
    launch {
        doAnotherCall()
    }
}

An easy one: I want a job queue, something that'd be like a PublishSubject in RxJava. I want an infinite stream that suspends if it goes over buffer and never drops values. I'm trying a MutableSharedFlow subscribed since object creation that acts on every new value:

val pub = MutableSharedFlow<WorkJob>().apply { 
  onEach { println("Each") }
  .onStart { println("Start") }
  .onCompletion { println("Complete") }
  .launchIn(GlobalScope)
}

...

somewhere else in code:

pub.tryEmit(WorkJob())

That triggers "start" but doesn't emit any value. The moment I put

MutableSharedFlow<Job>(replay = 10)

it works, printing "start" and "each". But I want an infinite stream, not a repetition or something that goes to the latest value.