What's the `Flow` equivalent to Rx Merge ? <http://reactivex.io/documentation/operators/merge.html>

# coroutines

[darkmoon_uk]

What's the Flow equivalent to Rx Merge ? http://reactivex.io/documentation/operators/merge.html

[zak.taccardi]

https://github.com/Kotlin/kotlinx.coroutines/issues/1491

[darkmoon_uk]

Gah, I need it fast 😞

[darkmoon_uk]

Better make one, shouldn't be too challenging...

[darkmoon_uk]

(Famous last words)

[darkmoon_uk]

@FlowPreview
fun <T> Flow<T>.mergeWith(other: Flow<T>): Flow<T> = flow {
    coroutineScope {
        launch {
            other.collect {
                emit(it)
            }
        }

        launch {
            collect {
                emit(it)
            }
        }
    }
}

[darkmoon_uk]

From: https://proandroiddev.com/an-early-look-at-kotlin-coroutines-flow-62e46baa6eb0
...looks sane.

[darkmoon_uk]

Just caught up on that thread @zak.taccardi; plugging the vararg size into concurrencyLimit feels like it's driving some dead-weight logic in this scenario

[darkmoon_uk]

Hope they go for a separate merge that runs more like your implementation - more direct.

[kevin.cianfarini]

@darkmoon_uk

The "gotcha" with concurrency limit is a strong motivator for a separate merge operator.

Looks like that's the desired approach

[bdawg.io]

IIRC You won't be able to emit from inside of those other jobs. you'll probably need to do a channelFlow

fun <T> Flow<T>.mergeWith(other: Flow<T>) = channelFlow {
    launch { collect { send(it) } }
    launch { other.collect { send(it) } }
}

[bdawg.io]

LOL the channelFlow documentation provides a merge implementation as its example https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/channel-flow.html