The Subject instances are both Observable instances and Observer instances. Like Observable instances, they can have multiple Observer instances, receiving the same notifications. That's why they can be used to turn cold Observable instances into hot ones. Like Observer instances, they give us access to their onNext(), onError(), or onCompleted() methods.

Let's look at an implementation of the preceding hot interval examples, using a Subject instance:

Observable<Long> interval = Observable.interval(100L, TimeUnit.MILLISECONDS); // (1)
Subject<Long, Long> publishSubject = PublishSubject.create(); // (2)
interval.subscribe(publishSubject);
// (3)
Subscription sub1 = subscribePrint(publishSubject, "First");
Subscription sub2 = subscribePrint(publishSubject, "Second");
Subscription sub3 = null;
try {
  Thread.sleep(300L);
  publishSubject.onNext(555L); // (4)
  sub3 = subscribePrint(publishSubject, "Third"); // (5)
  Thread.sleep(500L);
}
catch (InterruptedException e) {}
sub1.unsubscribe(); // (6)
sub2.unsubscribe();
sub3.unsubscribe();

The example is slightly different now:

There are four types of subjects that come with RxJava:

Using subjects may seem a cool way to solve various problems, but you should avoid using them. Or, at least implement them and their behavior in a method that returns a result of type Observable.

The danger with the Subject instance is that they give access to the onNext(), onError(), and onCompleted() methods, and your logic can get messy (they need to be called following the Rx contract, cited earlier in this chapter). They can be misused very easily.

Opt for using the ConnecatableObservable instance (that is, via the publish() method) over the Subject, when you need to create a hot Observable from a cold one.

But let's look at one good use of a Subject instance—the aforementioned reactive properties. Again, we are going to implement 'The Reactive Sum', but this time it will be quite different. Here is the class defining it:

public class ReactiveSum { // (1)
  private BehaviorSubject<Double> a = BehaviorSubject.create(0.0);
  private BehaviorSubject<Double> b = BehaviorSubject.create(0.0);
  private BehaviorSubject<Double> c = BehaviorSubject.create(0.0);
  public ReactiveSum() { // (2)
    Observable.combineLatest(a, b, (x, y) -> x + y).subscribe(c);
  }
  public double getA() { // (3)
    return a.getValue();
  }
  public void setA(double a) {
    this.a.onNext(a);
  }
  public double getB() {
    return b.getValue();
  }
  public void setB(double b) {
    this.b.onNext(b);
  }
  public double getC() { // (4)
    return c.getValue();
  }
  public Observable<Double> obsC() {
    return c.asObservable();
  }
}

This class has three double properties: two settable properties, a and b, and their sum, c. When a or b changes, c is automatically updated to their sum. There is a special method that we can use to track the changes to c. So how does it work?

This ReactiveSum class can be used like this:

ReactiveSum sum = new ReactiveSum();
subscribePrint(sum.obsC(), "Sum");
sum.setA(5);
sum.setB(4);

This will output the following:

Sum : 0.0
Sum : 5.0
Sum : 9.0

The first value is emitted on the subscribe () method (remember the BehaviorSubject instances always emit their last value on subscribing), and the other two will automatically be emitted on setting a or b.

Reactive properties can be used for implementing bindings and counters, so they are very useful for desktop or browser applications. But this example is far from any functional paradigm.