https://kotlinlang.org logo
#coroutines
Title
# coroutines
d

dan.the.man

09/09/2020, 2:31 PM
I seem to have an incorrect understanding and I was wondering if someone could help.
Copy code
Fragment {
  lifecycleScope.launch { vm.attachListeners }
}

VM {
  val locationListener:ReceiveChannel<Location>
  val navListener:ReceiveChannel<NavEvent>

  suspend fun attachListeners(){
     locationListener.consumeEach{//Do stuff}
     navListener.consumeEach{//do stuff}
  }
With the code above, if this was RxJava, it would behave correctly. With Coroutines the first
.consumeEach
is blocking it appears, never executing the second one. I can change it to
Copy code
suspend fun attachListeners(){
    coroutineScope{
     launch{
       locationListener.consumeEach{//Do stuff}
     }
     launch {
       navListener.consumeEach{//do stuff}
     }
  }
}
And that will work. The goal is to make it automatically cancel with the lifecycleScope, I'm wondering is there a better way of doing this? And why does .consumeEach block?
c

Casey Brooks

09/09/2020, 2:41 PM
consumeEach
is a suspending function, which means it will suspend at that line until the channel is closed and it has consumed everything in the channel. By wrapping each
consumeEach { }
in
launch
, they are each running in parallel, which is what you want. But now you have a different problem: the
coroutineScope { }
block will suspend until both channels complete, and so will
attachListeners
. This probably isn’t what you intended, you’re wanting
attachListeners
to start off the background jobs for consuming the channels and then return once they’re started (but not finished). For this case, to follow Structured Concurrency, you probably want
attachListeners
to have a receiver of
CoroutineScope
without the
suspend
modifier, so that the
launch { }
blocks are tied into the scope of the caller (the Fragment’s lifecycleScope)
Copy code
fun CoroutineScope.attachListeners(){
    launch{
        locationListener.consumeEach{//Do stuff}
    }
    launch {
        navListener.consumeEach{//do stuff}
    }
}
d

dan.the.man

09/09/2020, 2:52 PM
Ok thanks, there's not another way to launch it within the parent scope in a suspend function?
Copy code
Fragment {
  lifecycleScope.launch { vm.handleAction()}
}
VM {
  val locationListener:ReceiveChannel<Location>
  val navListener:ReceiveChannel<NavEvent>
  
  suspend fun handleAction(action:Action){
    attachListeners()  
}
  
  suspend fun attachListeners(){
     locationListener.consumeEach{//Do stuff}
     navListener.consumeEach{//do stuff}
  }
My full code looks something like this, so changing it to a
CoroutineScope.attach
is then giving me compilation error, I can then wrap it in
coroutineScope { attachListeners}
But that feels like it's the same problem I was just trying to solve by moving it to an extensino of Coroutine scope?
Is it suspending the coroutineScope block a problem? Why does moving it to the extension solve that, I had thought that
coroutineScope {}
would launch another coroutine scope within the parent scope that it was called ffrom, I could have the incorrect understanding on that though
c

Casey Brooks

09/09/2020, 3:11 PM
There’s a lot of subtlety and nuance here, for sure. There are multiple ways of setting it up, and most of them are just decisions on whether to mark a function
suspend
or have a
CoroutineScope
receiver, and where to put the
coroutineScope { }
block. But the overall design of your coroutines will be easier to understand (and follow other libraries) with the general rule that
suspend
functions are designed for things that need to run in sequence (and don’t need a coroutine builder inside them), and
CoroutineScope
receiver functions for running things in parallel (which needs a separate coroutine builder for each parallel task).
And by “running things in parallel” I’m meaning more of starting long-running tasks that don’t need to be joined back together, such as your example of receiving items from independent channels.
d

dan.the.man

09/09/2020, 3:15 PM
Gotcha, that makes sense. Thanks!
c

Casey Brooks

09/09/2020, 3:16 PM
For something like making 2 parallel API calls and joining the results into one, you’ll want to use a
coroutineScope
block inside a
suspend
function with
launch
or
async
inside that, so the two API calls can be in parallel, but to the outside world it’s just a sequential task that returns a single value (albeit, and aggregate value) (see examples here https://medium.com/@elizarov/structured-concurrency-722d765aa952#0be1, also just a good read on this topic)
d

dan.the.man

09/09/2020, 3:17 PM
Yup, I've done that same thing for multiple api calls in parallel, I think I was just getting thrown off that .consumeEach was blocking (suspendable or w/e) and was still thinking of it like it was RxJava Observables
b

bdawg.io

09/09/2020, 11:58 PM
I think your original change is what you wanted. It follows Structured Concurrency well and attaches to the lifecycleScope's lifecycle
2 Views