rxjava merge observables of different type

I'm new to rxjava. I need to combine two observables that emit objects of different type. Something like Observable<Milk> and Observable<Cereals> and get a Observable<CerealsWithMilk>. I couldn't find any operator for something like this. What would be the rx way of doing something like this? Note that Milk and Cereals are async.

Solution 1:

It's hard to say without knowing exactly what you need, but possibly zip() or combineLatest().

zip will take both Observable<Milk> and Observable<Cereals> and let you combine them into CerealsWithMilk via a provided function. This emits a new CerealsWithMilk each time you get get both a Milk and a Cereals.

combineLatest is similar to zip except it will emit a new CerealsWithMilk even if just a new Milk or just a new Cereals is emitted.

Solution 2:

If you want to merge observables of different type you need to use Observable.zip:

Observable<String> o1 = Observable.just("a", "b", "c");
Observable<Integer> o2 = Observable.just(1, 2, 3);
Observable<String> result = Observable.zip(o1, o2, (a, b) -> a + b);

result will be an observable generating the application of (a, b) -> a + b to o1's and o2's items. Resulting in an observable yielding "a1", "b2", "c3".

You can use any function. For example

    Observable<String> o1 = Observable.just("a", "b", "c");
    Observable<Integer> o2 = Observable.just(1, 2, 3);
    Observable<String> result = Observable.zip(o1, o2, (a, b) -> myFunc(a, b));
    //...

    private String myFunc(String a, Integer b) {
            //your code here
            return someString;
        }