Reactive Programming on Android with RxJava

Backpressure using Subscriber
Fine-grained control of how a consumer notifies a producer that it can consume more data
can be done by using a
Subscriber
. In addition to the standard observer methods
.onNext(T)
,
.onError(Throwable)
, and
.onComplete()
, a
Subscriber
has an additional method,
.onSub-
scribe(Subscription)
, which is called upon subscribing to a producer (i.e.
Flowable
). The
Sub-
scription
object received from
.onSubscribe(Subscription)
is then used to control the flow of
production from the producer through its method
.request(long)
. As soon as
.onSubscribe()
is invoked, we must immediately request data from the producer via
.request(long)
to begin
receiving items.
1
int
bufferSize
= 10;
2
bitmapZippedStream
.
observeOn
(
3
AndroidSchedulers
.
mainThread
(),
4
false
,
5
bufferSize
6
).
subscribe
(
new
Subscriber
<
Pair
<
Bitmap
,
Integer
>>() {
7
8
Subscription subscription
;
9
10
@Override
11
public
void
onSubscribe
(
Subscription s
) {
12
subscription
=
s
;
13
subscription
.
request
(1);
14
}
15

---

Chapter 6: Backpressure
85
16
@Override
17
public
void
onNext
(
Pair
<
Bitmap
,
Integer
>
pair
) {
18
// Do something with `pair`
19
subscription
.
request
(1);
20
}
21
22
@Override
23
public
void
onError
(
Throwable throwable
) {
24
}
25
26
@Override
27
public
void
onComplete
() {
28
}
29
});
Above, we request 1 item to be produced as soon as the
Subscriber
is subscribed. Afterwards, for
each
.onNext()
item, we request 1 more item to be produced. There’s nothing very special here and
it essentially behaves the same way as before. If however our consumption of data should be limited
and we want to communicate this to the producer (e.g. say we were caching the bitmaps and we
want to stop receiving them when the cache is full) we can do that by conditionally requesting from
the
Subscription
object in
.onNext()
.
1
.
subscribe
(
new
Subscriber
<
Pair
<
Bitmap
,
Integer
>>() {
2
// ...
3
@Override
4
public
void
onNext
(
Pair
<
Bitmap
,
Integer
>
pair
) {
5
// Do something with `pair`
6
if
(
shouldReceiveMoreBitmaps
()) {
7
subscription
.
request
(1);
8
}
9
}
10
// ...
11
});
Throttling and Buffering Items
In addition to backpressure, there are several handy operators that can be used to solve the problem
of a fast producer but a slow consumer. These operators allow you to either
throttle
or
buffer
items
so that a consumer can keep up. Oftentimes, these operators are preferred over backpressure as they
are much simpler to work with.

---

Chapter 6: Backpressure
86
Throttling
Let’s look at an Android-specific example where we might want to apply these principles–a device’s
accelerometer events. First, let’s create a class called
DeviceSensorManager
that can be queried for
an
Observable
that emits accelerometer events.
1