I am writing some extension functions to work with

Flow

and platform-specific API's, e.g. the web broadcastchannel API. However I choose to pass

CoroutineScope

when needed, but this leads to code like this:

states.broadcast(channel.deserialize(scope), scope)

Is it a code smell to have these kind of scopes everywhere? Or is it not a problem. My intuition says that it is not a problem, because this way I can easily control the scopes and no magic happens. So I can keep the structured concurrency.

so I have this block of code

lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        viewModel.uiState.collect { uiState ->
            if (uiState.authenticated) {
                setContent {
                    Home()
                }
            } else {
                setContent {
                    Authentication()
                }
            }
        }
    }
}

and the issue is that it’s collecting the stateFlow immediately before my viewModel gets the chance return the result, so it shows Authentication, then when the value comes back ti shows Home again I want it to…well observe like LiveData does

Hi, the docs of

Flow#shareIn

mention, that using

buffer(0).shareIn(scope, started, 0)

makes upstream suspend until all subscribers have processed the value. My question is whether that still holds true for

replay > 0

(for all currently active subscribers).

I've got a callback that can be called multiple times. How do I push the data passed to the callback into a

Channel

?

Hi everyone, what is the difference channel

receiveAsFlow

vs

consumeAsFlow

?

receiveAsFlow

can have multiple collectors and

consumeAsFlow

only 1?

I'm building a little helper on top of a callback routine via

suspendCoroutine

- but the code has some rather strict timing constraints, aka one code block should fully exit before another of the same kind should enter - how would I enforce that? Code looks like this:

suspend fun BluetoothReader.transmitEscapeCommandS(input: ByteArray): CommandAPDU {
    return suspendCoroutine { block ->
        if (this.transmitEscapeCommand(input)) {
            this.setOnEscapeResponseAvailableListener { _, response, errorCode ->
                if (errorCode == 0) {
                    block.resume(CommandAPDU(response))
                } else {
                    block.resumeWithException(ReaderCommandException("Error code from reader: $errorCode"))
                }
                this.setOnEscapeResponseAvailableListener(null)
            }
        } else {
            block.resumeWithException(ReaderCommandException("Couldn't send command. Bonded?"))
        }
    }
}

The listener can only really be set once, so there's a bit of a race condition there.

Should supervisorScope be inside launch or outside launch or doesn't it matter?

supervisorScope {
   launch(Dispatchers.Default) {
      supervisorScope {

Hello how can I transform a callback to a suspend fun?

interface Foo<T> {
  fun onSuccess(lis: (T?) -> Unit)
}

I'm trying to do something like this :

suspend fun getFoo(): Bar {
   suspendCoroutine<Bar?> { continuation -> 
     val listener = object : Foo<Bar?> {
        override fun onSuccess(lis: (Bar?) -> Unit) { }
     }
}

But I don't know how to return a Bar

I'm migrating to the new test coroutine API, and my tests that used to work using pause/resumeDispatcher aren't passing any more, eventhough I'm using

runCurrent

. So the problem, is that I have a ViewModel, that exposes a

StateFlow<MyState

, that's backed by

combine(flow1, flow2...

, I'm trying to assert the intermediate states that should be emitted, yet only the initial and the last state got emitted. Any idea what I'm doing wrong? (code in 🧵)

Is there an equivalent

trySendBlocking

to send exceptions down the channel?

Is there a

mapWithState

or similar for Channels, where I can keep a consistent state around?

My stackoverflow question didn't get any traction so I am hoping that someone will help here https://stackoverflow.com/questions/72843100/kotlin-flow-emit-loading-state-for-every-emission

Can anyone help me with this. I want to achieve something like this: fisrtApiCall -> secondApiCall() -> thirdApiCall() ->.. each api call if it fails then do the next one and if it succedd then finish with a function? How can I wrap this ideas with coroutines. My current code with rx single<> is quite messy.

What is a good way to schedule a repeating task in a platform agnostic way?

fun main() = runBlocking {
    val component = Component {
        var a = 0; var b = 0
        every(50.milliseconds) {
            if (it % 20 == 0) println("A: ${a++}")
        }
        every(1.seconds) {
            println("B: ${b++}")
        }
    }
    component.enable()
    delay(50.seconds)
    println("Done!")
}
context(Toggled) @Base
suspend fun every(period: Duration, block: suspend (Int) -> (Unit)) {
    simultaneously {
        var i = 0
        while (isActive) {
            block(i++)
            delay(period)
        }
    }
}

The problem with this is that over time the two tasks get way out of sync: A: 47 B: 49 After 50 seconds

This might be a little off-topic but I decided to give it a shot. I'm using a Kotlin Telegram bot framework from here. Now, the framework allows you to specify callback functions for certain actions. For example, if your Telegram bot receives a message, you can tell the bot to respond to the message. Here's the code:

fun main() = runBlocking {
    val bot = bot {
        token = "YOUR_API_KEY"
        dispatch {
            text {
                launch { println("launched") }
                bot.sendMessage(ChatId.fromId(message.chat.id), text = text)
            }
        }
    }
    bot.startPolling()
}

This code should set a callback for when the bot receives any text message from a user. However, the

launch { }

never gets launched. And I don't know why! I know all this Telegram bot stuff might be off-topic but it's related to coroutines and I can't comprehend why it wouldn't work. Can somebody tell me what I'm doing wrong?

Given a

StateFlow<T>

instance, I want to log every emission of

T

from the source standpoint. Is this possible without duplicating logs for each subscriber?

Shouldn't exceptions thrown in a suspend function bubble up like regular exceptions? Trying to catch a lower one, but it's not properly registered, just aborts the program ... err, works.

Hi! I have a question regarding flows. I want to use something like

.buffer(capacity = 0, onBufferOverflow = BufferOverflow.DROP_LATEST)

in my flow, but this doesn't work, since a capacity of

1

is used when using

BufferOverflow.DROP_LATEST

. What's the use case: The upstream flow items are produced whenever the user clicks a button to perform a network request, which happens downstream after the buffer. When the user presses the button while the network request is on-going, this event should be dropped (that's why I am using a buffer with capacity of 0). However I am not able to achieve this. A new network request is initiated shortly after the previous one is complete when the user clicks the button while the first network request is currently running. Any idea on how to implement this in a nice way?

i’m using MutableStateFlow to emit events but while collecting these events i’m adding

delay

for 500ms but due to this i’m missing all the other data and only last TempData is recieved

GlobalScope.launch() {
            flowMain
                .collect { data ->
                    Log.i("AppLogs", data.toString())
                    delay(1)
                }
        }

viewInputLvlArray.forEachIndexed { index, view ->
    flowMain.value = TempData(index, someotherValue)
}

Actually there are 15 items in array and i want to collect to be called 15 times but with specified delay

By default, do I use

default

or

io

dispatcher? Or rather, when would I want to use io over default? I get that its in the name, but what exactly is io work? Networking? Retrofit has coroutines support, so the thread wont be blocked there. Database? Room has it as well Only thing that comes to mind is file writing/reading? Or would it be bad to just use the

io

? I mean it contracts eventually, right? Can I see how many actual threads are in the

io

at some point?

Hi. there is something that I’d like to understand: Suppose we have this minimal piece of code:

suspend fun main() = withContext(MDCContext(mapOf("host" to "local"))) {
    val ktorClient = HttpClient(OkHttp)
    log("entering context")

    (1..10).forEach { i ->
        val newContextMap = MDC.getCopyOfContextMap()
        newContextMap["i"] = i.toString()
        withContext(MDCContext(newContextMap)) {
            log("before calling ktor client $i")
            ktorClient.get { url("<https://example.com/$i>") }.bodyAsText()
            log("after calling ktor client $i")
        }
    }

    log("exiting context")
    ktorClient.close()
}

when run, the MDC is shuffled totally indeterministic:

10:21:45.238 [main] - entering context {host=local}
10:21:45.240 [main] - before calling ktor client 1 {host=local, i=1}
10:21:45.824 [DefaultDispatcher-worker-1] - after calling ktor client 1 {}
10:21:45.824 [DefaultDispatcher-worker-1] - before calling ktor client 2 {host=local, i=2}
10:21:45.998 [DefaultDispatcher-worker-1] - after calling ktor client 2 {}
10:21:45.998 [DefaultDispatcher-worker-1] - before calling ktor client 3 {host=local, i=3}
10:21:46.102 [DefaultDispatcher-worker-1] - after calling ktor client 3 {host=local}
10:21:46.103 [DefaultDispatcher-worker-1] - before calling ktor client 4 {host=local, i=4}
10:21:46.208 [DefaultDispatcher-worker-3] - after calling ktor client 4 {}
...

But as soon as I change first line to

suspend fun main() = withContext(Dispatchers.Default + MDCContext(mapOf("host" to "local"))) {

it magically starts working as intended. (same on any other dispatcher, IO or Unconfined) Why is it so? Is it a problem with

suspend fun main

, ktor-client, or - most probably - me messing up something here?

📣 📣 📣 kotlinx.coroutines 1.6.4 is here! The highlights: • Module `kotlinx-coroutines-test`: added

TestScope.backgroundScope

for launching coroutines that perform work in the background and need to be cancelled at the end of the test. • Fixed the issue with 1.6.2 and 1.6.3 where the updated POM of

kotlinx-coroutines-debug

broke some builds.

I've got a

suspend fun

which produces

ByteArray

in 16 byte chunks. Reading them is kinda time-intensive, so I was wondering if there's a concept of a lazy channel of bytes, so I say "ready me X bytes" without it reading everything.

I have an algorithm that's very easy to run in parallel (think finding a best path), extremely CPU intensive, and would actually benefit from running only n branches at once for n CPU threads (once a solution is found, branches can be culled eagerly, so exploring a solution fully quicker is better than exploring more than n at once). Is there a dispatcher that would block new coroutines until some are finished, in such a way that only n would run for n CPU threads? Or is this too high level for dispatchers and I have to write my own orchestration logic?

Are there destructive primitives? I'm looking to write a parser.

first()

etc. doesn't advance though 😞 I'd also take a parser-combinator libraries if there's one around 😄

return channelFlow<TLV> {
        when(input.first()) {
            (0x00.toByte()) -> send(TLVNULL)
            (0x03.toByte()) -> {
                val length = parseLength(input)
                val bytes = input.take(length).toList()
                send(TLVNDEF(NdefMessage(bytes.toByteArray())))
            }
            (0xFD.toByte()) -> {
                val length = parseLength(input)
                val bytes = input.take(length)
                send(TLVProp(bytes.toList().toByteArray()))
            }
            (0xFD.toByte()) -> {
                send(TLVTerminator)
                close()
            }
        }
    }

I am testing viewModel using the new coroutines testing framework and Mockk library. It seems that the exceptions from child coroutines are swallowed though. It looks more or less like this example:

@Before
fun setUp() {
  Dispatchers.setMain(UnconfinedTestDispatcher())
}

@Test
fun `Some exemplary test`() = runTest {
    val dependency1 = mockk<Dependency1>()
    val viewModel = TestedViewModel(dependency1)
    viewModel.testedMethod()
    coVerify(exactly = 1) { dependency1.methodToBeCaled() }
}

The ViewModel itself is calling that method in viewModelScope

fun testedMethod() {
  viewModelScope.launch {
    methodToBeCalled()
  }
}

However I forgot to mock the called method. I am seeing in log

Exception in thread "Test worker @coroutine#7" io.mockk.MockKException: no answer found for: Dependency1(#1).methodToBeCalled(continuation {})

But the test passes. Is it known issue?

A few days I came across an issue with one of our services in a test environment, where we effectively had a deadlock in our coroutine setup. This took a while to track down as nothing showed in the jvm thread dump. Only after inserting lots of log statements and staring at the code was I able to track down the issue. Looking back on this now, it would have been very useful to be able to get some information on the coroutine state out of the running application. Is there any tooling available that would have helped in this instance? Is there something we could do to get thread dump like information of all active and suspended coroutines?

Does android studio not have the coroutines tab mentioned in https://kotlinlang.org/docs/debug-coroutines-with-idea.html#debug-coroutines ?