is there a kotlin pool library like commons-pool, maybe one that uses coroutines?

what would be a recommended way of registering a callback for the scope cancellation (or all children)? Would the empty

callbackFlow

with

awaitClose

work here?

fun CoroutineScope.onCancel(action: () -> Unit) {
        callbackFlow<Unit> { 
            awaitClose(action)
        }.launchIn(this)
    }

I need a simple way to send stateless notifications, currently I have:

val focusRequests = MutableSharedFlow<Unit>(onBufferOverflow = BufferOverflow.DROP_LATEST, extraBufferCapacity = 1)
focusRequests.tryEmit(Unit) // on frontend
focusRequests.collect { ... } // on backend

but I feel like this is an abuse of the SharedFlow. Is there more specific construct I should use in such cases?

Hi! How how would you go about creating an

actor

of buffer 1 that DROPS_OLD when the buffer is overflown. I’ve read the docs but I’m not sure I get it Is this it?

actor<Int>(capacity = Channel.CONFLATED)

I often wish there was a channel with input and output types, something like

Queue<In, Out>(process: suspend (In) -> Out, onBufferOverflow, capacity)

where I can send input and suspend until output is ready:

val out = queue.send(in)

, use

trySend

and so on. I'm curious if it's just me, do you ever find yourself needing something similar and what do you use?

What’s the equivalent of a UnicastSubject in Flow? SharedFlow with a buffer suspend might mimic emission behaviour however what about ensuring emissions aren’t missed if there are no collectors?

Could anyone please help me with understanding why

MutableStateFlow

would only emit last item in this example (running from Kotlin scratch file)?

val x = MutableStateFlow("Test")
CoroutineScope(<http://Dispatchers.IO|Dispatchers.IO>).launch {
    x.collect {
        println(it)
    }
}

x.value = "First"
x.value = "Second"
x.value = "Fourth"

println("Waiting...")
Thread.sleep(1000)

Result:

Waiting...
Fourth

Is there any way using the new test helpers to not skip delays during a test but rather controlling them using a

TestCoroutineScheduler

? I see

TestCoroutineScheduler

being used but delays are still being skipped which is super confusing to me as I don’t see the point of it if delays aren’t skipped

Hello my locationmanager is using a kotlin channel with this code

private val myLocationsChannel = Channel<MyLocation>(Channel.BUFFERED)
val myLocationsFlow = myLocationsChannel.receiveAsFlow()

How can i send an error to the channel? For example when i havent location permissions. Thanks

Hey, just wanted to run this by you guys and see what comes out.. I wanted a way to be able to define some ‘flags’ in the app - basically a bunch of compile-time defined, observable state holders. So, I came up with the following:

data class Flag<T : Any>(val key: String, val description: String, val defaultValue: T)

class FlagManager() {

    private val flagMap = mutableMapOf<Flag<out Any>, MutableStateFlow<Any>>()

    fun registerFlag(flag: Flag<Any>) {
        if (flagMap.containsKey(flag)) {
            throw IllegalStateException("Flag cannot be registered more than once")
        }
        flagMap[flag] = MutableStateFlow(flag.defaultValue)
    }

    fun <T : Any> getFlagState(flag: Flag<T>): StateFlow<T> {
        if (!flagMap.containsKey(flag)) {
            throw IllegalStateException("Flag not registered")
        }

        return (flagMap[flag] as MutableStateFlow<T>).asStateFlow()
    }

    fun <T : Any> updateFlagState(flag: Flag<T>, value: T) {
        if (!flagMap.containsKey(flag)) {
            throw IllegalStateException("Flag not registered")
        }

        (flagMap[flag] as MutableStateFlow<T>).value = value
    }
}

Im confused about the testing apis Am I expected to always inject the coroutine scope to stuff I want to test, even if I dont want to control time progress? Or is the correct way to make stuff synchronous/blocking with injecting Undispatched dispatcher driving my private scope? Asking for a friend who wants to test something like this

class Syncer {
   private val scope = CoroutineScope(SupervisorJob() + <http://Dispatcher.Io|Dispatcher.Io>)
   private val _state = MutableStateFlow<State>(Idle)
   val state: Flow<State> get() = _state
   
   fun sync() {
      scope.launch {
         _state.value = Syncing
         someSuspendingApiAndDbStuff()
         _state.value = Synced
      }
   }
}

I want to test if I see correct

state

emissions if I call

sync()

I'm relative new to Kotlin and I'm trying to test a use case which has a suspend function. The function calls an API, which I'm mocking using

mockK

.

class MyUseCase {
	operator suspend fun invoke(val someId: UUID) = withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
		val response = someAPI.requestDataById(someId)
		// process the response...
	}
}

Using JUnit5, there are a few exceptions I want to test, such as below:

@Test
fun testCustomExceptionIsThrown() {
	coEvery { someAPI.requestDataById(someId) returns someData }
	val thrown = assertThrows<CustomException> { 
		runBlocking { myUseCase(someId) }
	}
	assertEquals("customExceptionMessage", thrown.message)
}

This is always returning an

JobCancellationException: Parent job is Completed;

and I actually can't figure out why. Also, is this the right way to use those libraries?

do I need to cancel my coroutine scopes in tests if it seems the test process gets killed? or does it?

In general is it safe to assume the following are roughly similar in cost?

suspend fun getX() : X
fun getXFlow(): Flow<X>

val a = getX()
val b = getXFlow().first()

Obviously the implementations may be different (one off query vs query and subscribe). But can the implementation optimise the call to first() which is the when the subscription happens and short circuit it? Othewise it's tempting to put both variants on most repository methods. But it would be nice if there was an expectation that first() should optimise to the suspend fun case.

How would you test this?

class Syncer(private val db: Db) {
   private val scope = CoroutineScope(SupervisorJob + Dispatchers Io)
   
   fun sync() {
      scope.launch {
         ...
         db.insertSomething()
      }
   }
}

I'd like to assert that

db.readSomething()

reads what it should after I call

Syncer.sync()

There is nothing to suspend on to know when the sync actually finished (like some status flow) Is injecting

Dispatchers.Undispatched

instead of

Io

to make it synchronous, so I can assert right after

sync()

returns, idiomatic? Is there something better?

Hello all. We make extensive use of flows and will allow exceptions to be thrown into a flow to notify downstream consumers that a problem has occurred. We have recently been noticing some crash reports where the stacktrace indicates the location that the exception was constructed in the flow, but not where the collector is that received the exception. Typically we will handle exceptions by applying a catch {} operator to the flow near the collection site, but it seems in this case we have not added a catch and are having difficulty determining exactly where the collector is that initiated the exceptional flow. Does anyone have any advice about how to trace the site of the collector that received the exception?

Supposing I have a

Flow<T>

what would be the most idiomatic way to to the following:

someFlow.collect { v ->
  process(v)
  if (isVeryFirstItem) {
    someSharedFlow.emit("received first item")
  }
}

I.e. I want to do a side-effect only once after receiving the first emission. I could make

isVeryFirstFirstItem

some kind of a volatile or AtomicBoolean, but somehow I don't like this.

Why does the first block not print anything, but the second prints "launch launch"?

fun main() {
    val scope = CoroutineScope(Job())
    scope.launch { println("launch") }
}

fun main() {
    val scope = CoroutineScope(Job())
    scope.launch { println("launch") }
    scope.launch { println("launch") }
}

Am I right in thinking that this, which I totally didn't write, will permanently tie up one of the few Default threads, and that four of these running on a 4-core machine would prevent any coroutines from running on Default?

val someStateFlow: StateFlow<Unit> = ...
GlobalScope.launch(Dispatchers.Default) { someStateFlow.collect { ... } }

My coroutine education continues...any ideas on why this code prints nothing:

refreshScope = CoroutineScope(Job())
                refreshScope.launch(Dispatchers.Default) {
                    println("Clicked Refresh button pntln.")
                }

but this produces text?

refreshScope = CoroutineScope(Job())
                refreshScope.launch(<http://Dispatchers.IO|Dispatchers.IO>) {
                    println("Clicked Refresh button pntln.")
                }

The only difference is the Dispatcher (IO works, Default doesn't). The issue is somewhere else in my code, I expect, and there's far too much to include; both code snippets work in a vacuum. I'm just hoping for a direction to start looking.

Hi everyone ! I have a large number of flows and only want to collect up to N flows at once. Is there a way to apply some kind of LRU caching of flows and stop collecting them after being evicted from the cache ? I previously implemented that using `BroadcastChannel`s but since they are deprecated I have to find a solution with

SharedFlow

, which has no

close

method...

Hi, is is possible to uses coroutines to perform big operations like comparing every pixels colors of one image to multiple to find the closest image from all the other ones in a parallel way ? (Sort of multi-threading)

With the new dispatcher slicing

<http://Dispatchers.IO|Dispatchers.IO>.limitedParallelism(1)

does it guaranteed I'll get the same thread always? (Its for testing purposes, where I want to replace Main dispatcher with something single threaded)

How can I emit the result to the consumer from a different function. When I try the following nothing happens?

fun fetchInfo(): Flow<State<SportInfo>> =
    flow {
      val result = memoryInfoResult
      if (result != null) {
        emit(result)
      } else {
        fetchSportInfo()
        .onEach { result ->
          when (result) {
            is Data -> {
              memoryInfoResult = result
              emit(result)
            }
            is Error, is Loading -> {
              emit(result)
            }
          }
        }
      }
    }

I'm somewhat confused by 'advanceTimeBy' mechanics in kotlin tests using runBlockingTest Modifying a bit code present here, in the documentation: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-test/kotlinx.coroutines.test/-delay-controller/advance-time-by.html To something like that:

@Test
    fun advanceTimeTest() = runBlockingTest {
        foo()
        advanceTimeBy(2_000)  // advanceTimeBy(2_000) will progress through the first two delays
        println("Main 1")
        // virtual time is 2_000, next resume is at 2_001
        advanceTimeBy(2)      // progress through the last delay of 501 (note 500ms were already advanced)
        println("Main 2")
        // virtual time is 2_0002
    }

    fun CoroutineScope.foo() {
        launch {
            println("Foo 1")
            delay(1_000)    // advanceTimeBy(2_000) will progress through this delay (resume @ virtual time 1_000)
            // virtual time is 1_000
            println("Foo 2")
            delay(500)      // advanceTimeBy(2_000) will progress through this delay (resume @ virtual time 1_500)
            // virtual time is 1_500
            println("Foo 3")
            delay(501)      // advanceTimeBy(2_000) will not progress through this delay (resume @ virtual time 2_001)
            // virtual time is 2_001
        }
    }

The result is:

Foo 1

Foo 2

Foo 3

Main 1

Main 2

Shouldn't it be: Foo1, Foo2, Main1, Foo3, Main2 ?

on question not sure if you know the reason: so in android we use StateFlow and usually we use MutableStateFlow but the fragment only has the readonly so we created a function like this

val <T> MutableStateFlow<T>.readOnly
    get() = this as StateFlow<T>

now I realised there is a function called asStateFlow() but when you go into asStateFlow the function does something different

public fun <T> MutableStateFlow<T>.asStateFlow(): StateFlow<T> =
    ReadonlyStateFlow(this, null)

I'm seeing strange behavior with suspend functions that return

Result

. I can see that the body of the suspend function is returning what I expect. The

toString

return

Success(null)

or something like that. The caller of the suspend function is then getting an object with a toString of

Success(Success(null))

. When I return that from another suspend function it does not get wrapped a third time and still logs as

Success(Success(null))

. This is running on Android with Kotlin

1.5.31

and coroutines

1.5.2-native-mt

So my nightmare: • added a library that needs kotlinx-coroutines 1.4, was previously on 1.3. IntelliJ can no longer run the debugger because "Failed to find Premain-Class manifest attribute on kotlinx-coroutines-1.4.2" • "what the hell, I'm on java 13 with some half baked libs, time for an update", so I move to kotlin 1.5, kotlinx-coroutines 1.5, Java 17, | ◦ this blows up for many reasons. I work though most problems. ◦ im using zulu 17, a commuinty build of openJDK that bundles javafx as a jmod. IntelliJ's own highlighting understands this, and my ant build scripts understand it, but intelliJ's build button does not. what do I need to do to get intelliJ to build a java+kotlin project where javafx is from a jmod in the SDK? I've got an old cyclic dependency in my project that I'm sure isn't helping, so I've reverted all my work to sort that out first. any help is much appreciated, this is really really not fun.

I have a "service" which has a CoroutineScope inside and I want clients to be able to "dispose" this service, but I want Scope to be an implementation detail. But I'm not against returning some kind of other disposable object. I'd settle with

interface Service {
  fun start(): Job
}

val job = myServiceInstance.start()
job.cancel()

is it ok to implement start like

fun start() { val scope = createScope(); return scope.coroutineContext[Job] }

Or is there some better way?