Understand RxJava lifecycle
RxJava 3.0.x has been released but I think it’s good to learn the fundamental of RxJava — the lifecycle.
Let’s learn those states by this simple example.
Observable.fromArray(1, 2, 3)
.map(v -> v + 1)
.subscribe(System.out::println);Obviously the outputs are 2, 3, 4. Let’s see another version of the example so we could understand it better.
Observable<Integer> obsFromArray = Observable.fromArray(1, 2, 3);
Observable<Integer> obsMap = obsFromArray.map(v -> v + 1);
obsMap.subscribe(new Observer<Integer>() {
@Override
public void onNext(Integer integer) {
System.out.println(integer);
} @Override
public void onSubscribe(final Disposable d) {} @Override
public void onError(Throwable e) {}
@Override
public void onComplete() {}
});
Assembly Time
Observable<Integer> obsFromArray = Observable.fromArray(1, 2, 3);
Observable<Integer> obsMap = obsFromArray.map(v -> v + 1);“The preparation of dataflows by applying various intermediate operators. At this point, the data is not flowing yet and no side-effects are happening.”
Let’s see how it works under the hood:
- First, a new ObservableFromArray created
new ObservableFromArray<T>(items) // items are {1, 2, 3}- Then a new ObservableMap created with the previous ObservableFromArray and a mapper function as params
new ObservableMap<T, R>(this, mapper) // mapper is {v -> v + 1}This finishes the Assembly Time state. You might notice a pattern here when we have a stream like:
Observable.aaa()
.bbb()
.ccc()
.ddd()- aaa operator is called -> new ObservableAaa created
- bbb operator is called -> new ObserfableBbb(observableAaa, others)
- ccc operator is called -> new ObserfableCcc(observableBbb, others)
- ddd operator is called -> new ObserfableDdd(observableCcc, others)
After this state, a below Observable has a reference to the upper Observable. You can imagine it’s like a singly LinkedList established.
Subscription Time
obsMap.subscribe(new Observer<Integer>() {...});“When subscribe() is called on a flow that establishes the chain of processing steps internally. doOnSubscribe() methods are triggered.”
Get Tan Tran’s stories in your inbox
Join Medium for free to get updates from this writer.
Let’s see how it works under the hood:
obsMap.subscribe(new Observer<Integer() {...});
﹀
obsMap.subscribeActual(observer);
﹀
obsFromArray.subscribe(new ObservableMap.MapObserver<T, U>(observer, function));
﹀
obsFromArray.subscribeActual(mapObserver);
﹀
mapObserver.onSubscribe(d); You might notice that the subscription happens from downstream to upstream. Then onSubscribe() is called from upstream to downstream, if doOnSubscribe() was implemented, it would be triggered.
This finishes the Subscription Time. The Runtime will begin right after this.
Runtime
This is the actual code in ObservableFromArray#run()
for (int i = 0; i < n && !isDisposed(); i++) {
T value = a[i];
if (value == null) {
downstream.onError(new NullPointerException("The element at index " + i + " is null"));
return;
}
downstream.onNext(value);
}
if (!isDisposed()) {
downstream.onComplete();
}“When the flows are actively emitting items, errors or completion signals.”
Let’s see how it works under the hood:
- The downstream in the code above is the MapObserver
- MapObserver#onNext() will apply the mapper function {v -> v+1}, then pass the value to downstream.onNext(v)
- The downstream in MapObserver is the Observer that we have passed into obsMap.subscribe(new Observer<Integer>() {}). In this case, downstream.onNext(v) will print the first emitted value (no more downstream)
- So on and so forth until all items get emitted or the stream gets disposed
Exercise
It’s pretty much what happened during the RxJava lifecycle. Let’s do a small exercise to check the output of this code below. You can leave the result on your comment below.
AtomicInteger count = new AtomicInteger();
Observable.range(1, 10)
.doOnNext(ignored -> count.incrementAndGet())
.ignoreElements()
.andThen(Single.just(count.get()))
.subscribe(System.out::println);Conclusion
- There’re many things that happened under a piece of RxJava code which is interesting right? It would be more complex if we got more operators and nested streams involved.
- After this, I hope you could answer some RxJava questions such as how an x operator works, or what happens if we call subscribeOn() multiple times?
Thanks for reading!!
