Reactive Programming on Android with RxJava

Marble diagram of .merge()
Say, for instance, we wanted to combine the results of places recommended by 2 different recom-
mendation engines and merge the results of those two. If each recommendation was encapsulated
in an
Observable
, combining the results of these would be a trivial operation with
.merge()
:
1
Observable
<
Place
>
getNearbyPlacesGoogle
(
Location location
) {
2
// Method that makes a network call to a Google API
3
// to get all nearby places from `location`
4
}
5
6
Observable
<
Place
>
getNearbyPlacesFourSquare
(
Location location
) {
7
// Method that makes a network call to Four Square's API
8
// to get all nearby places from `location`
9
}
Using the two methods defined for retrieving place recommendations, we can combine the results
of each through
.merge()
:

---

Chapter 3: Operators
36
1
Observable
<
Place
>
allNearbyPlaces
=
Observable
.
merge
(
2
getNearByPlacesGoogle
(
location
),
3
getNearByPlacesFourSquare
(
location
)
4
);
5
allNearbyPlaces
.
subscribe
(
place
-> {
6
// Get all nearby places here
7
});
As soon as an
Observer
subscribes to the
Observable
allNearbyPlaces
, each passed in
Observable
is
individually subscribed to internally in
.merge()
and the result is then
merged
and combined. Note
that from the marble diagram of
.merge()
, the resulting emissions can be interleaved, and once an
.onComplete()
is received from an
Observable
, the merged
Observable
completes as well. If the
interleaving behavior is undesired, the
.concat() Operator
may be used to
concatenate
the results
rather than
merge
them. In effect, the passed in
Observable
instances are subscribed to serially
as each one completes rather than simultaneously. If the preceding
Observable
does not complete
however (i.e. an infinite stream), the subsequent
Observable
will not be subscribed to.