ConnectableObservable

A helpful form of hot Observable is ConnectableObservable. It will take any Observable, even if it is cold, and make it hot so that all emissions are played to all Observers at once. To do this conversion, you simply need to call publish() on any Observable, and it will yield a ConnectableObservable. But subscribing will not start the emissions yet. You need to call its connect() method to start firing the emissions. This allows you to set up all your Observers beforehand. Take a look at the following code snippet:

    import io.reactivex.Observable;
    import io.reactivex.observables.ConnectableObservable;

    public class Launcher {
      public static void main(String[] args) {

        ConnectableObservable<String> source =
                    
         Observable.just("Alpha","Beta","Gamma","Delta","Epsilon")
         .publish();

         //Set up observer 1
         source.subscribe(s -> System.out.println("Observer 1: " + s));

        //Set up observer 2
        source.map(String::length)
          .subscribe(i -> System.out.println("Observer 2: " + i));

        //Fire!
        source.connect();
      }
}

Take a look at the following code:

    Observer 1: Alpha
    Observer 2: 5
    Observer 1: Beta
    Observer 2: 4
    Observer 1: Gamma
    Observer 2: 5
    Observer 1: Delta
    Observer 2: 5
    Observer 1: Epsilon
    Observer 2: 7

Note how one Observer is receiving the string while the other is receiving the length and the two are printing them in an interleaved fashion. Both subscriptions are set up beforehand, and then connect() is called to fire the emissions. Rather than Observer 1 processing all the emissions before Observer 2, each emission goes to each Observer simultaneously. Observer 1 receives Alpha and Observer 2 receives 5 and then Beta and 4, and so on. Using ConnectableObservable to force each emission to go to all Observers simultaneously is known as multicasting, which we will cover in detail in Chapter 5, Multicasting.

ConnectableObservable is helpful in preventing the replay of data to each Observer. You may want to do this if replaying emissions is expensive and you would rather emit them to all Observers at once. You may also do it simply to force the operators upstream to use a single stream instance even if there are multiple Observers downstream. Multiple Observers normally result in multiple stream instances upstream, but using publish() to return ConnectableObservable consolidates all the upstream operations before publish() into a single stream. Again, these nuances will be covered more in Chapter 5, Multicasting.

For now, remember that ConnectableObservable is hot, and therefore, if new subscriptions occur after connect() is called, they will miss emissions that were fired previously.