This is not how replay/ cache works. Please read the #replay/ #cache documentation first.

replay

This operator returns a ConnectableObservable, which has some methods (#refCount/ #connect/ #autoConnect) for connecting to the source.

When #replay is applied without an overload, the source subscription is multicasted and all emitted values sind connection will be replayed. The source subscription is lazy and can connect to the source via #refCount/ #connect/ #autoConnect.

Returns a ConnectableObservable that shares a single subscription to the underlying ObservableSource that will replay all of its items and notifications to any future Observer.

Applying #relay without any connect-method (#refCount/ #connect/ #autoConnect) will not emit any values on subscription

A Connectable ObservableSource resembles an ordinary ObservableSource, except that it does not begin emitting items when it is subscribed to, but only when its connect method is called.

replay(1)#autoConnect(-1) / #refCount(1) / #connect

Applying replay(1) will cache the last value and will emit the cached value on each subscription. The #autoConnect will connect open an connection immediately and stay open until a terminal event (onComplete, onError) happens. #refCount is smiular, but will disconnect from the source, when all subscriber disappear. The #connect opreator can be used, when you need to wait, when alle subscriptions have been done to the observable, in order not to miss values.

usage

#replay(1) -- most of the it should be used at the end of the observable.

sourcObs.
  .filter()
  .map()
  .replay(bufferSize)
  .refCount(connectWhenXSubsciberSubscribed) 

caution

applying #replay without a buffer-limit or expiration date will lead to memory-leaks, when you observale is infinite

cache / cacheWithInitialCapacity

Operators are similar to #replay with autoConnect(1). The operators will cache every value and replay on each subsciption.

The operator subscribes only when the first downstream subscriber subscribes and maintains a single subscription towards this ObservableSource. In contrast, the operator family of replay() that return a ConnectableObservable require an explicit call to ConnectableObservable.connect(). Note: You sacrifice the ability to dispose the origin when you use the cache Observer so be careful not to use this Observer on ObservableSources that emit an infinite or very large number of items that will use up memory. A possible workaround is to apply takeUntil with a predicate or another source before (and perhaps after) the application of cache().

example

    @Test
    fun skfdsfkds() {
        val create = PublishSubject.create<Int>()

        val cacheWithInitialCapacity = create
            .cacheWithInitialCapacity(1)

        cacheWithInitialCapacity.subscribe()

        create.onNext(1)
        create.onNext(2)
        create.onNext(3)

        cacheWithInitialCapacity.test().assertValues(1, 2, 3)
        cacheWithInitialCapacity.test().assertValues(1, 2, 3)
    }

usage

Use cache operator, when you can not control the connect phase

This is useful when you want an ObservableSource to cache responses and you can't control the subscribe/dispose behavior of all the Observers.

caution

As with replay() the cache is unbounded and could lead to memory-leaks.

Note: The capacity hint is not an upper bound on cache size. For that, consider replay(int) in combination with ConnectableObservable.autoConnect() or similar.

further reading

https://blog.danlew.net/2018/09/25/connectable-observables-so-hot-right-now/

https://blog.danlew.net/2016/06/13/multicasting-in-rxjava/

If your event source (Observable) is an expensive operation, such as reading from a database, you shouldn't use Subject to observe the events, since that will repeat the expensive operation for each subscriber. Caching can also be risky with infinite streams due to "OutOfMemory" exceptions. A more appropriate solution may be ConnectableObservable, which only performs the source operation once, and broadcasts the updated value to all subscribers.

Here is a code sample. I didn't bother creating an infinite periodic stream or including error handling to keep the example simple. Let me know if it does what you need.

class RxJavaTest {

    private final int maxValue = 50;

    private final ConnectableObservable<Integer> source =
            Observable.<Integer>create(
                subscriber -> {
                    log("Starting Event Source");
                    subscriber.onNext(readFromDatabase());
                    subscriber.onNext(readFromDatabase());
                    subscriber.onNext(readFromDatabase());
                    subscriber.onComplete();
                    log("Event Source Terminated");
                })
                .subscribeOn(Schedulers.io())
                .filter(value -> value < maxValue)
                .publish();

    void run() throws InterruptedException {
        log("Starting Application");

        log("Subscribing");
        source.subscribe(value -> log("Subscriber 1: " + value));
        source.subscribe(value -> log("Subscriber 2: " + value));

        log("Connecting");
        source.connect();

        // Add sleep to give event source enough time to complete
        log("Application Terminated");
        sleep(4000);
    }

    private Integer readFromDatabase() throws InterruptedException {
        // Emulate long database read time
        log("Reading data from database...");
        sleep(1000);

        int randomValue = new Random().nextInt(2 * maxValue) + 1;
        log(String.format("Read value: %d", randomValue));
        return randomValue;
    }

    private static void log(Object message) {
        System.out.println(
                Thread.currentThread().getName() + " >> " + message
        );
    }
}

Here's the output:

main >> Starting Application
main >> Subscribing
main >> Connecting
main >> Application Terminated
RxCachedThreadScheduler-1 >> Starting Event Source
RxCachedThreadScheduler-1 >> Reading data from database...
RxCachedThreadScheduler-1 >> Read value: 88
RxCachedThreadScheduler-1 >> Reading data from database...
RxCachedThreadScheduler-1 >> Read value: 42
RxCachedThreadScheduler-1 >> Subscriber 1: 42
RxCachedThreadScheduler-1 >> Subscriber 2: 42
RxCachedThreadScheduler-1 >> Reading data from database...
RxCachedThreadScheduler-1 >> Read value: 37
RxCachedThreadScheduler-1 >> Subscriber 1: 37
RxCachedThreadScheduler-1 >> Subscriber 2: 37
RxCachedThreadScheduler-1 >> Event Source Terminated.

Note the following:

• Events only start firing once connect() is called on the source, not when observers subscribe to the source.
• Database calls are only made once per event update
• Filtered values are not emitted to subscribers
• All subscribers are executed in the same thread
• Application terminates before the events are processed due to concurrency. Normally your app will run in an event loop, so your app will remain responsive during slow operations.