Как преобразовать горячий поток в холодный и наоборот в RxJava?

Преобразование горячего потока в холодный и наоборот в RxJava достигается с помощью операторов:

Горячий в Холодный (например, из Subject)

Используется оператор hide(). Он возвращает Observable, который имитирует холодное поведение, скрывая тип источника.

// Горячий поток (пример с PublishSubject)
PublishSubject<Integer> hotStream = PublishSubject.create();

// Преобразование в холодный поток
Observable<Integer> coldStream = hotStream.hide();

// Подписка на холодный поток
coldStream.subscribe(value -> System.out.println("Cold: " + value));

// Эмиссия данных в горячий поток
hotStream.onNext(1);
hotStream.onNext(2);
hotStream.onNext(3);

Холодный в Горячий

Используются операторы publish() и connect(). publish() преобразует холодный Observable в ConnectableObservable, который не начинает эмиссию данных до вызова connect(). Множество подписчиков до вызова connect() будут получать одни и те же данные.

// Холодный поток
Observable<Integer> coldStream = Observable.range(1, 5)
        .doOnSubscribe(disposable -> System.out.println("Cold stream subscribed"));

// Преобразование в горячий поток
ConnectableObservable<Integer> hotStream = coldStream.publish();

// Подписчики до connect() будут получать те же данные
hotStream.subscribe(value -> System.out.println("Subscriber 1: " + value));
hotStream.subscribe(value -> System.out.println("Subscriber 2: " + value));

// Начать эмиссию данных (становится горячим)
hotStream.connect();

// Подписчик после connect() может пропустить данные, эмитированные до его подписки
hotStream.subscribe(value -> System.out.println("Subscriber 3: " + value));

Также можно использовать оператор share(). Это упрощенный вариант сочетания publish().refCount(). Он делает Observable горячим, но начинает эмиссию только при наличии хотя бы одного подписчика и останавливает, когда подписчиков нет.

// Холодный поток
Observable<Integer> coldStream = Observable.range(6, 5)
        .doOnSubscribe(disposable -> System.out.println("Cold stream subscribed (shared)"));

// Преобразование в горячий поток с помощью share()
Observable<Integer> hotStream = coldStream.share();

// Подписчики
hotStream.subscribe(value -> System.out.println("Shared Subscriber 1: " + value));
hotStream.subscribe(value -> System.out.println("Shared Subscriber 2: " + value));

// Данные начинают эмитироваться, когда есть подписчики.

Оператор cache() также делает поток горячим, но он также кэширует все данные, эмитированные источником, и ретранслирует их новым подписчикам.

// Холодный поток
Observable<Integer> coldStream = Observable.range(11, 5)
        .doOnSubscribe(disposable -> System.out.println("Cold stream subscribed (cached)"));

// Преобразование в горячий поток с кэшированием
Observable<Integer> hotStream = coldStream.cache();

// Подписчики
hotStream.subscribe(value -> System.out.println("Cached Subscriber 1: " + value)); // Получит все данные с начала
hotStream.subscribe(value -> System.out.println("Cached Subscriber 2: " + value)); // Получит все данные с начала