Hot and cold rx-java Observable

My own understanding of Hot and Cold Observable is quite shaky, but here is what I have understood till now!

Cold Observable

Consider an API which returns an rx-java Observable:

import obs.Util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rx.Observable;
import rx.schedulers.Schedulers;

public class Service1 {
    private static final Logger logger = LoggerFactory.getLogger(Service1.class);
    public Observable<String> operation() {
        return Observable.<String>create(s -> {
            logger.info("Start: Executing slow task in Service 1");
            Util.delay(1000);
            s.onNext("data 1");
            logger.info("End: Executing slow task in Service 1");
            s.onCompleted();
        }).subscribeOn(Schedulers.computation());
    }
}

Now, the first thing to note is that the typical Observable does not do anything until it is subscribed to:

So essentially if I were to do this:

Observable<String> op1 = service1.operation();

Nothing would be printed or returned, unless there is a subscription on the Observable this way:

Observable<String> op1 = service1.operation();

CountDownLatch latch = new CountDownLatch(1);

op1.subscribe(s -> logger.info("From Subscriber 1: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());

latch.await();

So now, what happens if there are multiple subscriptions on this Observable:

Observable<String> op1 = service1.operation();

CountDownLatch latch = new CountDownLatch(3);

op1.subscribe(s -> logger.info("From Subscriber 1: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());

op1.subscribe(s -> logger.info("From Subscriber 2: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());

op1.subscribe(s -> logger.info("From Subscriber 3: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());

latch.await();

With a cold observable the code would get called once more and the items emitted again, I get this on my machine:

06:04:07.206 [RxComputationThreadPool-2] INFO  o.b.Service1 - Start: Executing slow task in Service 1
06:04:07.208 [RxComputationThreadPool-3] INFO  o.b.Service1 - Start: Executing slow task in Service 1
06:04:08.211 [RxComputationThreadPool-2] INFO  o.b.BasicObservablesTest - From Subscriber 2: data 1
06:04:08.211 [RxComputationThreadPool-1] INFO  o.b.BasicObservablesTest - From Subscriber 1: data 1
06:04:08.211 [RxComputationThreadPool-3] INFO  o.b.BasicObservablesTest - From Subscriber 3: data 1
06:04:08.213 [RxComputationThreadPool-2] INFO  o.b.Service1 - End: Executing slow task in Service 1
06:04:08.214 [RxComputationThreadPool-1] INFO  o.b.Service1 - End: Executing slow task in Service 1
06:04:08.214 [RxComputationThreadPool-3] INFO  o.b.Service1 - End: Executing slow task in Service 1

Hot Observable - using ConnectableObservable

Hot Observable on the other hand does not really need a subscription to start emitting items. A way to implement a Hot Observable is using a ConnectableObservable, which is a Observable which does not emit items until its connect method is called, however once it starts emitting items, any subscriber to it gets items only from the point of subscription. So again revisiting the previous example, but with a ConnectableObservable instead:

Observable<String> op1 = service1.operation();

ConnectableObservable<String> connectableObservable =  op1.publish();

CountDownLatch latch = new CountDownLatch(3);

connectableObservable.subscribe(s -> logger.info("From Subscriber 1: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());

connectableObservable.subscribe(s -> logger.info("From Subscriber 2: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());

connectableObservable.subscribe(s -> logger.info("From Subscriber 3: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());

connectableObservable.connect();

latch.await();

and the following gets printed:

06:07:23.852 [RxComputationThreadPool-3] INFO  o.b.Service1 - Start: Executing slow task in Service 1
06:07:24.860 [RxComputationThreadPool-3] INFO  o.b.ConnectableObservablesTest - From Subscriber 1: data 1
06:07:24.862 [RxComputationThreadPool-3] INFO  o.b.ConnectableObservablesTest - From Subscriber 2: data 1
06:07:24.862 [RxComputationThreadPool-3] INFO  o.b.ConnectableObservablesTest - From Subscriber 3: data 1
06:07:24.862 [RxComputationThreadPool-3] INFO  o.b.Service1 - End: Executing slow task in Service 1

Hot Observable - using Subject

Another way to convert a cold Observable to a hot one is to use a Subject. Subjects behave both as an Observable and an Observer, there are different types of Subjects available with different behavior. Here I am using a Subject called a PublishSubject which has a Pub/Sub behavior - the items get emitted to all the subscribers listening on it. So with a PublishSubject introduced the code looks like this:

Observable<String> op1 = service1.operation();

PublishSubject<String> publishSubject = PublishSubject.create();

op1.subscribe(publishSubject);

CountDownLatch latch = new CountDownLatch(3);

publishSubject.subscribe(s -> logger.info("From Subscriber 1: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());

publishSubject.subscribe(s -> logger.info("From Subscriber 2: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());

publishSubject.subscribe(s -> logger.info("From Subscriber 3: {}", s),
        e -> logger.error(e.getMessage(), e),
        () -> latch.countDown());


latch.await();

See how the PublishSubject is introduced as a subscriber to the Observable and the other subscribers subscribe to the PublishSubject instead. The output will be similar to the one from ConnectableObservable.

This is essentially it, the extent of my understanding of Hot Observable. So to conclude, the difference between a Cold and a Hot Observable is about when the subscribers get the emitted items and when the items are emitted - with a Cold Observable they are emitted when they are subscribed to and typically get all the emitted items, with a Hot Observable the items are emitted without a Subscriber and subscribers get items emitted after the point of subscription typically.

Reference

1. http://www.introtorx.com/content/v1.0.10621.0/14_HotAndColdObservables.html
2. Excellent javadoc on rx-java - http://reactivex.io/RxJava/javadoc/index.html