Hello everyone! Just looking at problem where we need to boost performance for our coroutine code. Using visual vm and cpu sampling, I was able to figure out which synchronous calls were taking the longest, but it would be good to have some tools to figure out which suspending calls take the longest, does anybody have any suggestions or sites they can point to?

Is it allowed to use

runBlocking()

in my application if the

main()

function itself is

suspend

?

hello, which one is better to udapte StateFlow as a UI state private val _uiState: MutableStateFlow<State> = _MutableStateFlow_(initialState) #A

viewModelScope.launch {
     _uiState.update(newState)
 }

#B

_uiState.update(newState)

#C

_uiState.value = newState

hey guys ~ how to handle async error?? Is there a clean way?

Can I somehow transport a

Deferred

out form a callback? got this pattern:

kotlin
        val contract = ActivityResultContracts.StartActivityForResult()
        suspendCoroutine { cont ->
            registerForActivityResult(contract, contract.createIntent(this, client.signInIntent)) {
              <createing a Deferred here>
          }
        }

Can I somehow pull the

Deferred

directly into the

suspendCoroutine

, or do I have to do all the plumbing manually?

I use

runTest()

and I got this error:

Caused by: kotlinx.coroutines.TimeoutCancellationException: Timed out after 10m of _virtual_ (kotlinx.coroutines.test) time. To use the real time, wrap 'withTimeout' in 'withContext(Dispatchers.Default.limitedParallelism(1))'

I was wondering why

limitedParallelism(1)

is necessary, and how my test would behave without it? Thanks.

can I collect on a flow using main dispatcher? is this code correct or would it block the main thread?

CoroutineScope.launch(Dispatchers.Main){

flow.collect{}

}

👋 Need an advice about finding causes of this error crash in Android app:

kotlinx.coroutines.JobCancellationException: StandaloneCoroutine was cancelled

It has no stacktrace (due to https://github.com/Kotlin/kotlinx.coroutines/issues/1866). The only case that I was able to reproduce is when you throw this exception outside of coroutines. Any ideas?

I'm calling plain old blocking java code from a coroutine. Any guidance as to when to use

withContext(iodispatcher)

and when to use

runInterruptible(iodispatcher)?

Related: I call some java code that returns a stream. I convert it to a flow using

consumeAsFlow

how do I make sure that emission happens interruptibly?

What are differences of use between down below code? and What case should i use withContext not coroutine builder like async?

/-- no1
val job = async(dispatcher) { /**  do something */ }
/** do other processes */
val result = job.await()

/-- no2
val result = withContext(dispatcher) { /**  do something */ }

Is there a safe way to use

combine()

when you have a long list of Flows? I worry about the Flows and the param names becoming unsynced as the code is changed over time.

combine(
      stringFlow1,
      stringFlow2,
      intFlow1,
      intFlow2,
    ) { string2, string1, int2, int1 ->
     
    }

Hi, i am attempting to add Google Nearby Connections capability to my current android project. the issue i am facing is that although the initial connections work as expected i am having difficulty in implementing a repeatable process. This issue is caused by the status of the nearby connections becoming "out of step" with the information i have held in my applications classes. I believe a solution to this is issue is if i could transform the real time connection status of the nearby connection into a Kotlin (state) Flow and in that way i would have access to accurate connection details. i cannot see how to achieve this though. does anyone have some "pointers" on how i can Advertise, Discover and manage nearby connections using Kotlin Flows?

val flow = MutableSharedFlow<Int>()
val job = CoroutineScope(Job()).launch {
  flow.collect { println(it) }
}

suspend fun main() {
  yield()
  Thread.sleep(1000)
  flow.emit(1)
  yield()
  Thread.sleep(1000)
  job.cancelAndJoin()
}

why does collector get cancelled before receiving the emission in above example? In order to avoid race conditions, I put some yields and sleeps to make sure first collector appears before the emission, and also to make sure collector has time to collect after the emission. Still never happens It only happens with

delay(1)

after emission

Hi, I’m exploring the developments in Channels, Flow, SharedFlow, StateFlow, etc. Could anyone please guide me on whether channels are still required in Android, or if Flows are sufficient for all use-cases? Also, some blogs, videos, guides with updated content would be very useful to understand in depth. Thank you. gratitude thank you

Hi, we are exposing the same asynchronous API multiple times, once using coroutines, and once using RxJava. Because RxJava is a pain and adds complexity for our users, we are also thinking of exposing a synchronous/blocking API, for when performance is not key. There is of course a danger of users calling the new blocking API from within a coroutine. Is there any way we could check for this and warn our users?

I'm reading

Kotlin Coroutines deep dive

by

Marcin Moskala

. In the part of

understanding Kotlin Coroutine

of the book(specifically,

How does suspension work?

), I read this code below.

suspend fun main() {
  println("Before")
  suspendCoroutine<Unit> { continuation ->
  continuation.resume(Unit)
 }
 println("After")
}
// Before
// After

Notice that “After” in the example above is printed because we call

resume in suspendCoroutine. This statement is true, but I need to clarify. You might imagine

that here we suspend and immediately resume. This is a good intuition, but the truth is that there is an optimization that prevents a suspension if resuming is immediate.

Speaking of the

optimization

, in this case, Does it mean that "There is no suspension point in the Coroutine. So, there is no need to be suspended. That's why the compiler optimizes."?? For instances, 1)

suspend fun getUsers(): List<User> {
   // no asynchronous work and suspension point here
   // ....
  return users
}

in the code above, the IDE informs me that I don't need to use "suspend"(Redundant 'suspend' modifier). So, it also would be an optimazation. 2)

suspend fun getUsers(): List<User> {
  suspendCoroutine<Unit> { continuation -> // this code line is definitely a suspention point.
  // list sorting or something but not asynchronous work.
  continuation.resume(Unit)
 }
 return users
}

in the code above, it's almost same code of the example of the book. but i wonder how the compiler optimizes the code under the hood. and I wonder if there are any other optimizations for

susepnd

. i hope someone tells me about it.

I've got code which transforms an incoming flow and invokes actions based on basically a state machine. Currently the code does a few weird things, so I need to refactor it to be more predictable. Is there a

suspend fun Flow<R>.mapState(initial: T, fun: (T, R) -> (T, Flow<S>)): Flow<S>

somewhere?

Question I am not able to understand why SharedFlow has no effect of applying

flowOn

Hey all, I think I understand coroutines pretty well, but when I think deeper about them, maybe not so much. 😄 I have a question: what exactly happens when blocking code is called from within a suspending function, e.g.

suspend fun foo() =
  withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
    blockingCall()
}

I think this code is handled by a thread from the thread pool that is shared with the Default dispatchers pool, which means that if this suspending function was called from another thread, that thread can now be used for something else, correct? But if this code is reached from another suspending function already executing on that Dispatcher or the default dispatcher, I guess it will try to not switch to another thread? And the thread in the pool that is performing the blocking code is blocked I think. If there are by default 64 threads in that pool, does that mean that for instance in a microservice with a REST controller with suspending functions, only 64 requests can be handled in parallel?

Is there a coroutine function like

singleOrNull()/firstOrNull()

that does not suspend until first item is emitted? My use case:

suspend fun execute(dispatcher: Coroutines.Dispatcher) {
    withContext(dispatcher) {
        // do something that returns result
        // also process outstanding tasks from SharedFlow<() -> Unit>
        outStandingTask.state.singleOrNull()?.invoke() // suspends, so no result is returned
        return result
    }
}

Would a simple Queue fit better?

any documentain for api call efficient with coroutines

I'm trying to understand what coroutines-test virtual time is all about ( https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-test/ ). I tried using withContext(Dispatchers.Default.limitedParallelism(1)) - but when something happens inside the main context, it seems like that doesn't take effect. I honestly find this virtual time thing really unhelpful when it comes to testing real code. It breaks a lot normal coroutine use cases out of the box. Issue is discussed here on Github, but it seems like no change/plan for change as of yet: https://github.com/Kotlin/kotlinx.coroutines/issues/3179

Is EventBus pattern good choice for broadcast event’s in application? Im developing simple event worker application that subscribes Kinesis or Kafka. And should broadcast all events to consumers which interested in some type of Event. Plz review for my code snippets!

interface EventDispatcher {
    suspend fun start()
}

interface EventBus<T> {
    val events: SharedFlow<T>
}

interface MutableEventBus<T> : EventBus<T> {
    suspend fun emit(event: T)
}

interface EventConsumer : CoroutineScope {
    suspend fun consume()
}


@Component
class LifecycleManager(
    private val dispatcher: EventDispatcher,
    private val consumers: Set<EventConsumer>,
) {
    private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())

    @PostConstruct
    fun start() {
        consumers.forEach { consumer ->
            scope.launch {
                consumer.consume()
            }
        }
        scope.launch { dispatcher.start() }
    }

    @PreDestroy
    fun close() = runBlocking {
        consumers
            .map { consumer -> scope.async { consumer.cancel() } }
            .joinAll()
        scope.cancel()
    }
}

It’s my applications’ event driven structure. •

EventDispatcher

implementation should subscribe event stream service like AWS Kinesis or Kafka. And EventDispatcher dispatch event by call emit method of MutableEventBus. •

EventBus

implementation should broadcast T type event’s. And it’s provide only readonly events. •

MutableEventBus

implementation is extension of EventBus that provide emit(produce) event method. •

EventConsumer

implementation should consume events from EventBus. •

LifecycleManager

is Spring bean that managing whole lifecycle of event components.

Is it safe to catch all exception when calling a suspend function? For more specific, will it break the propagation of CancellationException to cancel other job?

What is the difference between channel.fori and channel.consumeEach ?

Hey 👋 I want your opinion on something I'm trying to do. There's a

SharedFlow

where I emit all the incoming WebSocket-like messages where • some incoming messages are "commands" I'm supposed to execute • some incoming messages are "responses" to the requests my app makes in order to get more details about said "commands" • (unfortunately, the details can't be transmitted along with the "commands" and must be explicitly requested by the app) A single collector of said

SharedFlow

determines what action should be taken once a "command" arrives. Processing a "command" may involve sending a request (for the details) and awaiting another answer using another ad-hoc spawned collector from the same

SharedFlow

. In short, it looks more or less like this (pseudo-code)

class SomeClassThatHasALifecycleBoundScope {
    fun onStarted() {
        lifecycleScope.launch {
            commandExecutor.collectCommands()
        }
    }
}

interface WebSocketClient {
    // replay = 0, buffer = 0, onBufferOverflow = SUSPEND, commands emitted using emit()
    val commands: SharedFlow<Command>
    
    suspend fun sendRequestForCommandDetails(commandId: String, clientToken: String)
}

class CommandExecutor {
    suspend fun collectCommands() = withContext(observingDispatcher) {
        webSocketClient.commands.collect {
            processCommand(it)
        }
    }
    
    suspend fun processCommand(command: Command) = coroutineScope {
        val clientToken = UUID.randomUUID().toString()
        
        val detailsJob = async {
            webSocketClient.commands
                .timeout(10.seconds) 
                .first { it.clientToken == clientToken }
        }
        
        webSocketClient.sendRequestForCommandDetails(command.id, clientToken)
        
        val commandDetails = detailsJob.await()
        
        doSomethingWithCommandDetails(commandDetails)
    }
}

Obviously, the problem above is that

detailsJob

can't receive an answer from

webSocketClient.commands

until the collector from

collectCommands()

unblocks the

SharedFlow

(

emit()

is suspended) So I'm thinking about processing the commands in dedicated coroutines to unblock the

SharedFlow

. The question is: is it OK from the structured concurrency point of view to

launch

in

collect

e.g. like this?

suspend fun collectCommands() = withContext(observingDispatcher) {
    webSocketClient.commands.collect {
        launch {
            try {
                processCommand(it)
            catch (e: Exception) { ... }
        }
    }
}

Is the new coroutine a child (or "grandchild") of

SomeClassThatHasALifecycleBoundScope.lifecycleScope

? Will it be cancelled properly when

lifecycleScope

gets cancelled?

Hey, I am trying to implement a search functionality that will hit the server endpoint. I am using

SharedFlow

,

collectLatest

and

debounce

so the endpoint won't be hit every time there is a text change. However,

collectLatest

is being called many times. why? and how can I fix it? full code

searchText
                .debounce(1000)
                .distinctUntilChanged { prev, next -> prev != next }
                .filter { it.isNotEmpty() }
                .sample(500)
                .collectLatest {
                    Log.d("Remon", "search collect - value: $it")
                }

Is a

MutableSharedFlow(0, 1, DROP_OLDEST)

guaranteed to immediately deliver events (without dropping any event ever) when emitting with a single single/threaded emitter (

tryEmit

) to collectors collecting with a

DirectDispatcher

if collectors with direct dispatcher don’t suspend? Even if there are other collectors with other dispatchers that can suspend?

object DirectDispatcher : CoroutineDispatcher() {
  override fun dispatch(…) {
    block.run()
  }
}

Not sure if you saw it https://kt.academy/workshop/coroutines?utm_source=twitter&amp;utm_medium=promotion&amp;utm_campaign=weekly