Rxjava: flatZip or call it however you want...

[pkafel]

Hi everybody,

Lately I have noticed that from time to time I have the following use case:

• Im zipping many observables by using Observable.zip
• Some values are actually not interesting from my point of view (and I can know that only during the zipping when I have all the values at hand) so I need to have some value that will represent those unwanted result - usually I use null
• Right after Observable.zip I have a Observable.filter where Im filtering out all the null s so only the correct values are propagated further

I think that pretty neat solution would be to have a version of Observable.zip method (it can be called completely diffrent) that is workin like zip but accepts function that returns Observable. Thanks to that in such a case as described above we could simply return Observable.empty() and could skip the filtering step.

What do you think about it ?

[pkafel]

Hey,

maybe I will give you an example of what I mean. I would like to have the following method in Observable class:

public static final <T1, T2, R> Observable<R> flatZip(Observable<? extends T1> o1, Observable<? extends T2> o2, Func2<? super T1, ? super T2, ? extends Observable<? extends R>> zipFunction) {
        // code goes here
    }

Thanks to it we will be able handle described above situation in more clear way like this

Observable.zip(obsA, obsB, (a,b) -> {
if(awesomeCondition(a,b)) return awesomeComputation(a,b);
else return Observable.empty();
});

[akarnokd]

Hi, why not pass the array along to a flatMap via an identity function?

zip(a, b, c -> c).flatMap(v -> empty())

[pkafel]

Hey @akarnokd ,

Thanks for respond however I dont see how could that work. Let me give a working example.

Observable<Integer> obs1 = Observable.range(1, 5);
Observable<Integer> obs2 = Observable.range(2, 6);
Observable.zip(obs1, obs2, (val1, val2) -> {
   int sum = val1 + val2;
   if((sum % 3) == 0) return sum;
   else return null;
}).filter(val -> val != null).subscribe(System.out::println);

With the method that I have proposed you could simply write

Observable<Integer> obs1 = Observable.range(1, 5);
Observable<Integer> obs2 = Observable.range(2, 6);
Observable.flatZip(obs1, obs2, (val1, val2) -> {
   int sum = val1 + val2;
   if((sum % 3) == 0) return Observable.just(sum);
   else return Observable.empty();
}).subscribe(System.out::println);

Of course overloading flatZip to accept more Observable s (just like in case of zip method) would be highly appreciated !

[akarnokd]

Here is a working example:

Observable<Integer> source = Observable.range(1, 10);
Observable.zip(Arrays.asList(source, source, source), v -> v)
.flatMap(a -> Observable.empty())
.subscribe(System.out::println);

You get all the values in a and you can decide what Observable to return.

[pkafel]

Ah, thats clever ! Yes, this will work but this solution comes with some drawbacks:

• inside flatMap you will need to operate on array and indexes instead of names of variables
• in case of different types of source Observable s you will receive Object[] in flatMap as argument and you will need to cast each element of the array
• when casting and reading from array you need to pay attention to order because compiler will not scream if something is wrong (this particular one is pretty small drawback since in case of the same types of source Observable s in zip method you need to do the same... )

[akarnokd]

You can always build your custom factory method out of the components:

public static <T1, T2, R> Observable<R> zipMap(
        Observable<T1> o1, Observable<T2> o2, Func2<T1, T2, Observable<R>> mapper) {
    return Observable.zip(o1, o2, mapper).flatMap(v -> v);
}

[pkafel]

Yes, I can :) Your method has exactly the signature that I wrote in my second comment in this thread. I thought maybe it is worth to extend rxjava with functionality like this but if you think this is not a common use case and I should just have my own utility class for this Im fine with that.

Thanks !

[tobilarscheid]

I like the idea - I think this is commonly useful and vote for reopening!

[francovitali]

I stumbled across this problem on multiple occasions, is a common use case in an API for example, where you need to aggregate data from multiple sources.
Please reopen and implement!

[akarnokd]

Please use composition as suggested in comments above.

[exi]

I would like to argue for a flatZip as well.

Using an Array only works if all components have the same type.

Using my own static method also seems very clunky since it breaks the fluid style of RxJava call chains.

A very common use-case for us is:

Single<Integer> iSingle = fetchInteger();
Single<Float> fSingle = fetchFloat();
Single<String> sSingle = fetchString();
Single.zip(iSingle, fSingle, sSingle, (i, f, s) -> fetchMoreData(i, f, s));

The problem here is that fetchMoreData also returns a Single which is currently not really easy to handle if i don't want to end up with a Single>.

So i'm stuck with:

Single<Integer> iSingle = fetchInteger();
Single<Float> fSingle = fetchFloat();
Single<String> sSingle = fetchString();
Single.zip(iSingle, fSingle, sSingle, MyCustomContainer::create)
  .flatMap(container -> fetchMoreData(container.getI(), container.getF(), container.getS());

Which is also very ugly and required me to have some custom tuple type just to be able to return some async value.

[counter567]

Yeah I also have the same problem, as long as there is no implemented way to get around it I will use the solution from @exi.

[exi]

@akarnokd @pkafel any opinions on my suggestions?

[akarnokd]

@exi https://github.com/ReactiveX/RxJava/issues/3543#issuecomment-160601248

[exi]

As mentioned, this breaks the fluent style and forces everyone to have one of these around for every async type + multiple numbers of arguments combination.
So my argument would be purely around convenience and code clarity here.