Reactive Programming on Android with RxJava

When to use Flowable
• Dealing with 10k+ of elements that are generated in some fashion somewhere and
thus the chain can tell the source to limit the amount it generates.
• Reading (parsing) files from disk is inherently blocking and pull-based which works
well with backpressure as you control, for example, how many lines you read from
this for a specified request amount).
• Reading from a database through JDBC is also blocking and pull-based and is
controlled by you by calling ResultSet.next() for likely each downstream request.
• Network (Streaming) IO where either the network helps or the protocol used
supports requesting some logical amount.
• Many blocking and/or pull-based data sources which may eventually get a non-
blocking reactive API/driver in the future.
Rewriting our example using backpressure with
Flowable
gives us:
¹⁶
https://github.com/ReactiveX/RxJava/wiki/What’s-different-in-2.0

---

Chapter 6: Backpressure
83
1
File
[]
imageFiles
=
// image files to process
2
3
Flowable
<
File
>
fileStream
=
Flowable
.
fromArray
(
4
imageFiles
5
).
subscribeOn
(
Schedulers
.
computation
());
6
7
Flowable
<
Bitmap
>
bitmapStream
=
8
fileStream
.
map
(
file
->
BitmapFactory
.
decodeFile
(
file
.
getAbsolutePath
()));
9
10
Flowable
<
Pair
<
Bitmap
,
Integer
>>
bitmapZippedStream
=
Flowable
.
zip
(
11
bitmapStream
,
12
Flowable
.
range
(1,
imageFiles
.
length
),
13
Pair
::
new
14
).
doOnNext
(
pair
->
Log
.
d
(
TAG
,
"Produced bitmap: "
+
pair
.
second
));
15
16
bitmapZippedStream
.
observeOn
(
17
AndroidSchedulers
.
mainThread
()
18
).
subscribe
(
pair
-> {
19
processBitmap
(
pair
.
first
);
20
Log
.
d
(
TAG
,
"Processed bitmap: "
+
pair
.
second
);
21
});
With this change, we replaced all references of
Observable
to
Flowable
. By default, the backpressure
strategy used by
Flowable
is to impose a
bounded
buffer–the buffer is filled by the producer and
production will halt until most items in the buffer are consumed by the consumer. It may still be the
case, however, that the above code would produce an
OutOfMemoryError
. This is because the default
buffer size is set to be 128 items and depending on your use-case, that size might be too much. To
mitigate this, we can specify a different buffer capacity in an overloaded version of
.observeOn()
1
// ...
2
int
bufferSize
= 10;
3
bitmapZippedStream
.
observeOn
(
4
AndroidSchedulers
.
mainThread
(),
5
false
,
6
bufferSize
7
).
subscribe
(
pair
-> {
8
// ...
9
});
Here we have changed the bounded buffer size to only contain 10 items. The second parameter to
.observeOn()
indicates whether or not an
.onError()
notification should cut ahead of
.onNext()
notifications, in this case, specifying
false
indicates that it can. Running the above code with
backpressure should print out:

---

Chapter 6: Backpressure
84
1
Produced bitmap
: 1
2
Produced bitmap
: 2
3
...
4
Produced bitmap
: 10
5
Processed bitmap
: 1
6
Processed bitmap
: 2
7
...
8
Processed bitmap
: 7
9
Processed bitmap
: 8
10
Produced bitmap
: 11
11
...
As you can see, production of bitmaps halted once the set buffer size of 10 has been reached. The
consumer was then allowed to catch up and production continued when the buffer was close to
empty (in this case, production continued once the buffer size contained 2 items).