2.x: replay() seems to be still replaying old data for late subscribers

[adevhere]

• Using RxJava 2.0.6

It seems this issue is still present in RxJava 2. When late subscribers subscribe() to a source with replay(time, unit) that has already expired, the source still emits the "expired emission" if onNext() is not called.

Posted this originally on StackOverflow but I'm now referencing it here because it seems to be an issue with the library.

[akarnokd]

Hi and thanks for reporting. Indeed this is a bug that was not resolved in 2.x:

io.reactivex.schedulers.TestScheduler scheduler = new io.reactivex.schedulers.TestScheduler();
        
io.reactivex.Observable<Integer> source = io.reactivex.Observable.just(3)
                .replay(2, TimeUnit.SECONDS, scheduler)
                .autoConnect();

source.test().assertResult(3);

source.test().assertResult(3);

scheduler.advanceTimeBy(3, TimeUnit.SECONDS);

source.test().assertResult();

The timed versions of Flowable.replay(), ReplayProcessor, Observable.replay() and ReplaySubject are all affected.

and appears to be partially broken in 1.x:

rx.schedulers.TestScheduler scheduler = new rx.schedulers.TestScheduler();
        
rx.Observable<Integer> source = rx.Observable.just(1)
                .replay(2, TimeUnit.SECONDS, scheduler)
                .autoConnect();

source.test().assertResult(1);

source.test().assertResult(1);

scheduler.advanceTimeBy(3, TimeUnit.SECONDS);

source.test().assertResult();

fails on the last line with Not completed!.

I'll post the fixes for the respective versions shortly.

[akarnokd]

Closing via #5140 & #5141.