Is there any elegant way to bind multiple channels...

# coroutines

[Matej Drobnič]

Is there any elegant way to bind multiple channels to the single scope?

I'm trying to make an actor that receives and processes data from multiple channels:

launch {
    val channelA = openChannelA()
    val channelB = openChannelB()
    val channelC = openChannelC()

    while (isActive) {
        select {
            channelA.onReceive {
                ...
            }
            channelB.onReceive {
                ...
            }
            channelC.onReceive {
                ...
            }
        }
    }
}

However, I also want all of those channels closed if the actor is terminated for any reason

[Matej Drobnič]

From what I can see, consume does that, but it is very ugly:

launch {
    val channelA = openChannelA()
    val channelB = openChannelB()
    val channelC = openChannelC()

    channelA.consume {
        channelB.consume {
            channelC.consume {
                while (isActive) {
                    select {
                        channelA.onReceive {
                            ...
                        }
                        channelB.onReceive {
                            ...
                        }
                        channelC.onReceive {
                            ...
                        }
                    }
                }
            }
        }
    }
}

[gildor]

Implementation with consume is incorrent, it will essentially consume them one by one

[gildor]

select is right way to do that

[gildor]

you also can just set select using forEach on list of channels to avoid duplication

[gildor]

also you can use whileSelect

[Matej Drobnič]

right but if I just use select, channels will not close when coroutine cancels

[Marko Mitic]

Channels don't have scopes but coroutines sending the data using channel do,

[Marko Mitic]

So, if openChannelA/B/C() launches a coroutine in same scope as actor, all those will be canceled with the actor and your channels will be garbage collected eventually

[gildor]

just close channels in context[Job]!!.invokeOnCompletion inside your launch

[gildor]

but even better open channel using produce on launch scope as Marko suggesting

[Matej Drobnič]

openChannelA() do not perform any launchers

[Matej Drobnič]

for example Flowable.openSubscription() from rx integration is not scope-aware

[Marko Mitic]

try/finally should work, not sure how elegant it is

[gildor]

for example Flowable.openSubscription() from rx integration is not scope-aware

Do not use it, instead use Flowable.asFlow, that it will be perfectly lifecycle-safe and scope-aware
Channel is actually bad wrapper for Flowable

[Matej Drobnič]

does flow support select though?

[gildor]

There is no select for Flow, there are much better operators, like merge/zip/concat depending on your use case

[Matej Drobnič]

not sure any of those fits my use case of creating an actor

[gildor]

if you have N Flowable, it’s very easy to create N Flow from them and listen all events and receive them when first of them available (like with single listener for multiple selects). Again, you do not show example how you use it in actor, if you just do the same kind operatio is just flattenMerge() operator

[Matej Drobnič]

all flows are different

[Matej Drobnič]

every one of them has different type and different logic

[Matej Drobnič]

so I can't just merge them

[gildor]

Okay, do not merge them, just collect all of them separately, or just move your logic to onEach operator of each flow and than use launchIn(myScope), where myScope is scope of your actor, everything will be correctly connected by structured concurrency

[Matej Drobnič]

But wouldn't that process them in parallel instead of single sequence like the actor example above?

[gildor]

depends on how you use them, it will run in parallel, but as I said, you can merge them (with some wrapper for every type)