Hey guys, whats the difference between

fun CoroutineScope.foo() = launch{//some stuff}
vs
fun foo() = CoroutineScope { launch{//some stuff}}
vs
fun foo(){ coroutineScope { launch { // some stuff } } }

? Thanks

Hi everyone, I was wondering if there is a way to get a callback when a

suspendCancellableCoroutine

completes so I can unsubscribe from the callback that might return more values later on. My use case is that I'm trying to get the user from Firebase using the

onAuthStateChanged()

listener as they suggest here. However, I don't want to keep listening for this after getting it at the start of the app and if I don't unsubscribe from it then I get the error

IllegalStateException: Already resumed, but proposed with update

. This is an example of my code:

suspend fun getUser(): User = suspendCancellableCoroutine { continuation ->
    val unsubscribe = firebaseAuth.onAuthStateChanged(
        nextOrObserver = { user -> continuation.resume(user) },
        error = { error -> continuation.resumeWithException(...) }
    )
}

What would the idiomatic way to set a

CompletableDeferred

as a mirror of another one? As in, I have two or more `CompletableDeferred`that I know are waiting for the same value, so I'd like to ‘merge' them so that completing the ‘master' automatically completes the ‘slaves'. So far I've come up with:

fun <T> CoroutineScope.merge(
  master: CompletableDeferred<T>,
  slave: CompletableDeferred<T>
) = launch {
  master.await().also { result ->
    slave.complete(result)
  }
}

However this creates a new coroutine, and I'm wondering if there's a better way to achieve a similar goal.

Hello everyone, is it a bad practice to expose flow inner channel like this?

fun scan() = callbackFlow<Device> {
    deviceChannel = channel
    hardwareChecker.scanDevices()
}

I have this problem where I need to listen for the Redis queue when message gets sent to it. I tried this. However, this is not working. Is there anyway to keep listener running in the background while producer sends the message.

suspend fun processRequest(request: Request): Result = coroutineScope {
    val jobId = UUID.randomUUID().toString()
    val deferredMessage = async {
        redisOperations.listenToChannel(jobId).awaitFirst()
    }

    val deferredProducer = async(start = CoroutineStart.LAZY) {
        adtMessageProducer.send(adtRequest, jobId)
    }

    deferredProducer.await()
    deferredMessage.await().message
}

@Test
fun `should fetch JsonObject when call executes successfully`() = runBlockingTest {

    coEvery { webClient.get().uri(any<URI>()).awaitExchange().awaitBody<String>() } returns "{}"

    val result = service.fetchData(paramsMap, transactionId)
    assertThat(result).isNotNull
    assertThat(result).isInstanceOf(JsonObject::class.java)
}

Hi Guys, my test is hanging for some reason, could you help me figure out why?

Will

CoroutineExceptionHandler

never handle

CancellationException

? That seems to be the case (see example program in thread) but I can’t find a clear authoritative statement on that this is the case. For example, docs for interface CoroutineExceptionHandler states

when no handler is installed […] CancellationException […] is ignored

rather than “CancellationException is always ignored with CoroutineExceptionHandler”

Any idea why this code is not returning? It keeps suspended forever

userRepository.streamUser().filterLoggedIn().firstOrNull()

The repository returns a Logged Out user, but I'm only interested in logged in users, I would expect that since this flow becomes an empty flow, then

firstOrNull

should return null, right or am I missing something?

Hey, I’m working on a library that awaits some

CoroutineScope

as an input in order to

launch

some new coroutines. If any of these coroutines throw an exception it should delegate them to the initially given

CoroutineScope

. However, I figured out, if this scope has a

SupervisorJob()

all exceptions of my launched coroutines are just silently swallowed, which is probably working as intended. Is there any way I still can crash the application? I tried to create my own scope via

CoroutineScope(initialScope.coroutineContext + Job())

, but it still keeps the behavior of the supervisor job.

Hello, I’m implementing some tests but right now I’ve hit a wall with the following:

@Test
fun fetchAvailableStuff() {
    runBlockingTest {
        val job = launch(Dispatchers.Main) {
            api.fetchAvailableStuff { success ->
                println("----------- TEST -----------")
                println(success)
            }
        }
        job.join()
    }
}

But it gives me:

java.lang.IllegalStateException: This job has not completed yet

I already tried some of the solutions that are in this channel but with no luck

Hey, I'm launching a long-running coroutine in some scope. Is there a way to obtain its

Job

from my scope later (e.g by marking it in some way)?

📣📣 📣 Rethinking atomic cancellation in channels: For a good reason we propose to introduce a breaking change in one corner-case of

kotlinx.coroutines

library in the next major release. It concerns "atomic cancellation" behavior of

Channel

send

and

receive

functions. It might affect your code if you've ever used channels to transfer resources (references that need to be closed, like open files, sockets, etc) between coroutines via channels and if you were relying on those guarantees to ensure that the corresponding resource get closed by either sender or receiver and thus never leaks. If this is the case, then, please, study the details of the proposed change and voice your concern if the change is going to negatively affect the code you maintain: https://github.com/Kotlin/kotlinx.coroutines/issues/1813

From the

withContext

docs:

The resulting context for the block is derived by merging the current

coroutineContext

with the specified

context

using

coroutineContext + context

(see

CoroutineContext.plus

).

From the

CoroutineContext.plus

docs:

Returns a context containing elements from this context and elements from other

context

. The elements from this context with the same key as in the other one are dropped.

From these docs I conclude that any

CoroutineExceptionHandler

in the

context

in

withContext(context)

call replaces the one from the parent scope of

withContext

. So why does @Manuel Vivo write in his excellent article (https://medium.com/androiddevelopers/coroutines-patterns-for-work-that-shouldnt-be-cancelled-e26c40f142ad):

CoroutineExceptionHandlers don’t work as you’d expect when the context is used in withContext since the exception will be re-thrown.

So how come that a thrown exception is handled by the parent exception handler (which might rethrow it) , instead of handled by the overridden exception handler?

Does anybody here know what resources (Java threads) coroutines are running on if you run ktor with netty? Especially how the http thread pool vs worker pool works? E.g. If I get a request in a route and then start to call a service layer without changing the dispatcher myself

Has somebody replaced a ThreadPoolExecutor + BlockingQueue with Coroutines? The Thread pool acts as a spooler for the blocking queue. Jobs can continue to arrive indefinitely while some of them are executed. Primarily each jobs does I/O, so I want to get rid of the current blocking solution in favor of a coroutine based solution. Right now I can control also the number of parallel jobs (which are tied to the number of threads) and wonder if I could still do the same with Coroutines. Any advice greatly appreciated!

Is there a ticket or issue to track the status of the Actor API?

After cancelling scope, should I throw away the object? For example, in android activity lifecycle, I'd want to have coroutine started on resume and cancelled on paused, but for every lifecycle pass, I'd need a new scope object, correct?

For the last couple of weeks, I worked really hard on a new open-source project called Kotlin Coroutines – Use Cases on Android. Today, I am really proud to present “version 1.0” 🙂 I hope this project is helpful for you!

Hey Folks, we are running an nice little spring-boot webflux service in Kotlin. To brigde between coroutines and the webflux world, we use org.jetbrains.kotlinx:kotlinx-coroutines-reactor. As you can see in this little example, I switch the context in the controller.

@RestController
class PageController(
    private val pageService: PageService,
) {
    private val context = <http://Dispatchers.IO|Dispatchers.IO>

    @GetMapping(PAGES_BY_SLUG)
    suspend fun getPage(
        @PathVariable slug: String,
    ): PageResponse = 
        withContext(context) { pageService.getPage(slug) }
   }

}

My motivation is to keep the webflux threadpool clean from everything that does not relate to the request till it reaches the controller. But does really have an positiv affect on performance? Or does the context switch here is even bad because it comes with an cost?

I want to control whether

delay(..)

reports IDLE or not in an Espresso test dynamically at runtime. What would be the right way to handle this?. For example, if a

delay(..)

launched on

Dispatchers.Default

is currently suspended, I would like to be able to control at any point through the test whether the dispatcher should report idle or not

should I care about closing/cancelling channel during object teardown if it's used only internally in the object and coroutine scope to operate it was already cancelled?

Question on ktor + exposed + coroutines

How to achieve scatter-gatter or divide-conquer pattern with Coroutines? I am trying to achieve the below code, where I used ForkJoinPool to recursively divide list and join the results back. Please find a sample code below

internal class MyRecursiveTask(private val names: List<String?>) : RecursiveTask<String>() {
    override fun compute(): String = if (names.size > MIN_LIST_SIZE) {
        val mid = names.size / 2
        val myRecursiveTask1 = MyRecursiveTask(names.subList(0, mid))
        val myRecursiveTask2 = MyRecursiveTask(names.subList(mid, names.size))
        myRecursiveTask1.fork()
        myRecursiveTask2.fork()
        val results = ArrayList<String>()
        results.add(myRecursiveTask1.join())
        results.add(myRecursiveTask2.join())
        concat(results) // This just concats strings
    } else {
        concat(names) // This just concats strings
    }

    companion object {
        private const val serialVersionUID = -5978274303314688527L
        private const val MIN_LIST_SIZE = 2 // In real-world, DO NOT have it below 10,000
    }
}

Hi! I'm trying to parallelize some work, to improve performance. Am I doing it right?

override fun foo(): List<String> =
    runBlocking {
        fooSuspended()
    }

suspend fun fooSuspended(): List<String> =
    coroutineScope {
        val res: List<Deferred<String?>> = (1..100).map { n ->
            async {
                doWork()
            }
        }
        res.mapNotNull { it.await() }
    }

fun doWork(): String? {
    Thread.sleep(500)
    return "hi"
}

to get the terminology right, these are green threads, right?

Is it possible to do cooperative cancellation with just the standard lib? I know kotlinx.coroutines offers APIs for this. But I’m wondering if there’s basic support in the standard lib directly. I haven’t seen anything so far. I’m trying to avoid using kotlinx if possible due to JS bundle size concerns.

Is there a good pattern for writing coroutine-based tasks yet? Is this one okay? 🤔

same as above, no classes needed and no side-effects in the constructor 😄

Oh, looks like that we must not pass a

Job

to

launch

as it breaks parent/child-relationship 😅 We’ve got to take the

Job

returned by

launch

instead.