Is there a way to do a cartesian product of two flows Someth kotlinlang #coroutines

in RxJava? I have a flow of files and a flow of processors and I need to run all the processors through all the files. I'm looking for something like:

filesFlow.join(processorsFlow) { file, processor -> processor.invoke(file) }

(

proccessor.invoke

returns the result of the processor so I get new flow of type result)

I was thinking in something like this

fun <A, B, AB> Flow<A>.join(
    flowB: Flow<B>,
    selector: (A, B) -> AB
): Flow<AB> {
    return this.flatMapMerge { a -> flowB.map { b -> selector(a, b) } }
}

But with this solution I'm subscribing multiple times to flowB.

Dominaezzz

02/06/2021, 12:57 PM

There isn't a built in operator for this, I'm guessing the expectation is to write one yourself (or maybe they haven't gotten to it yet).

This is an interesting use case. I guess you could use a

SharedFlow

with infinite replay buffer to only subscribe once.

Brais Gabin

02/06/2021, 1:03 PM

I was thinking about it... but then I need to handle the no-end problem of

SharedFlow

too. 😭 we need a sharable & finishable

Flow

Dominaezzz

02/06/2021, 1:08 PM

Yeah, passing a finish token through is a bit of a faff.

👍 1

ephemient

02/06/2021, 3:02 PM

Untitled

👍 1

Erik

02/06/2021, 5:48 PM

@ephemient nice solution. If the types (in this case files and processors) are different and known, then you might not need the intermediate step of mapping to an Either-Left or -Right type, but you can immediately check for the known types. Or am I missing something?

ephemient

02/06/2021, 8:30 PM

@Erik correct, if you can distinguish them in other ways then you don't need the wrapping type. any other method of tagging would work fine

10 Views

#coroutines Join Slack