I have a coroutine in a coroutine where latter one will rethrow a custom exception like:

// use case
launch { repoFunc() } // try catch does not work because exceptions are propagated, but not rethrown

// repo
suspend fun repoFunc(scope: CoroutineScope) {
  ...
   scope.launch {
      // operation lives as long as application and is only canceled by timeout
      try {
          withTimeout { someOperation() }
      } catch(e: CancellationTimeoutException) {
          throw CustomException("Timeout") // How to catch this in the parent?
      }
   }
}

My question: Are a global CoroutineExceptionHandler,

coroutineScope

and

supervisorScope

my only options to catch the

CustomException

in the parent coroutine?

hi guys, what is a best replacement for

ConflatedBroadcastChannel(Unit)

using the

StateFlow

or

SharedFlow

? my first try is with a

MutableStateFlow<Boolean>(false)

and every time I have a change in my logic that dispatch a signal, I do:

myflow.value = !myflow.value

, but this looks weird 😅

private suspend fun logRemotelyBestEffort(info: LogMessage) {
        try {
            withTimeout(1000) {
                pushApiClient.log(info)
            }
        } catch (ex: Exception) {
            when (ex) {
                is TimeoutCancellationException -> Unit
                is CancellationException -> throw ex
                else -> ...
            }
        }
    }

I hear I'm not supposed to catch CancellationException or I'll break coroutines. But then TimeoutCancellationException extends CancellationException Is that fine to catch? Or should withContext be outside the try catch?

Hello, I’m trying to validate my Google findings. Is it accurate to say that the best (only?) approach to making this testable is to have a dispatcher provider that masks which

Dispatcher

is provided during testing?

Is it considered a no-no for class to extend

CoroutineScope

? e.g.

class MyClass : CoroutineScope by CoroutineScope(Dispatchers.Default) {
  fun doSomething() = produce {
    trySendBlocking(...)
  }
}

The objective is to have

MyClass.doSomething

return a ReceiveChannel using

produce

. Extending CoroutineScope seemed to be the most straightfoward way to accomplish this.

Hi everyone. Here’s a blog that I’ve written about ‘*Structured Concurrency*’ (using Kotlin coroutines). There are certain caveats to consider and a few points to be mindful of, to avoid unexpected behaviour (especially around Job’s lifecycle, coroutineScope and coroutineContext)*.* I’ve tried to cover them with some examples. Hope it helps!

Hi guys, question... how to avoid the emission of the same item twice with

flows

😅? My scenario is the following... I'm implementing MVI with unidirectional data flow on an Android Project. When the user enters some screen an initial intent is emitted for load data. The user goes to another screen and when comebacks, the initial intent is triggered again, how can I avoid this? there is some operator? should I create a custom one (someone has a sample like that)?

I tried to make flow of the inline classes and looks like boxing/unboxing doesn’t work well. I will try to create a sample project and reproduce

Hi everyone. Who can help me my coroutine job leaks i dont get  it why?

private var searchJob: Job? = null
private fun search(query: String = "") {
        searchJob?.cancel()
        searchJob = viewLifecycleOwner.lifecycleScope.launchWhenStarted {
            viewModel.searchTaskList(query).collectLatest {
                adapter?.submitData(it)
            }
        }
    }
LeaksCanary: Fragment received onDestroy callback and FragmentManager is null

Hi there! I am switching to

kotlinx-coroutines 1.5.0

and this now has marked the GlobalScope as a delicate API. I totally understand the reason behind this but I am somewhat stuck now. So my question is: How can I create a

CoroutineScope

that I will then be reusing? I am doing this now:

val scope = CoroutineScope(EmptyCoroutineContext)

... and then I reuse this scope. Does this make any sense? And similar in KotlinJS when using Flows we often times have code like this:

fun callApi() {
  GlobalScope.launch {
    api.getStuff().collect {  // collect() is a suspend function so it needs a coroutine scope
      ...
    }
  }
}

What would be a clean way to get a reusable coroutine scope here? Thanks in advance!

While working with flow I'm having a type mismatch with the following code:

homeRepository
.requestStuff()
.map { stuff -> HomeState.Success(stuff) }
.catch {
   emit(HomeState.Error.Unknown(it))  <- type mismatch as FlowCollector is waiting for the HomeState.Success type
}

By adding

.filterIsInstance<HomeState>()

before

.catch {...}

it works, but I'm wondering if there is a better way 🤔 Thanks!

Hey guys, I’m a bit lost here with the StandaloneCoroutine{Cancelled}. In Android I have an addRepeatingJob call which consumes a Channel from a ViewModel. And in the ViewModel I am sending data to the Channel with :

getViewModelScope().launch {
    while (this.isActive) {
        val dateString = calendarInteractor.getDateString()
        dateStringFlow.value = dateString
        log.debug { "sending date string to channel on thread " + ThreadInfo.currentThread() }
        dateStringChannel.send(dateString)
        delay(100)
    }
}

And after going out of the app and returning I am always having a StandaloneCoroutine{Cancelled} for the sending to Channel part, how come? Why does it get cancelled? The launch block is called once and then stops. Any clues would be appreciated

I have a suspend function that looks something like:

suspend fun foo() {
    collection.forEach { element -> 
        withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
           // Some long running task processing 'element'
        }
    }
}

I'm wondering what would be the simplest way to parallelize the long running tasks (each element would be processed in parallel) I think this is what

async

might be for? It's just not something I've used so far, and I'm not sure what the syntax is..

An example: Two coroutines share a channel

c

. The first coroutine calls

c.send(i)

followed by `c.receive()`; the second does

val i = c.receive()

, then checks if

i

is even and only then does

c.send(i+1)

. The first coroutine will block forever on`c.receive()` if it sent an odd

i

in the beginning. It should be possible to reliably detect this situation under the assumption that no other coroutines know

c

, and detecting (asynchronously) that the dispatcher is out of work. Preferably I would like to be able to do this in the first coroutine, i.e. effectively doing

c.receiveOrThrowOnDeadlock()

.

Hello there, can artifact

kotlinx-coroutines-test

be used in KMM project in commonTest module? or is it still jvm only?

Hi guys, I'm stuck on this for weeks. I'm trying to recreate this but using Flows instead of RxJava.

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
fun <T : Any, U : Any> Observable<T>.notOfType(clazz: Class<U>): Observable<T> {
  checkNotNull(clazz) { "clazz is null" }
  return filter { !clazz.isInstance(it) }
}

/**
 * take only the first ever InitialIntent and all intents of other types
 * to avoid reloading data on config changes
 */
private val intentFilter: ObservableTransformer<TasksIntent, TasksIntent>
  get() = ObservableTransformer { intents ->
    intents.publish { shared ->
      Observable.merge(
        shared.ofType(TasksIntent.InitialIntent::class.java).take(1),
        shared.notOfType(TasksIntent.InitialIntent::class.java)
      )
    }
  }

Is for avoid fire the InitialIntent twice (MVI pattern) can someone know how can I accomplish this? My current implementation of it kinda sucks https://github.com/ChristopherME/counter-android-mvi/blob/685756dca1a6f0e89627cb0e[…]myapplication/presentation/features/counter/CounterViewModel.kt

Why is this happening?

What are your views on using

runBlocking { //room db call }

in

onDestroy()

of an

Activity

to execute a task which must happen when the

Activity

is killed? I could use a

lifecycleScope

which wraps both the db call and

super.onDestroy()

but I was wondering if that might cause any adverse effects.

What does

Dispatchers.Unconfined

do if it ever needs to dispatch ?

runBlocking {
      withContext(Dispatchers.Unconfined) {
        withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
          // do something in IO here
        }
        // I am still in an IO thread
        delay(10) // dispatch
        // what thread am I here?
      }
    }

How can I use channels as hot observers? I want to be able to send events to a channel regardless of anyone is listening or not. Omit events if no one is listening Bonus : The events are buffered if the consumers are consuming slower than the emitters are emitting

One of the 3rd party sdk bundles transitive

kotlinx-coroutines-android:1.4.3-native-mt

dependency. Can’t find information what does it mean? I need to force

-native-mt

variant across my app? Or because I build android application I can safely resolve to normal

1.4.3

variant?

runBlocking {
    val scopeJob = Job()
    val scope = CoroutineScope(scopeJob)
    val job = scope.launch {
    }
}

I am unclear on whether

job

and

scopeJob

are the same. They seem to print different job ids.

job

is the Job returned from scope.launch construct in which I have passed a context which which has scopeJob assigned to it. So they should be the same?

suspend fun doCall(list: ArrayList<String>) {
       list.forEach {
           Api.retrofitService.getProperties(it)
       }

}

In this code getProperties() is another suspend function that needs to be called inside doCall() if my list is having more than 1 item for the first item the network call happens and when the iteration comes for next item the list is null

I have a list of tasks and i before running them i sort them by an order field. (order 0 runs first, then order 1, etc). the generating of that list takes some and i want to start executing before the list is finished so could use a flow for it. but sorting would require the list to be complete, so instead i want a flow that first emits all the entries with order = 0, then when the flow is finished i want to sort the rest of the elements and execute those. is there a way to do that with flows?

How to make a list of objects as sharedFlow

val players: SharedFlow<List<Player>>   = MutableSharedFlow(,Player("Messi"), Player("Ronaldo"))

Is it possible to collect the changes of a list of object if one item is changed or updated?

This job has not completed yet
java.lang.IllegalStateException: This job has not completed yet
	at kotlinx.coroutines.JobSupport.getCompletionExceptionOrNull(JobSupport.kt:1190)
	at kotlinx.coroutines.test.TestBuildersKt.runBlockingTest(TestBuilders.kt:53)
	at kotlinx.coroutines.test.TestBuildersKt.runBlockingTest$default(TestBuilders.kt:45)

am trying to run some tests and I get this error? what is the cause of it and how to fix it?

channel.offer

is now deprecated. Can anybody please help me understanding the

trySend

? I can see 3 flags like

isSuccess

,

isFailure

and

isClosed

. When should I use what?

suspend fun fetch(): Result<List<User>> {
    collectFlow {
        if (...)

offer(Result.Success(listOf(...)))
        else
        offer(Result.Error(Exception())
    }
}

How this can be converted into

trySend

? Shall I use,

trySend(Result.Success...).isSuccess

or

trySend(Result.Error...).isSuccess

. When should I use

trySend...isFailure

and

trySend...isCancel

?

Does anyone else feel the need to post-fix

Flow<T>

class members with

Flow

?

class Server {
    val notificationsFlow: Flow<Notification>
    val notifications: Flow<Notification>
}

Do people prefer

notifications

or

notificationsFlow

?

Is

flatMapLatest

supposed to work correctly with

StateFlow

? In the following code:

val locallyAvailableProjectIdsFlow = storageFlow.flatMapLatest {
    it?.projectEntityPackageStorage?.projectIdsFlow
        ?: emptyFlow<WatchState<List<ProjectId>>>().asStateFlow(WatchState.Idle())
}

storageFlow

and

projectIdsFlow

are `StateFlow`s, but

locallyAvailableProjectIdsFlow

is not being updated.

asStateFlow

is defined as:

private fun <T> Flow<T>.asStateFlow(initialValue: T) =
    stateIn(coroutineScope, SharingStarted.Eagerly, initialValue)

Am I doing anything wrong?