Hi everyone, I’m getting

IllegalStateException Flow invariant is violated emission from another coroutine is detected

when running this code, is it possible to fix this without using channelFlow ?

flow<Action> {
    val result = downloadFile(url, directory) { event: ProgressEvent ->
        emit(DownloadProgress(event))
    }
    emit(DownloadFinished("path"))
}.onEach { L.d(it) }.launchIn(viewLifecycleOwner.lifecycleScope)

downloadFile method

suspend fun downloadFile(
    url: String, fileDir: String, onProgressEvent: suspend (event: ProgressEvent) -> Unit = {}) {
 val progressChannel = Channel<ProgressEvent>()

 CoroutineScope(coroutineContext).launch {
     progressChannel.consumeAsFlow().collect { onProgressEvent.invoke(it) }
  }
/* rest of the code */
}

What is the difference between

(1..3).asFlow().onEach { number -> println("$number") }.collect()

And

(1..3).asFlow().collect { number -> println("$number") }

Why would I want to use

onEach

when I can use

collect

directly?

Ok I'm still learning about coroutines, and I'm a bit confused with the coroutines builders zoo... Is this all correct? • launch: Creates a new coroutine. Code below

launch {}

continues being executed instantly (although because structured concurrency, coroutine will finally wait for its children to finish). • async: Creates a new coroutine. Code below

async {}

continues being executed instantly (although because structured concurrency, coroutine will finally wait for its children to finish). • runBlocking: Creates a new coroutine. Execution is blocked. Code below

runBlocking {}

won't execute until runBlocking has ended. • coroutineScope: Creates a new coroutine. Execution is suspended (it's the suspending version of runBlocking). Code below

coroutineScope {}

won't execute until coroutineScope has ended. Now, if this is correct, the only coroutine builder whose name makes sense is runBlocking 🤷‍♂️ The other ones seem to have absolutely random names. The 4 coroutine builders "launch" the coroutine, two of them are "asynchronous" in some sense (both launch and async), and 4 of 4 again have a "coroutine scope", what am I not understanding here? Why do they have these names?

Can someone please help me with a sensible close to real-world usecase/example, where Coroutines are better than ExecutorService (say with pool count = no.of cores), for NON-blocking operations? (without using

delay()

)

Hey guys. I was playing around with using

Result

as a return type and ran into a weird issue. Does anybody know if this is already known? I couldn't find anything in the issue tracker.

suspend fun foo() = suspendCancellableCoroutine<Unit> {
        thread {
            it.resumeWithException(RuntimeException("foo"))
        }
    }

    suspend fun fooWithResult(): Result<Unit> = runCatching { foo() }

    @Test
    fun resultTest() = runBlocking {
        // when fooWithResult is not inlined the exception is not caught by "runCatching"
        val result = fooWithResult() 
        Assertions.assertTrue { result.isFailure }
        delay(5000)
        Unit
    }

The test passes if I mark

fooWithResult

as inline. Without it the exception is not caught by

runCatching

When returning a Coroutines JDK Future to a Java library, what’s the best way to do this?

fun returnFuture(): CompleteableFuture<String> {
  return runBlocking {
    future {
      // blocking code
      "result"
    }
  }
}

or

fun returnFuture(): CompleteableFuture<String> {
    return GlobalScope.future {
      // blocking code
      "result"
    }
  }
}

We have coroutine code further up the chain that then calls GraphQL-java, which in turn is wired with DataFetchers that implement an interface returning a CompleteableFuture. Ideally we want to share the same coroutine context throughout (so we can leverage MDC via

MDCContext

, etc.)

I’m trying to understand how that rule works, particularly

this.coroutineContext[ContinuationInterceptor] as CoroutineDispatcher

class CoroutineTestRule : TestWatcher(), TestCoroutineScope by TestCoroutineScope() {

    override fun starting(description: Description?) {
        super.starting(description)
        Dispatchers.setMain(this.coroutineContext[ContinuationInterceptor] as CoroutineDispatcher)
    }

    override fun finished(description: Description?) {
        super.finished(description)
        Dispatchers.resetMain()
    }
}

What does it do exactly? While debugging, I see that

this.coroutineContext[ContinuationInterceptor]

returns a

TestCoroutineDispatcher

, is that because the rule implement TestCoroutineScope which has a TestCoroutineDispatcher as default dispatcher?

How would you go about making a blocking method such as

ByteBuffer.allocateDirect(bufferSize)

suspending, would you need to make a threadpool and do callbacks to unsuspend a coroutine or is there a better way?

Hi guys, question on the @ExperimentalCoroutinesApi, is that safe to use them in production? I’ve seen a bunch of interesting components marked like that but I’m a bit scared to use them because of the annotation. What do you do in this case?

In a flow why doesn't catch capture errors for an onEach?

Hi! In a coroutine, we often want to use try-catch to catch all exceptions except

CancellationExceptions

. If we would catch

CancellationExceptions

, the Coroutine would continue to run until the next suspension point instead of cancel itself. Is there some convenience function in the stdlib that does this? Or have you written your own one?

message has been deleted

Hi! I saw around a bunch of example where suspending function gets defined with coroutineScope like in

suspend fun myFun() = coroutineScope {}

I wonder, given that a suspending function must be called within a scope, what’s the point of creating a new scope like in there?

We are using coroutines-debug with DumpCoroutines but we are not getting all the active coroutines in the dump. How is this possible?

Sorry folks, I have another question around suspending function and structured concurrency. As far as I understood, suspect modifier will make the function suspendable, which means that the underneath thread can continue doing some work. So far so good. Now say that I have a function like this

suspend fun doSomething() = coroutineScope {
   val channel: Channel<Int> = Channel(capacity)
   val results = aggregator.aggregate(channel)
   channel.close()
   results
}

where

aggregator

is a class that exposes a suspending function

aggregate

that uses the channel to listen to messages with

channel.receive()

Now, would

channel.close()

be ever invoked before the aggregator finishes to receive all the messages? Or since I’m not creating a new scope (by using launch or async for example) guarantees that each line of my snippet is executed one after the other finishes? And if that’s the case what’s the purpose of marking a function like

aggregate()

with a suspend modifier (although it suspends since it will wait for mesages in the channel)?

There really is no way to avoid passing the channel as an argument and avoid

produce

? I don't want to avoid produce, it's just for curiosity

can I combine kotlinx.coroutines.sync.Semaphore.acquire and kotlinx.coroutines.withTimeout ?

import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.withTimeout

private val concurrentRequests = Semaphore(50)

suspend fun foo() {
    withTimeout(10) {
        concurrentRequests.acquire()
    }
    try {
        doStuff()
    } finally {
        concurrentRequests.release()
    }
}

Hey guys, i'm new to coroutines and have being searching for some open souce web project to try to understand more and see how/where can i apply in real world cases, but couldn't found. Can you guys recommend any?

Hi guys, if I want to declare an interface with a method that runs in the scope of the caller, is that ok to do something like

interface MyGuy {
  suspend fun CoroutineScope.doSomething()
}

or is there a better way when it comes to interface declaration?

I loved the way the book "Kotlin in action" explains things, not only as a reference but as a way to go back to when I can write some code but not sure about some details. • Is there plans to make a new edition for coroutines? • is there a different book that covers coroutines? The official docs and the official / unoficcial blog posts are good, but they can't replace a good book for some situations.

Hi! I published a new blog post about “Why exception handling with Kotlin Coroutines is so hard and how to successfully master it!“. https://www.lukaslechner.com/why-exception-handling-with-kotlin-coroutines-is-so-hard-and-how-to-successfully-master-it/. I hope I can help a few people with this complex subject!

Hey all, so I want to prevent making duplicate requests to a server. Does it make sense to store a bunch of

Deferred<T>

s and reuse them? Or is this a coroutine anti-pattern? I'm pretty new to coroutines so I'm just wonder if there is a better way.

private val notesDeferred = HashMap<String, Deferred<Resource<String>>>()
suspend fun getNotes(id: String): Resource<String>{
    if(notesDeferred.containsKey(id) && notesDeferred.get(id)!!.isActive){
        return notesDeferred.get(id)!!.await()
    } else {
        val newRequest = GlobalScope.async {
            remoteSource.getNotes(id)
        }
        notesDeferred.put(id, newRequest)
        return newRequest.await()
    }
}

Hi guys, I’m experimenting exception handling inside the launch, and I’ve created a very small test to do this

@Test
    fun testingScope() = runBlocking {
        myfun1()
    }

    suspend fun myfun1() = coroutineScope {
        val handler = CoroutineExceptionHandler { _, exception ->
            println("CoroutineExceptionHandler got $exception")
        }

        launch(handler) {
            throw Exception()
        }
    }

the test is failing because the exception gets bubbled up to the runblocking although I’m using an exception handler. Doing some debugging I’ve seen that the handler is set in the right child context, but still never called. Does anybody know why?

On a similar note, I see old dependencies in ktor-client-core-jvm as well:

|    +--- io.ktor:ktor-client-core-jvm:1.4.0
|    |    +--- org.jetbrains.kotlin:kotlin-stdlib:1.4.0 (*)
|    |    +--- org.slf4j:slf4j-api:1.7.30
|    |    +--- org.jetbrains.kotlin:kotlin-stdlib-common:1.4.0
|    |    +--- org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.8-native-mt-1.4.0-rc -> 1.3.9 (*)

as well as ktor-network:

|    |              +--- io.ktor:ktor-network:1.4.0
|    |              |    \--- io.ktor:ktor-network-jvm:1.4.0
|    |              |         +--- org.jetbrains.kotlin:kotlin-stdlib:1.4.0 (*)
|    |              |         +--- org.slf4j:slf4j-api:1.7.30
|    |              |         +--- org.jetbrains.kotlin:kotlin-stdlib-common:1.4.0
|    |              |         +--- org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.8-native-mt-1.4.0-rc -> 1.3.9 (*)
|    |              |         \--- io.ktor:ktor-utils:1.4.0 (*)

@pambrose You're getting kotlinx.coroutines 1.3.7, not 1.3.9

I am trying to enable Coroutine Debugging in Android Studio 4.0.1. I am using

org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.9

coroutines and the

1.4.0-release-Studio4.0-1

Kotlin plugin. However, I don't see coroutines in my debugger menu as shown below. I've restarted and have

Disable coroutine agent

un-checked. What am I missing?

infix fun <E, R> ReceiveChannel<E>.zip(
    other: ReceiveChannel<R>
): ReceiveChannel<Pair<E, R>> (source)

is deprecated. What is the alternative?

is there a way to close a channel and drop not yet consumed elements?

Hi! I’m using AppAuth on Android to login to my backend and coroutines (ktor) on my transport layer to call APIs on that backend. I need to pass an access token to each of my calls. Using AppAuth I refresh this token with the

AuthState.performActionWithFreshTokens(callback: (freshToken: String?))

which uses a callback to give me the fresh token. And my transport service uses coroutines, something like:

class TransportService {
  suspend fun getAllTasks(): List<Task> {
    authState.performActionWithFreshToken() { freshToken ->
      return ktorClient.get<List<Task>>()
    }
  }
}

Now, I have an error where I use the ktor client:

suspend function can be called only within coroutines scope

. That is because the callback function is not a coroutine scope. If I create a new scope within the callback, will this work well with the main function? What can I do to keep my transport with suspend function and not use completion handler everywhere? Thanks 🙂