What s the `Flow` equivalent to Rx Merge <http reactivex io kotlinlang #coroutines

Channels

100daysofkotlin

100daysofkotlin-2021

algolialibraries

android-architecture

android-databinding

androidgithubprojects

androidx-xprocessing

arrow-contributors

beepiz-libraries

cincinnati-user-group

codemash-precompiler

codingconventions

competitive-programming

competitivecoding

compose-android

compose-desktop

compose-ui-showcase

connect-audit-events

cscenter-course-2016

domain-driven-design

effective-kotlin

effectivekotlin

embedded-kotlin

event21-community-content

fb-internal-demo

fluid-libraries

framework-elide

getting-started

github-workflows-kt

graviton-browser

image-processing

intellij-plugins

intellij-tricks

introduce-yourself

java-to-kotlin-refactoring

jhipster-kotlin

jvm-ir-backend-feedback

k2-early-adopters

kotest-contributors

kotlin-by-example

kotlin-data-storage

kotlin-foundation

kotlin-in-action

kotlin-multiplatform-contest

kotlin-pakistan

kotlinforbeginners

kotlingforbeginners

kotlinprogrammers

kotlintest-devs

kotlinultimatechallenge

kotlinx-datetime

lambdaworld_cadiz

language-evolution

language-proposals

leedskotlinusergroup

library-development

machinelearningbawas

myndocs-oauth2-server

navigation-architecture-component

ohio-kotlin-users

pattern-matching

pocketgitclient

practical-functional-programming

prozis-android-backup

python-contributors

realworldkotlin

refactoring-to-kotlin

refreshversions

russian-kotlinasfirst

skia-wasm-interop-temp

sofl-user-group

spring-security

squarelibraries

stayhungrystayfoolish

strife-discord-lib

swagger-gradle-codegen

swiss-user-group

tricity-kotlin-user-group

Title

d

darkmoon_uk

09/24/2019, 3:03 AM

What's the

Flow

equivalent to Rx Merge ? http://reactivex.io/documentation/operators/merge.html

z

zak.taccardi

09/24/2019, 3:04 AM

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

d

darkmoon_uk

09/24/2019, 3:04 AM

Gah, I need it fast 😞

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

(Famous last words)

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

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

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

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

Hope they go for a separate

merge

that runs more like your implementation - more direct.

k

kevin.cianfarini

09/24/2019, 5:27 PM

@darkmoon_uk

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

Looks like that's the desired approach

b

bdawg.io

09/24/2019, 6:56 PM

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) } }
}

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

🤣 1

😅 1

3 Views

#coroutines Join Slack