Hello folks, I got the idea that catching and not rethrowing CancellationExceptions would be an anti-pattern but couldn't really find any documentation specifying that, does anyone know any docs saying this or any possible problems due to that? Thank you in advance!

This message was deleted.

I have a flow of type Boolean named selected and I want to get that value inside autoLocation Observer. how can I do that? , here's my code: PreferencesManager.kt

class PreferencesManager(private val context: Context) {
                     ...
val selected : Flow<Boolean>
        get() = context.counterDataStore.data.map { preferences ->
            preferences[AUTO_LOCATION] ?: false
           }
}

LocationFragment.kt

class LocationFragment : Fragment() {
                   ...
      lifecycleScope.launch {
            mainViewModel.autoLocation.collectLatest {
              //  check if selected      
                    it?.let {
                        mainViewModel.addPlaceItem(it)
                      }
                   }
            
          }
}

MainViewModel.kt

class MainViewModel(
     private val application: Application
    private val placeRepository: PlaceRepository,
    private val preferencesManager: PreferencesManager,
) : ViewModel() {
                        ...
val selected = preferencesManager.selected
}

hey team, i am trying to bump kotlin coroutines 1.6.0 and see the following error:

* What went wrong:
Execution failed for task ':app:mergeDebugJavaResource'.
> A failure occurred while executing com.android.build.gradle.internal.tasks.MergeJavaResWorkAction
   > 2 files found with path 'META-INF/kotlinx_coroutines_core.version' from inputs:
      - ../org.jetbrains.kotlinx/kotlinx-coroutines-core/1.6.0/c0277601cc4f7f24fe200651c5680e2e424a0a85/kotlinx-coroutines-core-1.6.0.jar
      - ../org.jetbrains.kotlinx/kotlinx-coroutines-core-jvm/1.6.0/f3b8fd26c2e76d2f18cbc36aacb6e349fcb9fd5f/kotlinx-coroutines-core-jvm-1.6.0.jar
     Adding a packagingOptions block may help, please refer to
     <https://google.github.io/android-gradle-dsl/current/com.android.build.gradle.internal.dsl.PackagingOptions.html>
     for more information

I am adding this dependency:

org.jetbrains.kotlinx:kotlinx-coroutines-core:1.6.0

I wonder who adds

kotlinx-coroutines-core-jvm

version? Is it included under

kotlinx-coroutines-core

?

Is it possible to launch a coroutine in a flow operator?

suspend fun producer() = flow {}

private suspend packetFlow(...): Flow<ResponsePacket> {
  return producer()
     .transformWhile {
        // how to launch coroutine here that does not block?
        // coroutineScope {} blocks so it's not an option
        launch {
          delay(2000L)
          state = 1 // set state
          return@transformWhile false // terminate flow
        }
     }
}

I would like to start a timer and terminate the flow after time has expired.

Why does the combine operator of flow works with arrays(instead of list or iterator)? From source code:

public inline fun <reified T, R> combine(
    flows: Iterable<Flow<T>>,
    crossinline transform: suspend (Array<T>) -> R
): Flow<R> {
    val flowArray = flows.toList().toTypedArray()
    return flow {
        combineInternal(
            flowArray,
            arrayFactory = { arrayOfNulls(flowArray.size) },
            transform = { emit(transform(it)) })
    }
}

We have a search screen with paging 3. Each time a new word is typed we create a new flow with a new PagingSource doing api requests. How do I plug those new flows to my ViewModel flow? Right now I’m reEmiting them to my flow, but I’d rather reassign my flow to paging flow each time, which I can’t do

With a

suspend () -> Unit

, you can create a

Continuation

by calling

createCoroutine()

. How can you create a

CancelableContinuation

?

If

Flow

is stream and

shared (state) flow

is hot stream can i somehow sent

completion

event through?

Hey guys, I have a piece of code which was making a single request to get something and then producing a calculation with the result. The first get (getPotatos) was taking around X milliseconds. But now I have to get two things and I’ve came up with the following solution. Although my problem is that now the first get is taking twice the time it was taking in first place. Is it possible that the way I’m doing the requests concurrently can affect each other?

suspend fun getApples(): List<Apple> = ...

suspend fun getPotatos(): List<Potato> = ...

fun work(): ReceiveChannel<SomeResult> = GlobalScope.produce {

    try {
        val (apples, potatos) = coroutineScope {
            val getApples = async {
                val (resp, time) = measureTimedValue {
                    getApples()
                }
                <http://logger.info|logger.info> { time.inWholeMilliseconds }
                resp
            }
            val getPotatos = async {
                val (resp, time) = measureTimedValue {
                    getPotatos()
                }
                <http://logger.info|logger.info> { time.inWholeMilliseconds }
                resp
            }
            getApples.await() to getPotatos.await()
        }


        send(doSomething(apples, potatos))...

    } catch(ex: Exception) {
        handle...
    }
}

Hey folks, Question about kotlin coroutines testing api in 1.6.0 when i should use

val dispatcher = StandardTestDispatcher()
val scope = TestScope(dispatcher)

@Test
fun testFoo() = runTest(dispatcher) {
    // ...
}

vs

val dispatcher = StandardTestDispatcher()
val scope = TestScope(dispatcher)

@Test
fun testFoo() = runTest {
    // ...
}

what if i don’t pass the dispatcher, I know it will default to

StandardTestDispatcher

but wouldn’t be those two different

dispatcher

here? 1. created by runTest 2. created by

dispatcher

property

Hi everyone. I'm trying to create a key-based mutex for the JVM. Can anyone tell me if this looks like a correct and safe implementation?

class KeyedMutex<T> {
    @PublishedApi
    internal val mutexes = ConcurrentHashMap<T, Container>()

    data class Container(
        val mutex: Mutex,
        val holders: Int
    )

    suspend inline fun <U> use(key: T, closure: () -> U): U {
        val result = mutexes
            .compute(key) { _, container ->
                container?.copy(holders = container.holders + 1)
                    ?: Container(Mutex(), holders = 1)
            }!! // always associates the key
            .mutex
            .withLock(action = closure)

        mutexes.computeIfPresent(key) { _, container ->
            if (container.holders > 1) container.copy(holders = container.holders - 1)
            else null
        }

        return result
    }
}

Hi guys, where should i handle networking errors with the api in the MVVM pattern?

What’s the best to do in tests? Use

UnconfinedTestDispatcher

and advance the coroutines automatically? or Use

StandardTestDispatcher

and manually execute the jobs? Do we have any recommendations from the official docs?

I have a download progress listener and want to conflate the results so only the latest progress is persisted to the db. I only have a callback

override fun onProgress(Int)

. Can I use a Flow for this?

Is there a way to observe a Job for changes in child jobs - so that we can potentially visualise a 'queue' of jobs - without having to further abstract?

what's the best way to emit the results of one flow into another? For example i am using a bluetooth library that has states for

Connected

and

Disconnected

as a

StateFlow

. I am trying to add more states such as

Connecting

or

LookingForSavedDevice

at a higher level in my program so I would like to use a

StateFlow

which both takes in emissions from the library functions as well as allow me to mutate the

StateFlow

myself. So I think I'm looking for a way to collect into another flow:

bluetoothLibrary.stateFlow.collect(myOtherStateFlow)

I'm aware I can just emit into the new flow in a

collect { }

block or

onEach { }

but I'm not sure what is canonical

I have a long computation in a non-suspending function which consists of several transformations (map, reduce etc.) on a

Sequence

. How do I make this computation properly cancelable? All the Sequence functions are non-suspending, and the coroutine context is not available in a non-suspending function, so I can’t check for

isActive

. Can I capture the coroutine context in the outermost (suspend) function and pass it into all the (non-suspending) sequence functions?

So, I have a list of tasks that I need to run based on their priority when the android app is launched. I would like to make the Main Thread wait the finish of these tasks execution, but I would like to delegate the execution to background threads....

runBlocking(Dispatchers.Main.immediate) {
    initialize(application)
}

suspend fun initialize(app: Application) = coroutineScope {
    registry.groupBy { initializer ->
        initializer.priority().ordinal
    }
        .toSortedMap()
        .forEach { entry ->
            val tasks = entry.value.map { initializer ->
                async {
                    initializer.initialize(app)
                }
            }
            tasks.awaitAll()
        }
}

The code above runs, but if I change

async {

to

async(Dispatchers.Default) {

Makes the app blocks forever. Would anyone have an idea of how to fix it ?

Is there a way to use sequence {} with a suspend context? I'm using ktor to stream audio from an endpoint and would like to yield this in parts, but I get the error "Restricted suspending functions can only invoke member or extension suspending functions on their restricted coroutine scope"

Hello, I am studying the internal working of coroutine. I have a simple function like below.

fun main() = runBlocking {
 delay(1000)
 println("Bye, world!")
}

If decompile its bytecode into Java, will get the following output:

public final class MainKt {
   public static final void main() {
      BuildersKt.runBlocking$default((CoroutineContext)null, (Function2)(new Function2((Continuation)null) {
         int label;

         @Nullable
         public final Object invokeSuspend(@NotNull Object $result) {
            Object var3 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
            switch (this.label) {
               case 0:
                  ResultKt.throwOnFailure($result);
                  this.label = 1;
                  if (DelayKt.delay(1000L, this) == var3) {
                     return var3;
                  }
                  break;
               case 1:
                  ResultKt.throwOnFailure($result);
                  break;
               default:
                  throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
            }

            String var2 = "Bye, world!";
            System.out.println(var2);
            return Unit.INSTANCE;
         }

         @NotNull
         public final Continuation create(@Nullable Object value, @NotNull Continuation completion) {
            Intrinsics.checkNotNullParameter(completion, "completion");
            Function2 var3 = new <anonymous constructor>(completion);
            return var3;
         }

         public final Object invoke(Object var1, Object var2) {
            return ((<undefinedtype>)this.create(var1, (Continuation)var2)).invokeSuspend(Unit.INSTANCE);
         }
      }), 1, (Object)null);
   }
}

If the function end with

return var3

in

case 0:

in the code above, there should be a code that can enter the function again. I opened runBlocking and delay but couldn’t find it. Where are giving code to continue with label 1 ? (Sorry for the long code snippet)

If I get an exception for a coroutine scope, it’s marked as canceled even with a handler. What is the best practice to make another call with that scope? since it’s marked canceled my instinct is to create a new one, is that the preferred approach? I thought the handler would ensure it was not canceled, which doesn’t seem to be the case.

Can anyone pls explain how come

delay

sometimes takes 30m instead of expected 4m (that were supplied as argument)? It is observed on Android.

IntelliJ (and dokka soon) shows a warning when using `suspend fun CoroutineScope.foo() { }`:

Ambiguous coroutineContext due to CoroutineScope receiver of suspend function

, because

CoroutineScope.coroutineContext

and

coroutineContext

. But why does coroutines-core itself use this construct? https://github.com/Kotlin/kotlinx.coroutines/blob/52cbf01c63c0737ee745ff08568f26385271a9ab/kotlinx-coroutines-core/common/src/Builders.common.kt#L50

Hey folks 👋, I’m relatively new to coroutines I have a question about coroutine scopes which I can’t seem to find a consistent answer for. Any input is appreciated! Say I have in an application I have two classes, an outer which will live for the whole lifecycle of the application and an inner class which is only alive for part of the lifecycle. For the outer class I create a coroutine scope which will be cancelled if the class is closed:

class Outer : Closeable {
    private val scope = CoroutineScope(Dispatchers.Main + SupervisorJob())
    private var inner: Inner? = null

    fun startInner() {
        inner = Inner(InnerRepo(), InnerApi())
    }

    fun stopInner() {
        inner.close()
        inner = null
    }

    override fun close() {
        scope.cancel()
    }
    ... // More code that uses the outer scope
}

And in the inner I have some code that calls suspend functions:

class Inner(
    private val repo: InnerRepo,
    private val api: InnerApi
) : Closeable {
    private val innerScope = CoroutineScope(<http://Dispatchers.IO|Dispatchers.IO> + SupervisorJob())    

    fun sendData() {
        innerScope.launch {
            api.sendData(repo.getData())
        }
    }

    override fun close() {
        innerScope.cancel()
    }
}

Is there a way to link the

innerScope

to the

outerScope

as it makes sense that the

innerScope

would always want to be cancelled if the outer scope cancels, or is that even something that should be done? Also, if the pattern above is completely wrong let me know! Cheers

suspend fun main() {
	coroutineScope {
   		flowOf<Unit>().shareIn(this, SharingStarted.WhileSubscribed())
    }
}

https://pl.kotl.in/gaJVXPMWr = Evaluation stopped while it's taking too long️. Why? The child coroutine of shareIn should, according to docs, never start

Is there a way to

cancelAllChildren

of a

runTest

scope? I have a test where I inject the test

coroutineContext

into a ViewModel and I want to cancel all running jobs inside the ViewModel when the test ends.

What is the best way to create a copy of a Map in Kotlin without causing ConcurrentModificationException in a multi-threaded environment? I almost would like

map.toMap()

to happen atomically. I wanted to see whether something exists before creating my own synchronization blocks.