How to convert a hot stream to a cold one and vice versa in RxJava?

Transforming hot streams into cold streams and vice versa in RxJava is achieved using the following operators:

Hot to Cold (e.g., from Subject)

The hide() operator is used. It returns an Observable that mimics cold behavior by hiding the source type.

// Hot stream (example with PublishSubject)
PublishSubject<Integer> hotStream = PublishSubject.create();

// Convert to cold stream
Observable<Integer> coldStream = hotStream.hide();

// Subscribe to the cold stream
coldStream.subscribe(value -> System.out.println("Cold: " + value));

// Emit data into the hot stream
hotStream.onNext(1);
hotStream.onNext(2);
hotStream.onNext(3);

Cold to Hot

Operators publish() and connect() are used. publish() converts a cold Observable into a ConnectableObservable, which does not start emitting data until connect() is called. Multiple subscribers before connect() will receive the same data.

// Cold stream
Observable<Integer> coldStream = Observable.range(1, 5)
        .doOnSubscribe(disposable -> System.out.println("Cold stream subscribed"));

// Convert to hot stream
ConnectableObservable<Integer> hotStream = coldStream.publish();

// Subscribers before connect() will receive the same data
hotStream.subscribe(value -> System.out.println("Subscriber 1: " + value));
hotStream.subscribe(value -> System.out.println("Subscriber 2: " + value));

// Start emitting data (becomes hot)
hotStream.connect();

// Subscriber after connect() may miss data emitted before subscription
hotStream.subscribe(value -> System.out.println("Subscriber 3: " + value));

You can also use the share() operator. It is a simplified version of publish().refCount(). It makes the Observable hot, but only starts emission when there is at least one subscriber and stops when there are none.

// Cold stream
Observable<Integer> coldStream = Observable.range(6, 5)
        .doOnSubscribe(disposable -> System.out.println("Cold stream subscribed (shared)"));

// Convert to hot stream using share()
Observable<Integer> hotStream = coldStream.share();

// Subscribers
hotStream.subscribe(value -> System.out.println("Shared Subscriber 1: " + value));
hotStream.subscribe(value -> System.out.println("Shared Subscriber 2: " + value));

// Data starts emitting when there are subscribers.

The cache() operator also makes the stream hot, but it caches all data emitted by the source and replays it to new subscribers.

// Cold stream
Observable<Integer> coldStream = Observable.range(11, 5)
        .doOnSubscribe(disposable -> System.out.println("Cold stream subscribed (cached)"));

// Convert to cached hot stream
Observable<Integer> hotStream = coldStream.cache();

// Subscribers
hotStream.subscribe(value -> System.out.println("Cached Subscriber 1: " + value)); // Will receive all data from the start
hotStream.subscribe(value -> System.out.println("Cached Subscriber 2: " + value)); // Will also receive all data from the start