Hi, did anyone combine the values from several `ReceiveChannel`s? What approach did you take?

You could create a subclass of channel that takes vararg amounts of channels as a constructor param. For each channel, create a coroutine and

recieve

the element from that channel. Each time you receive,

send

to

this

channel

I usually redesign. If I can't be arsed I use

select

.

Use asFlow and then combineLatest

I use to use

select

in a

produce

, but I haven't looked at any of the new

Flow

stuff

Oh, I forgot to mention that all channels have different types and that I want to combine them into a data class sent/broadcast to a new channel each time one new value is received

@louiscad is asking a question on coroutines channel. Must be a tough one. Sorry for the noise. 😉

@gaetan Pretty sure he'll ace the implementation though. 😜 Also sorry for noise.

again, fairly easy to do with

select

and

produce

My current implementation is 47 lines of boilerplate code (not counting imports, no comments/KDoc lines) to combine 4 channels, each of different element types into a data class broadcast in a new

ConflatedBroadcastChannel

. I wish I could have a shorter solution that could scale better, that's why I asked. Also, I have no clue how I can do this with

select

and

produce

.

What do you mean by combine? You want to get an item from each one, put all the items in a data class with 4 fields and then return the instance? Why not just call receive on each channel. Or you want to wait for the the first item from any channel and return a transformation/map of the item?

For now, that's what I do, I have a call to receive for each channel, I keep the last received value and each time I receive on new value for any of each channel, I update the last value and update the data class for all last values to be broadcast. I wish I found a more generic or less verbose way to do it for just 4 channels. It's basically over 10 lines of code per channel for now, and I use a few extensions

How about

DataClass(channel1.receive(), channel2.receive(), channel3.receive(), channel4.receive())

?

I also need updates, when only one of the channels produces a new value, I need a new dataclass with the last values previously received from the other channels.

Oh I get it now, you need a stream of latest values.

Yea that's basically what I had in my mind, but with

= produce {

Here's a rather generic approach inspired by the snippet Dominic shared:

fun <E1, E2, R> CoroutineScope.combineLatest(
    one: ReceiveChannel<E1>,
    two: ReceiveChannel<E2>,
    transform: (one: E1, two: E2) -> R
): ReceiveChannel<R> {
    val output = Channel<R>(capacity = Channel.CONFLATED)
    output.invokeOnClose {
        one.cancel()
        two.cancel()
    }
    launch {
        var latestOne: E1 = one.receive()
        var latestTwo: E2 = two.receive()
        output.send(transform(latestOne, latestTwo))
        while (true) {
            select<Unit> {
                one.onReceive { latestOne = it }
                two.onReceive { latestTwo = it }
            }
            output.send(transform(latestOne, latestTwo))
        }
    }.invokeOnCompletion {
        one.cancel()
        two.cancel()
    }
    return output
}

(Edited to return

ReceiveChannel<T>

.) Do you think it's correct, especially regarding closing handling?

If one or two channel gets cancelled

var latestOne: E1 = one.receive()
var latestTwo: E2 = two.receive()

select<Unit> {
        one.onReceive { latestOne = it }
        two.onReceive { latestTwo = it }
}

then these statements will throw

CancellationException

. Is this the behaviour you want?

Yes, that shouldn't be a problem since it'll be eaten by

launch

. I should cancel the two other channels though. I edited the snippet to return

ReceiveChannel<T>

BTW.

Yes, that shouldn't be a problem since it'll be eaten by

launch

But it will cancel the whole scope if it is not

Supervised

.

runBlocking {
            val a = produce {
                while (true) {
                    delay(1_000)
                    send(1)
                }
            }
            val b = produce {
                while (true) {
                    delay(2_000)
                    send(2)
                }
            }

            coroutineScope {
                launch {
                    while (true) {
                        println("Test")
                        delay(1_000)
                    }
                }

                launch {
                    combineLatest(a, b) { a, b -> Pair(a, b) }.consumeEach {
                        println(it)
                    }
                }

                delay(10_000)
                a.cancel()
            }
        }

When

a.cancel()

is called it will throw an

Exception

and cancel the coroutine that prints "Test".

This:

runBlocking {
    launch {}.cancel()
}

doesn't throw because cancellation is not a crash, so it doesn't propagate to parent scope when the coroutine is concurrent (as with

launch

or

async

). So I don't see a problem here.

@louiscad if one channel receives 2 items before the other receives any, does your solution drop the second item for the first channel? Depends on

onReceive

behaviour.

If this is the only receiver, it doesn't.

@Dico The second value would not be received until the select is reached, and then, it'd be received immediately (unless the other channel has a value that is selected first because of bias), and would trigger a new transform and sending.