What is the correct way to implement a cache? I'm thinking of something like:

val cache = HashMap<Id, Deferred<Value>>()
val lock = Semaphore(1)
val cacheJob = SupervisorJob()
val cacheScope = CoroutineScope(cacheJob)

suspend fun get(id: Id) {
  val r = lock.withPermit {
    if (cache[id] == null) {
      val c = CompletableDeferred()
      cacheScope.launch {
        c.complete(getElementFromNetwork(id))
      }
      c
    } else cache[id]
  }

  r.await()
}

I think that's generally safe, but I'm not sure if it will correctly handle all cases (the ID is invalid, ...)

Hi everyone! I'm trying to design a Kotlin API for ZeroMQ. My API currently resembles a lot to the `SendChannel`/`ReceiveChannel` API. I would like to implement some

onReceive: SelectClause1<Message>

and

onSend: SelectClause2<Message, SendSocket>

similar to the ones of the

Channel

API. Searching in this channel, I found: 1. some helpers to build select clauses https://gist.github.com/zach-klippenstein/fa4366388295282fa409c5085abada23 by @Zach Klippenstein (he/him) [MOD], however it seems to me that there are made to work with event callbacks, 2. a suggestion by @Matej Drobnič to use something like

async { mySuspendFunction() }.onAwait

, however I can't seem to build a proper

SelectClause1

with this. As ZeroMQ uses a polling mechanism, I don't quite figure out how to use either of these solutions. Also, the

AbstractChannel

code is quite overwhelming. Is there a paper/article discussing the possible

SelectClause

implementations somewhere? Do anyone of you have some hints or advices to give me? Thanks in advance!

Hi, I am using a callbackFlow{} and I am not sure if I should use `offer`(deprecated) or

trySend

(experimental API). Could someone tell me what is the best choice?

In crashlytics I'm seeing bunch of caught exceptions like this

Non-fatal Exception: kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled

Non-fatal Exception: kotlinx.coroutines.JobCancellationException: Job was cancelled

and no stacktrace; are there supposed to be ignored?

I recently started using

Result

as a return type as of Kotlin 1.5. I’m having seemingly random crashes with suspending functions that return

Result

(I haven’t checked with other inline/value classes). it’s happening in the generated

invokeSuspend

:

Fatal Exception: java.lang.ClassCastException
java.util.ArrayList cannot be cast to kotlin.Result

I don’t have reliable repro steps for it, it seems to happen on some android devices, at different places but all the same crash, but not always 😢 I’m really stumped with it. I’ve been looking at the difference in the decompiled bytecode between suspending and non suspending functions that return

Result

, I can see casting of some

Object

to

Result

, assuming that’s what’s failing 😕

I have a concurrent execution of some tasks using async and awaitAll. How do I wrap the awaitAll in such a way that I can handle each task failure separately. Currently only the first failure is caught. I am using supervisor scope

I read that coroutine delays are not precise. What will cause a

delay

to take longer than the specified time? And do you think this will work to get the actual passed time after a delay (for example for a countdown timer)?

val timeBefore = SystemClock.elapsedRealtime()
delay(TICK_DELAY)
val timeAfter = SystemClock.elapsedRealtime()
val passedTime = timeAfter - timeBefore

Hello guys do you have any example of coroutine 

SharedFlows

 

shareIn

 in github or gitlab, I couldn't find good resources or articles with a sample. What I am trying to achieve is combine(flow1, flow2) with 

shareIn

, so I only need to

getfetch

once and share this to at least 3 fragments. I really appreciate if someone could help. Best Regards to all.

Is there any reason that there is no

SendChannel<E>.sendCatching(e: E): ChannelResult<Unit>

like there is a

ReceiveChannel<E>.receiveCatching(): ChannelResult<E>

?

I was wondering, is it safe/good idea to use

coroutineScope

inside a flow operator? I have a flow I’m mapping on, and I want, depending on the value do some multiple suspending calls that I would optimally prefer to be done in parallel. Specific example inside the thread 🧵

Hi. I'd like a review / feedback of an approach I'm using, joining two libraries outside of our control - one that uses CompletableFuture<>, and one that uses

suspend fun

hey guys, is it possible to test Flows on Kotlin Multiplatform (

commonTest

,

androidTest

,

iosTest

) ? how to deal with Coroutine Context?

Question: is looking up the dispatcher a job is running on as simple as doing myJob[CoroutineDispatcher]? More generally, what’s the best way to check from a unit test which dispatcher a given job was run on, short of mocking <scope>.launch? Or is that in fact the best way

When it comes to structured concurrency, it's often said that a class that encapsulates asynchronous operations (for example, a cache, an http client) should take a constructor parameter to explicitly setup structure. Should that parameter be a

Job

or a

CoroutineScope

? So far I've been adding

Job

parameters, but then I have to declare another

CoroutineScope

field per class.

Should I make a suspend network call in a

viewModelScope

or inside compose function with side-effect with compose?

Can we do something like this, but with

CoroutineScope

instead of a

Job

? I need to launch a few suspend functions, and cancel them if a) the parent scope is canceled b) the VH scope is canceled (whatever comes first)

Does the nested

launch

block around

taskDao.increaseMillisCompletedToday

(which is a database operation) look dangerous? I use this to measure time but without the nested

launch

, I think the length of the database operation itself is causing imprecision. The tick interval is every 1s.

fun startTimer() {
    timerJob?.cancel()
    timerJob = applicationScope.launch {
        val selectedTask = selectedTask.first()
        if (selectedTask != null) {
            taskDao.updateLastActiveTimestamp(selectedTask.id, System.currentTimeMillis())
            startTimerService()
            timerRunningFlow.value = true
            while (true) {
                val timeBefore = SystemClock.elapsedRealtime()
                delay(TICK_DELAY)
                val timeAfter = SystemClock.elapsedRealtime()
                val elapsedTime = timeAfter - timeBefore
                launch {
                    taskDao.increaseMillisCompletedToday(selectedTask.id, elapsedTime)
                }
            }
        }
    }
}

Question --- Channel.consumeEach() (a suspend function) seems to be implemented internally as

for (e in this) action(e)

— which looks like it’s implicitly calling this.iterator() and iterating over the channel that way. I assume that can’t actually be what’s happening (right?) since Iterator.next() is blocking. Or does Channel.consumeEach() actually block?

hi, if I cancel a coroutine 

Job

 is there a way to find out where it was stuck? I can pass a 

CancellationException

 in, but what I really want is a traceback for where it was when it got cancelled.

This might be more of a Spring Webflux question, but here goes… Say I have this controller method

@PostMapping(path = ["/foo"])
  suspend fun example(): Flow<String> {
    val flux: Flux<String> = service.returnsFlux()
    return flux // <-- how to convert to Flow?
  }

(this is a trivial example, but I have an API that returns a

FLux<T>

) I’m trying to figure out how to go from a

Mono/Flux

back to the kotlin coroutines approach for Spring Webflux. I’d appreciate any help .

Hello I'm wandering what can the advantage of having such kind of structure ?

Hello, I’m trying to understand best practice for creating a CoroutineScope for some component I have. I don’t know what the difference is between these two approaches: CoroutineScope from Dispatcher:

class MyComponent {
  private val scope = CoroutineScope(<http://Dispatchers.IO|Dispatchers.IO>)
  
  fun close() {
    scope.cancel()
  }
}

CoroutineScope with Dispatcher + Job:

class MyComponent {
  private val job = Job()
  private val scope = CoroutineScope(<http://Dispatchers.IO|Dispatchers.IO> + job)

  fun close() {
    job.cancel()
  }
}

The second example is used e.g. in https://kenkyee.medium.com/android-kotlin-coroutine-best-practices-bc033fed62e7 I’m grateful for any insight! Thank you!

what are the rules for how coroutineContext is looked up? I’m finding it a little bit surprising to say the least. I’ve got a test (using mockk.io) structured roughly like this:

@Test fun `some particular invariant holds true`() = runBlockingTest {
    coEvery { someSuspendingFunction() } coAnswers { 
        val dispatcher = coroutineContext[CoroutineDispatcher]
        ....
    }
    ....
}

No matter which CoroutineDispatcher someSuspendingFunction() was called on,

dispatcher

in this code seems to end up being the test dispatcher created by runBlockingTest. If, however, I structure it like this:

private suspend fun mockImplementation() {
    val dispatcher = coroutineContext[CoroutineDispatcher]
}

@Test fun `some particular invariant holds true`() = runBlockingTest {
    coEvery { someSuspendingFunction() } coAnswers {
        mockImplementation()
    }
}

dispatcher

(in mockImplementation()) matches the dispatcher that someSuspendingFunction() was called on. IE… it looks an awful like the coAnswers block is running on that dispatcher, but

coroutineContext

specifically inside that block is reporting something different. Put differently --- coroutineContext seems to be returning the context from the enclosing lexical scope, rather than the context of the actual running coroutine. Is that expected?

What is BroadcastChannel getting replaced by?

I’m sorry to keep spamming this channel. I’ll try and keep it down after this. But: I’m having real trouble wrapping my head around the semantics of test dispatchers, specifically around how to test code that involves multiple dispatchers. From a test, I’m trying to call <some channel>.send(<some object>). This yields, which seems to cause dispatcher.advanceUntilIdle() to return (since the consumer is running on another dispatcher), which causes the test dispatcher to try to clean up and exit (even though the test isn’t finished), which causes it to throw IllegalStateException (as if the test had ended). I assume this is it functioning as it’s supposed to, but it leaves me at a little bit of a loss as to how to test this functionality.

Uncharted Teritory. While attempting to run my generated javascript code on react native, I am getting

process.nextTick is not a function. (In 'process.nextTick(this._get_messageQueue__1()._processQueue)', 'process.nextTick' is undefined)
at node_modules/@picortex/pi-monitor-client-sdk/pi-monitor-client-sdk.js:13160:17 in <anonymous>

Checking that line number, I see the following generated code

NodeDispatcher.prototype.scheduleQueueProcessing_1 = function () {
    process.nextTick(this._get_messageQueue__1()._processQueue);
  };

I think for some reasons, the library thinks it is running in a node environment. Is there a way I can make this work differently? Any work around?

I was getting

Unresolved reference: last

when referencing `

import kotlinx.coroutines.flow.last

from a unit test. Any ideas?

Hello there, I am bit confused I was trying to excute the Result once the main job completes but it never emit inside

invokeOnCompletion

even after I surround it within coroutine

liveData {
    val mainJob = viewModelScope.launch {
           emit(Result.LOADING)
           repo.add(data)
          
    }
    mainJob.join()
    mainJob.invokeOnCompletion {
        viewModelScope.launch {
        if(it != null) emit(Result.FAILED(it) else emit(Result.SUCCESS(null)
        } 
     }
}

Alternatively if I do

liveData {
   val mainJob = viewModelScope.launch {
            emit(Result.LOADING)
            repo.add(data)
            emit(Result.SUCCESS(null)
           
     }
     mainJob.join()
}

But I need to handle if it complete with in failure or success

Whats the best practice when you have a try catch block that catches

Throwable

inside a coroutine with CancellationException? I’m not relying on

CancellationException

anywhere in code, and I don’t want it to show up in my logs. Should I just be watching for it and then re-throwing it, or is it acceptable to just eat it and not throw it?

} catch (e: CancellationException) {
                throw e
            } catch (t: Throwable) {
                logger.w(RuntimeException("Error validating file", t))
                count++
            }