Reactive Programming on Android with RxJava

Observable.observeOn()
Using
.subscribeOn()
, we are able to modify where an
Observable
does its work. But say, for
example, that we performed a network call to retrieve a
User
’s information using
Schedulers.io()
and we want to update the UI once the network call succeeds. Can you spot the bug in the following
line of code?
1
// `userObservable` retrieves a User object over the network
2
Observable
<
User
>
userObservable
=
// ...
3
4
userObservable
.
subscribeOn
(
Schedulers
.
io
())
5
.
subscribe
(
user
->
textView
.
setText
(
user
.
getName
()));
As you might have noticed in the preceding example, all subsequent operations after
.sub-
scribeOn()
will run on the thread provided by the specified
Scheduler
. In other words, the call
to
textView.setText(user.getName())
will run on a thread other than the UI thread, which will
cause the app to crash with a
CalledFromWrongThreadException
. To circumvent this issue, we can
wrap the call to update the
TextView
in a
Runnable
and have it run on the main thread; however,
doing that is about as elegant as our plain old
Thread
example above. Ideally, we want a declarative
approach to switch threads and post emissions to a separate thread (in this case, to the Android main
thread). For this, we make use of the
.observeOn()
operator.
.observeOn()
modifies the
Observable
so that its emissions will be posted in the provided
Scheduler
.
.observeOn()
is the
Operator
we want if we want the downstream consumer to operate
on a separate thread. To fix the code above, we would simply add an
.observeOn()
before the
.subscribe()
call.

---

Chapter 4: Multithreading
52
1
// `userObservable` retrieves a User object over the network
2
Observable
<
User
>
userObservable
=
// ...
3
4
userObservable
.
subscribeOn
(
Schedulers
.
io
())
5
.
observeOn
(
AndroidSchedulers
.
mainThread
())
6
.
subscribe
(
user
->
textView
.
setText
(
user
.
getName
()));
Running the above code will now execute
textView.setText(user.getName())
in the Android main
thread. Note that the
Scheduler
we used is
AndroidSchedulers.mainThread()
, which is provided
by the
RxAndroid¹²
extension library to RxJava.