I have a question... situation not fully clear, but the observation is this: I invoke 3 "nodes" which each start 2 launch coroutine builders.... While each node is starting I do Thread.sleep(5000) on may thread (I run the stuff on main function) But what I observe is that 3rd node does not start until first one is destroyed (tus freeing some threads) As I told each node has 2 launchers where one is infinite cycle waiting on server connection (nonbloicking) incomming messages, but 2nd launch does .receive on channel So my version is that first 2 nodes somehow exhaust ForkJoin pool and my 3rd node does not start until I free something... can it be?

Something like that?

val channel = produce {
  delay(500) // only one value with delay, but can be wrapped to some loop)
  send("some value")
}
for (value in channel) {
   //Do something
}

Depends on your case of course. What exactly do you want to do?

the

CompletableDeferred

was completely internal, I was using

return deferred.await()

and

object.onCallback {deferred.complete(it)}

Could someone explain the concrete difference between a CoroutineDispatcher and a CoroutineInterceptor please?

Hello, can someone help me with kotlin channels? I wanted to create a behavior similar to

ConflatedBroadcastChannel

, but discarding the event when it's broadcasted. Is there any channel implementation that is capable of that?

Hey all ,Just need an advice here using Coroutines in kotlin : Should we call async methods in properties of class, and await in class methods whenever they would be called For Ex -

class DataAccess(val id:String) {
    
    private val fooAsync = async { fetchDataFromTableA(id) }
    private val barAsync = async { fetchDataFromTableB(id) }

    fun getData(): Data {
        var foo =fooAsync.await()
        var bar = barAsync.await()
        
        return compute(foo,bar)
    }
}

Just want to know ,Is it right way to do it ?

is there an example anywhere showcasing the difference between

suspendAtomicCancellableCoroutine

and

suspendCancellableCoroutine

?

Hey, what is the recommended way to convert a

ReceiveChannel

to a shared

BroadcastChannel

? I try to convert my Reactor SSE example to Coroutines where I want to use a single stream from my Mongo database to a shared broadcasted stream to browsers. Reactor version: https://github.com/sdeleuze/spring-kotlin-deepdive/blob/step3/src/main/kotlin/io/spring/deepdive/web/ArticleController.kt#L51-L52. Coroutines version without the broadcast feature: https://github.com/sdeleuze/spring-kotlin-deepdive/blob/step3-coroutine/src/main/kotlin/io/spring/deepdive/web/ArticleController.kt#L52-L57.

Anyone has experince moving from rxjava to coroutines? Any tips? Do I need that or not? Our app is written clean arch style, with repos, interactors presenters and rx binding it all together.

Multithreading is a lot easier on plain Coroutines IMHO

Okay, I might have overlooked the “channels” thing, it looks pretty promising and resolving 90% of tasks inside our android app. Thank you for rx vs coroutines opinions, I should research more

Is there a way to launch a coroutine that 'listens' to stdin input while the rest of the program runs? something like

launch { while(true) { //listen to keyboard input and do stuff } }

yeah, I’d want multiple subscribers 😕

Hi, does anyone knows if kotlinx.coroutines on #kotlin-native are planned anytime soon now that multiplatform support is there? See also related issue: https://github.com/Kotlin/kotlinx.coroutines/issues/246

asyncHandler(this::ping)

Will complain that KFunction<RoutingContext, Unit> doesnt match suspend (RoutingContext) -> Unit

Questions on

Mutex.kt

internal class MutexImpl(locked: Boolean) : Mutex, SelectClause2<Any?, Mutex> {
    // State is: Empty | LockedQueue | OpDescriptor
    // shared objects while we have no waiters
    private val _state = atomic<Any?>(if (locked) EmptyLocked else EmptyUnlocked)

    // resumeNext is: RESUME_QUIESCENT | RESUME_ACTIVE | ResumeReq
    private val _resumeNext = atomic<Any>(RESUME_QUIESCENT)

    public override val isLocked: Boolean get() {
        _state.loop { state ->
            when (state) {
                is Empty -> return state.locked !== UNLOCKED
                is LockedQueue -> return true
                is OpDescriptor -> state.perform(this) // help
                else -> error("Illegal state $state")
            }
        }
    }

1. Please, explain, what is meant by each of

Empty | LockedQueue | OpDescriptor

? 2. What is for

resumeNext is: RESUME_QUIESCENT | RESUME_ACTIVE | ResumeReq

? 3. What means the `help`comment in

is OpDescriptor -> state.perform(this) // help

? 😅

Does anyone have any suggestions on how to create a coroutine that tracks the completion of other jobs before launching a final coroutine? ie, I have two jobs that are running: Job A, Job B I have a third job, that must wait for A and B to finish: Job C

/ -- [ Job A ] -- \
----                  ---- [ Job C ]
    \ -- [ Job B ] -- /

i'm using a ConflatedChannel for a view model so that it can be updated while there's no view to consume but when the view comes back it gets the latest. This works great, except it appears that the incoming view always has to suspend to get that buffered instance. This means that the view state gets restored before the data (which then effectively drops the state). Is there a way to have a consumer of a conflated channel receive the last known value (if any) synchronously inside an unconfined coroutine? Or is there something else built-in that I can use for this?

@elizarov were you able to check that coroutine bug where it's throwing uncaught exceptions outside the RX chain?

@elizarov Just out of curiosity - how does one go about adding support for 'suspend' functions to their own interface of their library - dunno if the question makes sense --- https://twitter.com/JakeWharton/status/964556938322341894

Ignore above

You can also have a look at ConflatedBroadcastChannel: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-conflated-broadcast-channel/index.html

Hi, where can I find the suspension point icon that we can see in the IDE (in vector format or high res png)?

has anyone run into needing to debounce a channel yet?

Is there, or are there any plans to do functional operators for channels? (map, filter, etc)

@elizarov oh ok, I didn't know the behavior of

withContext

was different, is that difference documented?

That just burns CPU while it waits. Are you sure you want that?

Is there a way to set a generic upper bound to

suspend Function<R>

? Exactly

Function

because I don't know yet how many arguments it will accept. Use-case: hierarchical web routing.

interface Address<T, HANDLER : suspend Function1<Unit>>

class Address1<T> : Address<T, suspend (T) -> Unit>

handles

/{a}/

addresses, fetches an object with

a

path, supplies fetched object to handler

class Address2<T, U> : Address<U, suspend (T, U) -> Unit>

handles

/{a}/{b}/

addresses, fetches

T

bound to

a

path, fetches its child

U

bound to

b

path, supplies them to handler and so on...

Hi, is there an empty ReceiveChannel available in kotlinx.coroutines ?