Rxjava for the real beginner: Subject of RXjava

What is Rxjava? Where are here? Who am I?

Introduction

This article is for a very very beginner of Rxjava. Maybe it just has a basic knowledge with you, but I wish there was a very basic article for a newbie like me, who has a gap with belief inside the soul.

Thanks for

 — my mentor, who supported me with very basic questions.

BASIC

Rx java, stand for Reactive JAVA. Reactive is reactive. To learn something, you need to read more and more about it. In simple words, In Rx programming data flows emitted by one component and the underlying structure provided by the Rx libraries will propagate those changes to another component that are registered to receive those data changes.

Rx is based around two fundamental types is Observable and Observer. We will go on very fast with example.

1. PublishSubject

is the most straight-forward kind of subject. When a value is pushed into the subject pushes it to every subscriber that is subscribed to it at that moment.

void testPublishSubject() {
    PublishSubject<Integer> subject = PublishSubject.create();
    subject.onNext(0);
    subject.onNext(1);
    subject.onNext(2);
    subject.subscribe(value -> System.out.println("publishSubject: LATE " + value),
            throwable -> System.out.println("publishSubject: ERR"),
            () -> System.out.println("publishSubject: Complete"));
    subject.onNext(3);
    subject.onNext(4);
    subject.onNext(5);
    subject.onComplete();
    subject.onNext(6);
}//OUTPUT:
publishSubject: 3
publishSubject: 4
publishSubject: 5
publishSubject: Complete/** As you see, it value after subscribe.

2. ReplaySubject

has the special feature of caching all the values pushes to it. When a new subscription is made, the event sequence is replayed from the start for the new subscriber. After catching up, every subscriber receives new events as they come.

void testReplaySubject() {
    ReplaySubject<Integer> subject = ReplaySubject.create();
    subject.onNext(0);
...//OUTPUT
replaySubject: 0
replaySubject: 1
replaySubject: 2
replaySubject: 3
replaySubject: 4
replaySubject: 5
replaySubject: Complete/** As you see, it cached value before subscribe.

ReplaySubject can create with limit size and limit time. To make clear, with limit sized, you can cache up to max limit onNext. With limit time, we can cache up to max time onNext.

void testReplaySubjectWithLimit() {
    ReplaySubject<Integer> subject = ReplaySubject.createWithSize(1);
    subject.onNext(0);
...//OUTPUT:
replaySubject: 2
replaySubject: 3
replaySubject: 4
replaySubject: 5
replaySubject: Complete/** As you see, it's don't save value 0.

3. BehaviorSubject

will save one value before subscribing (optional), It’s same as ReplaySubjectWithSize(1)

void testBehaviorSubject(){
    BehaviorSubject<Integer> subject = BehaviorSubject.create();
    subject.onNext(0);
...//OUTPUT
behaviorSubject: 2
behaviorSubject: 3
behaviorSubject: 4
behaviorSubject: 5
behaviorSubject: Complete/** As you see, it's save value 2 (which before subscribe). And it same as ReplaySubject.createWithSize(1).

4. AsyncSubject

also caches the last value. The difference now is that it doesn’t emit anything until the sequence completes. Its use is to emit a single value and immediately complete it.

void testAsyncSubject() {
    AsyncSubject<Integer> subject = AsyncSubject.create();
    subject.onNext(0);
...//OUTPUT
asyncSubject: 5
asyncSubject: Complete/** As you see. It just emit last value before complete.

---

Apply for real problems…

Conclusion

Okay. It’s enough for the first basic article. You can try implement it and make it easy. Why don’t you use Rxjava in your next project? Take a cup of coffee and reply to me if you did not agree with my vision.

I’d move this post from medium to my blog