given a

Job

, how can I fail it with an exception? I want to cancel it abnormally so its parent fails

I have this http api that I'm trying to make beautiful. It's paginated, and any api call(page 1,2,3,4,5 etc) could 429. So I have N queries that I'm converting via a helper into a flow of the entities themselves. So callers get a Flow of entities instead of having to implement the loop logic. In the example I'm linking, it's a "service" that have 10 pages of Integers that it returns. It fails randomly. The structure is a flow that

emitAll

a flow that has the retry appended. Again, it's all working perfectly, but something seems just off about it.. Seems like too many flows. Thanks https://pl.kotl.in/ysrpoKNa7

Can we build jars with a main class having

suspend fun main

, or do we have to convert it to

fun main

and add

runBlocking

?

Is this the best way to cancel a Deferred? Or is there a better way without the need to access the current `CoroutineContext`'s

Job

?

suspend <T> fun executeQuery(query: T) {
   val deferred = client.queryCall(query).toDeferred()

 currentCoroutineContext()[Job]?.invokeOnCompletion {
       if (it is CancellationException) {
          deferred.cancel(it)
       }
   }
   deferred.await()
}

I would like to send an event where the receiver may not be immediately listening for the event, but when it does, I would like that event to get sent to the receiver. I assumed I could use a Broadcast channel with a capacity of 1 and using the

send

suspend function would suspend the coroutine until receiver started listening but that doesn’t seem to be the case. Can anyone provide some input on how to make this work? Sender class

private val _myEvents = BroadcastChannel<String>(1)
val myEvents = _viewEvents.asFlow()

...
fun doSomething() {
   _myEvents.send("sample string")
}

Hi everyone, what is the dependency that I should use on Kotlin Multiplatform iosMain for coroutines? (compliant with the new 1.4.0 Kotlin version) I am getting this error:

Could not find org.jetbrains.kotlinx:kotlinx-coroutines-core-native:1.3.9.

@elizarov is it possible to use flows as a solution to back-pressure across multiple processes? [replicating data over TCP from process A to process B - trying to deal with process A not sending so much that process B can not keep updating its local state] Or should I use something different?

@elizarov Is there a plan for native/iOS multi-threaded coroutines to support a dispatcher backed by more than 2 threads? If so, what's the super rough timeline? Would be great to be able to fully utilize the CPU.

Hey everyone! I am trying to test my 

StateFlow

 by asserting the list of values received by it.

data class State(val isLoading: Boolean, val name: String?)

fun main() {
    val mutableStateFlow = MutableStateFlow(State(false, null))
    val stateFlow = mutableStateFlow as StateFlow<State>

    val scope1 = CoroutineScope(Job())
    val scope2 = CoroutineScope(Job())

    scope1.launch {
        delay(1000)
        mutableStateFlow.value = State(true, null)
        mutableStateFlow.value = State(false, "name")
    }

    scope2.launch {
        stateFlow.collect {
            println("value = $it")
        }
    }
    Thread.sleep(2000)
}

Here is my code. Instead of getting 

State(true, null)

 as the second value, I only receive the initial value and the last value (

State(false, name)

)

Is converting a

Flow

to an

Observable

safe? The reason I’m wondering is because the

Observable

can be cancelled, but the

Flow

can’t. Is the suspending behavior of the coroutine sufficient? With Kotlin Multiplatform for the JS side we’re converting `Flow`s to

rxjs

Observable

s. Code inside the thread.

So I've got a bad idea: What if we embedded a progress bar control and a delegating

CoroutineDispatcher

into a coroutineContext such that we could ~automagically make progress bars based on the number of suspension points? I think this is akin to reflecting on the number of states the coroutine state machine has. Of course, while-loops and if-statements blow this up, but still I feel like there might be something here. Has anybody got an implementation doing something like this?

I’ve been trying to wrap a library that needs to have an initialize method called. This method takes a callback and calls that callback when an object is ready. Now I also want to expose suspend methods that internally use the object that the callback receives. So these suspend methods should suspend waiting for the object until we can then execute normally. I’m not sure how can I suspend waiting for the object in a class-wide scope, as initialize will only be called once at init block. Any ideas? Thanks!

Which Coroutine scope do I use in an Android MVVM Repository? I want to execute a Retrofit call when the repository is created and store its result in a MutableLiveData object within it. That way as long as the repository is not destroyed the use could navigate to the screen and the data would be available. It won't be downloaded over and over if I were to call the retrofit call in the viewModelScope.

Is there a way to use

shareIn

/

stateIn

before it is released? Do i need to use a

-SNAPSHOT

dependency? Which one? thanks.

Hi, I have a question about performing CPU bound operations Should I be using

Dispatchers.Default

or create separate scope with fixed thread pools equal to number of cores? what are the pros and cons? what are the criteria's to choose between one of this approach? for ex. Library writer, App writer etc. I have posted detailed explanation and my use cases here https://stackoverflow.com/questions/63915078/can-i-use-dispatchers-default-for-blocking-cpu-bound-operations

Does kotlinx.coroutines require a new version for 1.4.10? 1.4.10 seems to be released, but I haven’t seen any announcement or anything.

Hello, I got a

Flow

like

flow {
    while (true) {
        emit(something)
        delay(interval)
    }
}.flowOn(injectedDispatcher)

In my test, I inject the dispatcher and test like

val result = mutableListOf<Int>()
val interval = 30.seconds
val flow = getSavedCredentials(interval)
val job = launch {
    flow.toList(result)
}

repo.add(1)
advanceTimeBy(interval)

repo.add(2)
advanceTimeBy(interval)

repo.add(3)
advanceTimeBy(interval)

So I would expect

result

to be

[ 1 ], [ 1, 2 ], [ 1, 2, 3 ]

But instead it is

[ 1, 2, 3 ], [ 1, 2, 3 ], [ 1, 2, 3 ]

What’s wrong with it? 😵

Am I correct in thinking that

launch {
  singletonReceiveChannel.consumeEach { printLine(it) }
}

will produce a memory leak due to

consumeEach

never getting canceled, even if `launch`'s CoroutineScope ends?

This may sound stupid but believe me when I do

users.postValue(repository.getUsers(20))

, the getUsers function executes but the value isn't posted but when I do

list = repository.getUsers(20)

and then do

users.postValue(list)

it works totally fine. What's the deal here?

How would you write a suspend function to share the result of a function between parallel requests? Let me explain with pseudo-code

private var shared: Deferred<Result>? = null

private suspend fun actuallyGrabResult(): Result { ... }

suspend fun grabResult(): Result {
  val deferred = shared.takeUnless( !shared.isActive }
    ?: async { actuallyGrabResult() }
        .also { shared = it }
  return deferred.await()
}

The idea is that only 1

actuallyGrabResult()

coroutine at a given time should be running and any request to

grabResult()

should wait for it and get the result. Until the result is delivered, next requests will cause another call. With RxJava you'd do something like:

private val shared: Single<Result> = actuallyGrabResult()
        .toObservable()
        .replay(1)
        .refCount()
        .singleOrError()

private fun actuallyGrabResult(): Single<Result> { ... }

fun grabResult(): Single<Result> {
  return shared
}

Is there an idiomatic way of doing this with coroutine? (I've only recently started using coroutines)

I found a code from Ktor has

(dispatcher as Closeable).close()

inside of

GlobalScope.launch(super.coroutineContext, start = CoroutineStart.ATOMIC)

Intellij IDEA reported it as inappropriate blocking method calls. How I can fix it?

Hello! What is the status of coroutines multithreading on KMP? I’ve heard that it is not supported atm, but probably it was changed in 1.4?

I have a small problem when converting a callback to a suspend function. Im trying to wait for a field to be created on a document (firestore) and when the field is created, return the data and continue execution. It works, but since the document listener still is active, Ill get an exception on the next change of the document (alreadyResumedError)

suspend fun waitForField(id: String): DocModel? = suspendCancellableCoroutine {
    val listener = db.collection("collection").document(id)
        .addSnapshotListener { snapshot, e ->
            if (snapshot == null) return@addSnapshotListener
            val res = snapshot.toObject(DocModel::class.java)
            if (res?.calculation != null) {
                // here i have to disconnect the listener
                it.resume(res)
            }
        }

    it.invokeOnCancellation {
        listener.remove()
    }
}

Did something change with the “Inappropriate blocking method call” inspection? I’m see it in places like this now:

val serverModel = withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
            clientAssociation.retrieveModel()
        }

where

retrieveModel()

throws

IOException

. Is this not an appropriate use of

<http://Dispatchers.IO|Dispatchers.IO>

?

I believe at one point there was a way to create a dispatcher or a context that borrowed threads from another dispatcher, but also ensured that only one thread was accessing the coroutine at a time. Was that ever implemented?

We love how #ktor works, but are really missing some insights into queuing/delays for co-routines. Would really love to have a way to collect metrics for those things. Right now we suspect it's getting backed up by some slow requests, but hard to diagnose exactly in production. Anyone know of a temporary way to do that while we wait for this issue? https://github.com/Kotlin/kotlinx.coroutines/issues/1360

Hey all, so I've just started using 1.3.9-native-mt on iOS/native and I'm getting concurrency/freezing exceptions on code that would otherwise run correctly on the JVM. Is there a simple way to fix this code? Can I just force everything to run on the main thread?

I see here https://mvnrepository.com/artifact/org.jetbrains.kotlinx/kotlinx-coroutines-core that there's such thing as 1.3.9 and 1.3.9-native-mt I only need regular old coroutines. What's 1.3.9-native-mt?

I've got an app that only uses coroutines in one spot (where I'm using ktor) and I want to fire-and-forget a set of requests (I don't want to block the caller on the responses, I just log when responses are received or when they time out). I have an existing IO thread pool for tasks like this. I was originally doing (simplified):

fun my update() = runBlocking {
    subscribers.forEach { url -> 
        launch(myIoPool.asCoroutineDispatcher()) {
            try {
                val resp = client.postJson(...)
                <http://logger.info|logger.info>("Got response...")
            } catch (e: HttpRequestTimeoutException) {
                logger.error("Timed out...")
            }            
        }
}

But I've now realized this is blocking until all the calls to

launch

finish. What's the best way to fire-and-forget here? I think I've seen I can do this with

GlobalScope.launch

(and getting rid of the

runBlocking

), but from what I've read it seems like it's rarely a good idea to use that.