Cold Observables

Cold Observables are much like a music CD that can be replayed to each listener, so each person can hear all the tracks at any time. In the same manner, cold Observables will replay the emissions to each Observer, ensuring that all Observers get all the data. Most data-driven Observables are cold, and this includes the Observable.just() and Observable.fromIterable() factories.

In the following example, we have two Observers subscribed to one Observable. The Observable will first play all the emissions to the first Observer and then call onComplete(). Then, it will play all the emissions again to the second Observer and call onComplete(). They both receive the same datasets by getting two separate streams each, which is typical behavior for a cold Observable:

      import io.reactivex.Observable;

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

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

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

        //second observer
        source.subscribe(s -> System.out.println("Observer 2 Received:
          " + s));

      }
    }

The output is as follows:

    Observer 1 Received: Alpha
    Observer 1 Received: Beta
    Observer 1 Received: Gamma
    Observer 1 Received: Delta
    Observer 1 Received: Epsilon
    Observer 2 Received: Alpha
    Observer 2 Received: Beta
    Observer 2 Received: Gamma
    Observer 2 Received: Delta
    Observer 2 Received: Epsilon

Even if the second Observer transforms its emissions with operators, it will still get its own stream of emissions. Using operators such as map() and filter() against a cold Observable will still maintain the cold nature of the yielded Observables:

    import io.reactivex.Observable;

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

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

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

        //second observer
        source.map(String::length).filter(i -> i >= 5)
          .subscribe(s -> System.out.println("Observer 2 Received: " + 
        s));

      }
   }

The output is as follows:

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

As stated earlier, Observable sources that emit finite datasets are usually cold.

Here is a more real-world example: Dave Moten's RxJava-JDBC (https://github.com/davidmoten/rxjava-jdbc) allows you to create cold Observables built off of SQL database queries. We will not digress into this library for too long, but if you want to query a SQLite database, for instance, include the SQLite JDBC driver and RxJava-JDBC libraries in your project. You can then query a database table reactively, as shown in the following code snippet:

    import com.github.davidmoten.rx.jdbc.ConnectionProviderFromUrl;
    import com.github.davidmoten.rx.jdbc.Database;
    import rx.Observable;
    import java.sql.Connection;

    public class Launcher {
      public static void main(String[] args) {
        Connection conn = 
      new ConnectionProviderFromUrl("jdbc:sqlite:/home/thomas 
            /rexon_metals.db").get();
        
      Database db = Database.from(conn);

        Observable<String> customerNames =
          db.select("SELECT NAME FROM CUSTOMER")
          .getAs(String.class);

        customerNames.subscribe(s -> System.out.println(s));
      }
   }

The output is as follows:

    LITE Industrial
    Rex Tooling Inc
    Re-Barre Construction
    Prairie Construction
    Marsh Lane Metal Works

This SQL-driven Observable is cold. Many Observables emitting from finite data sources such as databases, text files, or JSON are cold. It is still important to note how the source Observable is architected. RxJava-JDBC will run the query each time for each Observer. This means that if the data changes in between two subscriptions, the second Observer will get different emissions than the first one. But the Observable is still cold since it is replaying the query even if the resulting data changes from the underlying tables.

Again, cold Observables will, in some shape or form, repeat the operation to generate these emissions to each Observer. Next, we will cover hot Observables that resemble events more than data.