Hello 👋 Playing around with

Channel

based pipelines and was wondering whether you have any suggestions to make it better. NOTE: for this exercise want to stick to channels only Consider following basic example - fetch list of items and for each one of them fetch additional information (details + reviews) that will be used to final result

val result = mutableListOf()

val summaries = retrieveSummaries()
for (summary in summaries) {
   val details = retrieveDetails(summary.id)
   val reviews = retrieveReviews(summary.id)

   result.add(buildResult(summary, details, reviews)
}

return result

I could easily parallelize the above by doing something like

summaries.map { summary -> 
  async {
    val details = async { retrieveDetails(summary.id) }
    val reviews = async { retrieveReviews(summary.id) }
    
    buildResult(summary, details.await(), reviews.await())
  }
}

Currently trying to figure out what would be the "best" way to achieve the same using

Channel

. "Simplest" would be to do a sequence of channels

val summaries: ReceiveChannel<Summary> = summariesChannel()
val detailsAndSummaries: ReceiveChannel<Pair<Summary, Detail>> = detailsAndSummariesChannel(summaries)
val result: ReceiveChannel<Result> = resultChannel(detailsAndSummaries)

Looks like

BroadcastChannel

is marked obsolete but I guess logically it would make sense to broadcast summaries and then send both details and reviews from separate channels but I am unsure how to read from multiple channels at once (i.e.

buildResult

would need results from 3 channels). Any ideas/suggestions?

Hello, I had to use coroutines reactive but I ran in a strange behavior while testing. This is a minimal not working example

class CoroutinesReactiveTest {

    private val dep = mock<Dep>()
    private val underTest = UnderTest(dep)

    @Test
    fun `Test someOtherMethod called`() = runBlocking {
        underTest.methodToTest().test()

        verify(dep).someMethod() // Commenting this line test passes
        verify(dep).suspendingFun()
        verify(dep).someOtherMethod()
    }
}

class Dep {

    fun someMethod() {
        println("someMethod called")
    }

    suspend fun suspendingFun() {
        delay(100)
        println("suspendingFun called")
    }

    fun someOtherMethod() {
        println("someOtherMethod called")
    }
}

class UnderTest(private val dep: Dep) {

    fun methodToTest(): Completable {
        return Completable.fromCallable { dep.someMethod() }
            .andThen(rxCompletable { dep.suspendingFun() })
            .andThen(Completable.fromCallable { dep.someOtherMethod() })
    }
}

commenting the first verify test passes

When using MutableLiveData works

If I catch a

CancellationException

should I rethrow it?

Is it possible to

filterNotNull

a

MutableStateFlow

and keep returning a

StateFlow

instead of a

Flow

? It is forcing me to add an

stateIn

to retransform the

Flow

into

StateFlow

when using

MutableSharedFlow

is there any consequence for choosing a buffer of

Int.MAX_VALUE

? i.e. does it allocate an array of that size or something? often i would like some buffer to exist but don't have a definitive size in mind so i default to max value rather than something arbitrary between

0 .. MAX_VALUE

I call this function when app startup and stuck on loading screen . Here is the log console when debug:

Log: Before delay
Log: Init Status: true
Log: Job Status: true

getUserInfo()

will not be called at all ??? What’s happen with flow ?

interface InitRepository {
    val initFinish: Flow<Boolean>
}

suspend fun startApp() {

    checkStatus()
    waitInitializeDone()  -> App is stuck on waiting here... Forever waiting
    getUserInfo()
}


suspend fun waitInitializeDone() {
    withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
                val waitJob = launch {
                    Log.d("Before delay")
                    delay(10000L)
                    throw TimeoutException("Init process timeout")
                }
                initRepository.initFinish
                    .onEach { initFinish ->
                        Log.d("Init Status: $initFinish")
                        Log.d("Job Status: ${waitJob.isActive}")
                        if (initFinish && waitJob.isActive) {
                            waitJob.cancel()
                        }
                    }
                    .launchIn(this)
            }
}

why does idea complain about a blocking call in this snippet?

return@coroutineScope withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
                    Files.readAllBytes(path)
                }

Hi everyone, I am doing some work with OpenGL/SurfaceTexture on Android and am considering having a dedicated thread for these operations. I have a reference implementation (written in Java) where this is the approach. Is there a more Kotlin-idiomatic way of doing this? Some alternatives I should consider? I appreciate your thoughts, thanks.

Hi everyone, how do you use coroutines to tackle with the SingleLiveEvent I've tried the code below but when rotating the device the latest value is displayed, i want the value to be consumed only once just as SingleLiveEvent is

protected val loadingState = BroadcastChannel<Boolean>(Channel.BUFFERED)
    val actionReceiver = loadingState.asFlow()

Is there an operator to go from

List<Flow<A>>

to

Flow<List<A>>

? I’m currently applying the following:

.flatMapLatest { items: List<Flow<A>> ->
  combineTransform(items) {
    emit(it.toList())
  }
}

but am wondering if this is common enough and/or is handled in a better way with a different operator.

Is there a convenience to make a suspend function take atleast X ms in total? Or should I manually time it and add delay

Hi, I'm working with compose-desktop and coroutines. I am updating app state through a StateFlow and am a little confused about what thread the StateFlow needs to be updated on. For example, a user enters some text, a coroutine is launched to save the text to the database, then a new state object is created with the new text, the state object is then added to the StateFlow, the composable listening to the StateFlow is updated. The last part, when the state object is added to StateFlow, does that need to happen on Dispatchers.UI in stead of Dispatchers.Main because the UI will be updated?

This might be slightly offtopic, but how do you unit test concurrent code? If I just let loose two threads at a function, im not guaranteed contention will happen, right?

So I was doing some tests with newSingleThreadContext and withContext and I was wondering, does withContext block the parent coroutineScope?

import kotlinx.coroutines.*
fun main() = runBlocking {
    launch {
        delay(100)
        println("@ launch")
    }
    withContext(newSingleThreadContext("new Context")) {
        delay(200)
        println("@ withContext")
    }
    println("@ runBlocking")
}
// Outputs
@ launch
@ withContext
@ runBlocking

but

import kotlinx.coroutines.*

fun main() = runBlocking {
    launch {
        delay(100)
        println("@ launch")
    }
    launch(newSingleThreadContext("new Context")) {
        delay(200)
        println("@ withContext")
    }
    println("@ runBlocking")
}
// Outputs
@ runBlocking
@ launch
@ withContext

could someone please explain this?

If some code does (de)serialization or other potentially CPU-intensive work like using a template system to build HTML (all in-memory), do you do delegate the execution to the

Default

dispatcher? Or do you you accept that the current dispatcher (whatever it is) is potentially blocked for a few milliseconds? I have the habit of wrapping such code in

withContext(Dispatchers.Default) { … }

, but I’m not actually sure if that’s necessary. Or maybe it’s even a bad idea.

If I have a Cancellable implementation - can I somehow listen for the job cancellation and invoke cancel on Cancellable?

Hi 👋 I have a question regarding multiple collectors for Flow : how do you properly implement it? I tried to transform a Flow to a SharedFlow, to be able to do so. But it's no working, only the first collector called is doing his job:

//Inside UseCase
suspend fun state(): SharedFlow<HomeState> = locationInteractor
    .canGetLocations
    .flatMapLatest { ... }
    .flowOn(ioDispatcher)
    .shareIn(
        scope = CoroutineScope(Job() + ioDispatcher),
        started = SharingStarted.Eagerly,
        replay = 1,
    )

//Inside ViewModel
suspend fun success(): Flow<List<HikeCard>> = homeInteractor
    .state()
    .filterIsInstance<HomeInteractor.HomeState.Success>()
    .map { ... }

suspend fun error(): Flow<ErrorMessages> = homeInteractor
    .state()
    .filterIsInstance<HomeInteractor.HomeState.Error>()
    .map { ... }

//Inside View
outputs.success().collect { ... }  // the only one collected because of order
outputs.error().collect { ... } // never collected, except if moved above previous collector

Am I missing / misunderstanding something? Thanks for your help!

Can I set the min interval for the downstream to consume? (using flow or channel) for example:

I am reading basics of coroutines from docs, I have a doubt like :

launch {
   doingNetworkCall()  // this is suspend function
}
// doing some work on main thread here like loading

so assume doingNetworkCall() is taking some time, till that time that function get suspended OR it just works in background thread and on completion it returns like callback and do other work and if this work is getting suspended, then when it is getting resumed to achieve desired output

The docs on

MutableSharedFlow

are unclear, and I haven’t had enough coffee this morning to figure it out from looking at the code: if there’s no buffer and no replay, is there any difference between

DROP_OLDEST

and

DROP_LATEST

? I don’t think there should be, since if there’s no buffer, and no subscriber waiting for a value at the time of emission, the only thing to do is drop the value and there’s no concept of the “oldest” or “latest” value, just “the value trying to be emitted”.

When I run the following code, the

Job from coroutine scope

is printed first, not

Job1

, why is that?

fun main() = runBlocking<Unit> {
    launch {
        println("Job 1")
    }

    launch {
        println("Job 2")
    }

    launch {
        println("Job 3")
    }

    coroutineScope {
        launch {
            println("Job from launch inside coroutine scope")
        }

        println("Job from coroutine scope ")
    }
}

Hello. What is idiomatic way to merge two flows in precise positions. Assume following code:

val flowA = (1..10).asFlow()
val flowB = flowOf("a","b", "c", "d", "e")

Result I want to get is:

1, 2, a, 3, 4, b, 5, 6, c, 7, 8, d, 9, 10, e

My current solution is to use ReceiveChannel, but it does not look very idiomatic:

launch {
  val channelB = flowB.produceIn(this)
  flowA.withIndex().transform { item -> 
    emit(item)
    if (item.index > 0 && item.index % 2 == 1)
      emit(flowB.receive())
  }.collect { /* something useful */ }
}

Hi Need some help in understand some basic. I am holding a list of API requests I would like to execute in parallel. I need to wait for all the responses, each response is a list and combine all of them into one list. I still struggle to change the way I think with parallel Vs async - I ll be glad if someone can point me to a similar example in kotlin when async is used to fork and join a collection. thanks

According to the Dispatchers.IO doc:

... using withContext(Dispatchers.IO) { ... } does not lead to an actual switching to another thread — typically execution continues in the same thread.

So how does it do not to blocking thread but using same thread? When does it really switch another thread? (or never?) --- Edit: The key is

Thread pool

?

This message was deleted.

I have

private val testCoroutineDispatcher = TestCoroutineDispatcher()
  private val testCoroutineScope = TestCoroutineScope(testCoroutineDispatcher)

I try to run suspend function in it

fun runBlockingTest(block: suspend TestCoroutineScope.() -> Unit) =
    testCoroutineScope.runBlockingTest { block() }

And particular is

dispatcherRule.runBlockingTest {
      val msg = Gallery.Effects.LoadGallery.run.invoke(this, Gallery.Deps(photoGalleryManager))
      assertThat(msg).isEqualTo(Gallery.Msg.FilesLoaded(failure))
    }

I’m getting

This job is not completed yet

Even if I change it to:

private val testCoroutineScope = TestCoroutineScope(testCoroutineDispatcher + Job())

I still get that exception, what can I do? If I run test inside the global scope then test pass

Is there a common pattern for starting a coroutine, when you only ever want one or zero instances of it running? Something like

class MyActivity {
    var coroJob : Job? = null
    val coroScope = MainScope()
  
    // Start my coroutine, may be called from any thread
    fun startCoro () {
        // coroJob should only be accessed from main thread
        coroutineScope.launch (Dispatchers.Main) {
            if (coroJob == null) {
                coroJob = coroutineScope.launch (<http://Dispatchers.IO|Dispatchers.IO>) {
                    // Do the actual work here ...

                    withContext(Dispatchers.Main) {
                        coroJob = null
                    }
                }
            }
        }
    }  
}

But reducing the boiler plate?

Another possibility might be

val coroJob = AtomicReference<Job> ()
    /// Start my coroutine, may be called from any thread
    fun startCoro () {

        val coro = coroutineScope.launch(<http://Dispatchers.IO|Dispatchers.IO>, start = LAZY) {
            // Do the actual work here ...
            coroJob.set(null)
        }

        // If coroJob isn't null, do nothing.
        if (coroJob.compareAndSet(null, coro))
            coro.start()
    }

which avoids the explicit context switches and should still be thread-safe. Is there a more idiomatic and safe way to do that?