Subjects

- Subjects act as both an observable and an observer.
A Subject is a sort of bridge or proxy that is available in some implementations of ReactiveX that acts both as an observer and as an Observable. Because it is an observer, it can subscribe to one or more Observables, and because it is an Observable, it can pass through the items it observes by re-emitting them, and it can also emit new items

There are four subject types in RxJava:

1. PublishSubject
2. BehaviorSubject
3. ReplaySubject
4. AsyncSubject

Publish Subject :

It emits all the subsequent items of the source Observable at the time of subscription.In simple language observer gets item from observable or source after it subscribe.

So let's understand by marble diagram,first row is source it emitting 3 items and second and third is subscriber, so second

subscribe after item one emits so it will get item 2 and 3. And third subscriber will get only 3 rd item.

Here, if a student entered late into the classroom, he just wants to listen from that point of time when he entered the classroom. So,Publish will be the best for this use-case.

Code : 

PublishSubject<Integer> source = PublishSubject.create(); // It will get 1, 2, 3, 4 and onComplete source.subscribe(getFirstObserver()); source.onNext(1); 

source.onNext(2); 

source.onNext(3); // It will get 4 and onComplete for second observer also. source.subscribe(getSecondObserver()); source.onNext(4); 

source.onComplete();

Let's understand code, I have created a Publish subject of return integer type, which emiting 1,2,3,4 items, first observer subscribe, in observer onNext method it will get 1,2,34 and then oncomplete gets call.
2nd observer subscribe after 1,2,3 items emitted so it will not available to him, so in his onNext block he will get 4 and then onComplete call back method call.

Don't go in code details just understand concept and when can we use in our code.

Behavior Subject :

It emits the most recently emitted item and all the subsequent items of the source Observable when an observer subscribes to it.

Lets understand by marble diagram, here first raw is observable source emitting three items and second and third row is observer or consumer, as you can see in second row in observer it subscribed after item 1 emitted by source but he 1 as well 2 and 3. and similar for 2nd subscriber he received 2 and 3, although he subscribed after 2 emitted.    

Here, if a student entered late into the classroom, he wants to listen the most recent things(not from the beginning) being taught by the professor so that he gets the idea of the context. So, here we will use Behavior.

Code Snippet: 

BehaviorSubject<Integer> source = BehaviorSubject.create();

// It will get 1, 2, 3, 4 and onComplete 

source.subscribe(getFirstObserver());
source.onNext(1); 

source.onNext(2); 

source.onNext(3);

// It will get 3(last emitted)and 4(subsequent item) and onComplete 

source.subscribe(getSecondObserver());
source.onNext(4); source.onComplete();

Replay Subject :

It emits all the items of the source Observable, regardless of when the subscriber subscribes.

Here, if a student entered late into the classroom, he wants to listen from the beginning. So, here we will use Replay to achieve this.
See the below example:

ReplaySubject<Integer> source = ReplaySubject.create();
// It will get 1, 2, 3, 4

 source.subscribe(getFirstObserver());
source.onNext(1); 

source.onNext(2); 

source.onNext(3); 

source.onNext(4);

 source.onComplete();

// It will also get 1, 2, 3, 4 as we have used replay Subject 

source.subscribe(getSecondObserver());

Async Subject :

It only emits the last value of the source Observable(and only the last value) only after that source Observable completes.

Here, if a student entered at any point of time into the classroom, and he wants to listen only about the last thing(and only the last thing) being taught, after class is over. So, here we will use async.

See the below example:

AsyncSubject<Integer> source = AsyncSubject.create();

// It will get only 4 and onComplete 

source.subscribe(getFirstObserver());
source.onNext(1);

 source.onNext(2);

 source.onNext(3);

// It will also get only get 4 and onComplete 

source.subscribe(getSecondObserver());
source.onNext(4); 

source.onComplete();

Key Points to remember : 

• Subjects are observables that are also observers.
• You can send events over subjects by using onNext, onError and onComplete.
• PublishSubject is used when you only want to receive events that occur after you’ve subscribed.
• BehaviorSubject will relay the latest event that has occurred when you subscribe, including an optional initial value.
• ReplaySubject will buffer a configurable number of events that get replayed to new subscribers. You must watch out for buffering too much data in a replay subject.
• AsyncSubject only sends subscribers the most recent next event upon a complete event occurring.
• The RxRelay library can be used with relays in place of subjects, to prevent accidental complete and error events to be sent.

1.