Learning RxJava - Second Edition

Caching

When you want to cache all emissions indefinitely for the long term and do not need to control the subscription behavior to the source with ConnectableObservable, you can use the cache() operator. This will subscribe to the source on the first downstream Observer that subscribes and hold all values indefinitely. This makes it a candidate for an infinite Observable or a large amount of data that could exhaust the memory. The following is an example of ConnectableObservable usage:

import io.reactivex.rxjava3.core.Observable;public class Ch5_14 {    public static void main(String[] args) {        Observable<Integer> cachedRollingTotals =                Observable.just(6, 2, 5, 7, 1, 4, 9, 8, 3)                          .scan(0, (total, next) -> total + next)                          .cache();        cachedRollingTotals.subscribe(System. ...