📣 📣 📣 kotlinx.coroutines 1.6.0-RC3 is here! Mostly to fix the

Flow.collect

broken in RC2. Full changelog: https://github.com/Kotlin/kotlinx.coroutines/releases/tag/1.6.0-RC3

Hi, I have the following flow that im collecting to control my android activity loading state...

When fetching a very large contact list from an Android device (4000+) is is beneficial to run mutliple coroutines concurrently with the

async { }

builder? Or does the performance large depends on the I/O speed of the UFS storage?

var resumePoint: Continuation<String>? = null
var hasJumpedBack = false
suspend fun test() = suspendCoroutineUninterceptedOrReturn<String> {
    resumePoint = it
    "Hello "
}

fun jumpBack() {
    if (!hasJumpedBack) {
        hasJumpedBack = true
        resumePoint?.resume("World")
    }
}
runBlocking {
    println("Starting")
    val result = test()
    println(result)
    jumpBack()
}

It had to be tested

Is it possible to implement repeatable task (with interval) using coroutines?

// Run job that prints "ABC" string every 1000ms
val repeatJob = repeatJob(1000) { println("ABC") }

//until job is canceled
repeatJob?.cancel()

Are there nice patterns in Kotlin Coroutines/Flows for having a state machine based on the inflight requests? On each request start/completion - have it evaluate whether to change some app state.

Is is possible detect when the scope is canceled in a flow

.onCompletion

? I have this flow :

flow {
  for(foo in bar downTo 0){
    emit(foo)
    delay(1_000L)
  }
}

Then I consume it :

myFlow()
 .onEach {
    updateUI
  }
 .onCompletion {
   //Do something but only when the timer is 0 not when cancelled
 }
 .launchIn(scope)

In some point I do cancel the scope and it automatically get into `onCompletion` so even if the timer is not over yet, it does the thing of timer finished because I have the code there, is there any way I can get if the completion is because ok the for loop or for a cancellation of scope?

Could someone explained what the Key in the CoroutineContextHandler is used for? I can’t figure it out from the documentation

Hi all, when using workmanager, how could we listen when worker is stopped within the worker ?

I am launching a coroutine like

scope.launch{ // doing some work here which is having loops }

I want to stop doing that work on button click so I tried doing like this:

job = scope.launch{ // doing some work here which is having loops }

and then canceling the job :

job?.cancel

but it seems like loops are executing till termination and the job is not getting canceled Any other workaround?

📣 📣 📣 kotlinx.coroutines 1.6.0 is here! • Brand new

kotlinx-coroutines-test

module with reworked API and multiplatform support •

CoroutineDispatcher.limitedParallelism

and elastic

<http://Dispathers.IO|Dispathers.IO>

• Out-of-the-box support of new K/N memory model •

CopyableThreadContextElement

for mutable context elements • And a lot of improvements and bug fixes: https://github.com/Kotlin/kotlinx.coroutines/releases/tag/1.6.0 We also have a blogpost and video with additional details, examples and explanation

Hi all! Does anybody know if synchronized works in that way with coroutines? Could it occur any problems when I get data from map in UI thread? suspend fun loadData() { val data = getDataFromDB() data?.let { updateData(it) } } @Synchronized private fun updateData(data: Map<String, Data>) { map.clear() map.putAll(data) } @Synchronized fun getData(key: String): Data? { return map[key] }

I'm having troubles understanding the exception handling in the flow. Could someone explain why the throwable completes the flow even with the 

catch { }

 in the chain? If I'd want to have a infinite flow I must handle the exceptions with 

try catch

 in the lambdas because flow will close after any exception "leaking" to the stream, right? Or is there a way to do some explicit recovery from the exception? Example code: only 

0

 will be emitted, next emit 

1

 is never executed due to flow completion after exception.

flow {
    emit(0)
    emit(1) // never executed
}
    .onEach { if (it == 0) throw Throwable() }
    .catch { println(it) } // catch and ignore
    .collect()

I have a StateFlow<SyncState> which is written to by coroutines running on Dispatchers.IO I collect this in a ViewModel to drive UI, so the scope is driven by Dispatchers.Main I also collect this in some other place, where the scope is driven by IO Sometimes the ui driven collector doesnt receive emits (and the io collector does, so the StateFlow does actually gets written to correctly), randomly, like 1 in 10-15 I suspect this is when the writes are too close together in time? Is there a legitimate reason for all emits to be dropped in Main threaded collector due to some backpressure or something? Shouldnt I atleast see the last emit? (And yes the scope is still active, I checked) Bug?

The new limited parallelism dispatcher, does that actually limit parallelism on coroutines or is it only working in the case where you have blocking code?

hey anyone have any idea how to resolve this . Seeing this when upgrading to Kotlin Coroutines 1.6.0 . i am currently using kotlin version1.6.10

No matching variant of org.jetbrains.kotlinx:kotlinx-coroutines-core:1.6.0 was found. The consumer was configured to find an API of a platform, preferably optimized for Android, as well as attribute 'com.android.build.api.attributes.BuildTypeAttr' with value 'debug', attribute 'store' with value 'googleplay', attribute 'org.jetbrains.kotlin.platform.type' with value 'androidJvm' but:
          - Variant 'commonMainMetadataElements' capability org.jetbrains.kotlinx:kotlinx-coroutines-core:1.6.0:
              - Incompatible because this component declares a usage of 'kotlin-api' of a library, as well as attribute 'org.jetbrains.kotlin.platform.type' with value 'common' and the consumer needed an API of a platform, as well as attribute 'org.jetbrains.kotlin.platform.type' with value 'androidJvm'

Where is the best write-up on the

offer(..)

and

trySend(..)

change?

Hi there. I have a question regarding Flows. How come this sample code (from docs):

flow {
        // The WRONG way to change context for CPU-consuming code in flow builder
        withContext(Dispatchers.Default) {
            for (i in 1..3) {
                Thread.sleep(100) // pretend we are computing it in CPU-consuming way
                emit(i) // emit next value
            }
        }
    }
        .onEach { value -> println(value) }
        .launchIn(scope)

crashes with java.lang.IllegalStateException: Flow invariant is violated…, but this one does not:

flowOf(1, 2, 3)
        .transformLatest {
            withContext(Dispatchers.Default) {
                Thread.sleep(100) // pretend we are computing it in CPU-consuming way
                println("emitting $it, on thread: ${Thread.currentThread().name}")
                emit(it) // emit next value
            }
        }
        .onEach {
            println("collect.onEach: $it, on thread: ${Thread.currentThread().name}")
        }
        .launchIn(scope)

to me, it should be the same case as the first (should crash with same exception). What am I missing here? logs for 2nd sample is this:

emitting 1, on thread: DefaultDispatcher-worker-1
collect.onEach: 1, on thread: main
emitting 3, on thread: DefaultDispatcher-worker-1
collect.onEach: 3, on thread: main

class Presenter {
   init {
      presenterScope.launch {
         syncer.syncingFinishedEvent
	   .collect {
	      navigateSomewhere()
	   }
      }
      syncer.sync()
   }
}

class Syncer {
   fun sync() {
      syncerScope.launch {
         ...
         syncFinishedEvent.emit(Unit)
      }
   }
}

Is there a way to only call

syncer.sync()

after

syncer.syncingFinishedEvent

is for sure subscribed to? I cannot miss a event

Hi Guys, I have been tyring to write a streaming client in kotlin with Flow apis. Streaming server is TCP based. I have to send request to server for subscribing / unsubscribing and listen the for the updates. I am using callback flow to keep sending the updates back to the subscribbers. Take a look at what i have implemented so far.

private fun createConnection() {

val hostAdd = InetAddress.getByName(endpoint)

val connection = Socket(hostAdd, port)

this.socketConnection = connection

inputStream = DataInputStream(connection.inputStream)

outPutStream = DataOutputStream(connection.outputStream)

}

private fun sendStreamRequest(request: String) {

try {

outPutStream?.writeBytes(request)

outPutStream?.flush()

Log.d(tag, "sendStreamRequest $request")

} catch (ex: Exception) {

ex.printStackTrace()

}

}

The below is the methods subcribes for the updates via callback flow

override fun subscribe(request: String):Flow<String> = callbackFlow {

sendStreamRequest(requestFormatter.formatSubscribeRequest(request))

try {

inputStream?.bufferedReader().takeUnless {

it == null

}.use {

it?.forEachLine { response ->

Log.d(tag, "Streaming $response")

val result = trySend(response)//trySendBlocking(response)

Log.d(tag, "result  ${result.isSuccess}")

}

}

} catch (ex: Exception) {

ex.printStackTrace()

}

awaitClose {

Log.d(tag, "awaitclose")

sendStreamRequest(requestFormatter.formatUnSubscribeRequest(request))

}

}

There are few problems that i see in the subscribe method 1. The whole code is written insider a singleton class so everytime i subscribe this code

inputStream?.bufferedReader()

will create addition objects (Let me know if i am right) 2. if i move the code `inputStream?.bufferedReader()`out of the subscribe method how can i update the subscribber ?? Can i use the produdeScope i get from callback flow builder and call trySend method from outside? 3. I also want to listen for network connection changes if the connetion looses i will establish the connection again and let the subscribber listen the updates again but not sure how to do that. Any guidance on this to improve would be of great help. I know listen streams using TCP may sound weird but it is a legacy code which i am trying to convert into Kotlin.

If I use the new

limitedParallelism

api like this

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

, will the dispatcher always dispatch to the same thread?

is there any operator in flow that allows me to just log an error without having to do

catch { log(it) throw it}

? Something like

doOnError

?

Does anyone knows if I have a custom view that extends

ConstraintLayout

is it good to implement the

CoroutineScope

? Or it's better to create a scope

private val scope = MainScope()

And use it and cancel it depending on if the view is visible or not (attached/detached). The thing is that I'm doing some things and painting the custom view so I'm using a CoroutineScope for this but I'm afraid that if the CustomView is gone the coroutine is not and then try to do something and crash the app. Have anyone of you faced something similar? I used to inject the lifecycle of the "parent" let's say an

Activity

or a

Fragment

and once this is gone the coroutine is cancelled automatically.

Technically, if I dont cancel a CoroutineScope, but all its Jobs completed, then I don't leak memory, right?

Hello folks, I've been trying to migrate from runBlocking to runTest but to no avail. More details inside the thread. Thank you

Hi all! I’m having trouble with a unit test. I’m trying to validate the behaviour of a StateFlow. But, the StateFlow is only in this particular state until a certain suspend function completes. During testing, my test double uses a delay to simulate a long running task. But, due to the nature of

runBlockingTest()

, this delay is skipped, so this function completes immediately, and the StateFlow essentially bypasses the value I’m trying to test. It’s worth mentioning, I’m using CashApp/Turbine as well.

I'd like some advice on backpressure • Cold flows handle backpressure by default by suspending the emitter • Downstream can modify this behavior with

buffer

or

conflate

• Hot streams default behavior varies • MutableSharedFlow is suspend by default but can modify this behavior in the ctor • MutableStateFlow is conflated by default and you CANNOT modify this behavior • In both hot streams, downstream CANNOT modify this behavior (i.e. buffer or conflate dont do anything) Everything correct? (Most importantly whether downstream is able buffer on backpressure from hot upstream)

I'd like to test the following class with the new `runTest`utility:

class SomeManager(
  private val source: DataSource,
  private val externalScope: CoroutineScope
) {
  private val cache: SharedFlow<Foo> by lazy {
    source.fooFlow.shareIn(externalScope, SharingStarted.EAGERLY, 1)
  }

  suspend fun readCached(): Foo {
    return cache.first()
  }
}

Why does the following test hangs forever? How can I test this kind of class that requires an external CoroutineScope ?

@Test fun test() = runTest {
  val subject = SomeManager(someSource, this)
  val cachedValue = subject.readCached()
  assertEquals(expectedFoo, cachedValue)
  // HANGS HERE - due to having at least one child coroutine?
}