Reactive Programming on Android with RxJava

public class
LocationManager
{
2
3
private
Subject
<
Location
>
locationSubject
=
PublishSubject
.
create
();
4
5
/**
6
* Invoke this method when this LocationManager should
7
* start listening to location updates.
8
*/
9
public
void
connect
() {
10
this
.
googleApiClient
.
connect
();
11
}
12
13
@Override
public
void
onConnected
(
@Nullable
Bundle bundle
) {
14
// start listening to location updates
15
final
LocationRequest locationRequest
=
new
LocationRequest
();
16
locationRequest
.
setInterval
(1000);
17
locationRequest
.
setPriority
(
LocationRequest
.
PRIORITY_HIGH_ACCURACY
);
18
LocationServices
.
FusedLocationApi
.
requestLocationUpdates
(

---

Chapter 5: Reactive Modeling on Android
74
19
googleApiClient
,
20
locationRequest
,
21
location
-> {
22
locationSubject
.
onNext
(
location
);
23
}
24
);
25
}
26
27
/**
28
* Call to receive device location updates.
29
* @return An Observable emitting location updates
30
*/
31
public
Observable
<
Location
>
observeLocation
() {
32
return
locationSubject
.
hide
();
33
}
34
}
In this new implementation, the subtype
PublishSubject
is used which emits events as they arrive
starting from the time of subscription. That is, if a subscription is performed at a point when
location updates have already been emitted, past emissions will not be received by the observer,
only subsequent ones will be emitted. Also notice that the call
.hide()
was invoked on the
Subject
which is generally what you want when exposing the
Subject
externally so that consumers can
only use it as an
Observable
and not as an
Observer
.
In addition to
PublishSubject
, RxJava has a few other varieties of
Subject
types:
•
PublishSubject
: emits to an observer only items that are emitted after the time of the
subscription
•
AsyncSubject
: emits to an observer only the last value after the
Observable
successfully
completes
•
BehaviorSubject
: option subscription, a BehaviorSubject will emit to an observer the most
recently emitted item and any subsequent items
•
ReplaySubject
: emits to an observer all previously emitted items and any subsequent items
View Events
UI events such as a
Button
clicks or changes in an
EditText
are another good place to use RxJava.
An open source library,
RxBinding¹³
(developed by Jake Wharton), provides the necessary bindings
to turn
View
events into
Observable
streams.
Using RxBinding, you can turn
View
click events into an
Observable
.
¹³
https://github.com/JakeWharton/RxBinding

---

Chapter 5: Reactive Modeling on Android
75
1
Observable
<
Object
>
clicks
=
RxView
.
clicks
(
view
);
2
clicks
.
subscribe
(
obj
-> {
3
// Get click events here
4
});
This example is quite trivial but the benefits of using Rxjava become apparent when used in
conjunction with operators. Say for example we wanted to detect double clicks (i.e. 2 clicks that
happen within 300 milliseconds apart).
1
Observable
<
Object
>
clicks
=
RxView
.
clicks
(
view
);
2
3
Observable
<
List
<
Object
>>
clickAggregate
=
4
clicks
.
buffer
(300,
TimeUnit
.
MILLISECONDS
);
5
6
Observable
<
List
<
Object
>>
doubleClicks
=
7
clickAggregate
.
filter
(
list
->
list
.
size
() >= 2);
8
9
doubleClicks
.
subscribe
(
o
-> {
10
Timber
.
d
(
"Received double clicks."
);
11
});
Normally the above method calls to the click
Observable
would be chained, but for educational
purposes I’ve separated it out line-by-line so we can understand the transformations at each step.
First, we convert click events from
view
to an
Observable
using
RxView.clicks()
. After that, we
use the operator
.buffer()
which buffers items that occur within a provided timespan–in this case
300 milliseconds–into a
List