3.x: Fix replay() not resetting when the connection is disposed

src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableReplay.java

@@ -179,7 +179,7 @@ public void connect(Consumer<? super Disposable> connection) {
  }
 
  // create a new subscriber-to-source
- ReplaySubscriber<T> u = new ReplaySubscriber<>(buf);
+ ReplaySubscriber<T> u = new ReplaySubscriber<>(buf, current);
  // try setting it as the current subscriber-to-source
  if (!current.compareAndSet(ps, u)) {
  // did not work, perhaps a new subscriber arrived
@@ -249,9 +249,13 @@ static final class ReplaySubscriber<T>
  /** Tracks the amount already requested from the upstream. */
  long requestedFromUpstream;
 
+ /** The current connection. */
+ final AtomicReference<ReplaySubscriber<T>> current;
+
  @SuppressWarnings("unchecked")
- ReplaySubscriber(ReplayBuffer<T> buffer) {
+ ReplaySubscriber(ReplayBuffer<T> buffer, AtomicReference<ReplaySubscriber<T>> current) {
  this.buffer = buffer;
+ this.current = current;
  this.management = new AtomicInteger();
  this.subscribers = new AtomicReference<>(EMPTY);
  this.shouldConnect = new AtomicBoolean();
@@ -266,9 +270,7 @@ public boolean isDisposed() {
  @Override
  public void dispose() {
  subscribers.set(TERMINATED);
- // unlike OperatorPublish, we can't null out the terminated so
- // late subscribers can still get replay
- // current.compareAndSet(ReplaySubscriber.this, null);
+ current.compareAndSet(ReplaySubscriber.this, null);
  // we don't care if it fails because it means the current has
  // been replaced in the meantime
  SubscriptionHelper.cancel(this);
@@ -1198,7 +1200,7 @@ public void subscribe(Subscriber<? super T> child) {
  return;
  }
  // create a new subscriber to source
- ReplaySubscriber<T> u = new ReplaySubscriber<>(buf);
+ ReplaySubscriber<T> u = new ReplaySubscriber<>(buf, curr);
  // let's try setting it as the current subscriber-to-source
  if (!curr.compareAndSet(null, u)) {
  // didn't work, maybe someone else did it or the current subscriber

src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableReplay.java

@@ -174,7 +174,7 @@ public void connect(Consumer<? super Disposable> connection) {
  // create a new subscriber-to-source
  ReplayBuffer<T> buf = bufferFactory.call();
 
- ReplayObserver<T> u = new ReplayObserver<>(buf);
+ ReplayObserver<T> u = new ReplayObserver<>(buf, current);
  // try setting it as the current subscriber-to-source
  if (!current.compareAndSet(ps, u)) {
  // did not work, perhaps a new subscriber arrived
@@ -240,8 +240,12 @@ static final class ReplayObserver<T>
  */
  final AtomicBoolean shouldConnect;
 
- ReplayObserver(ReplayBuffer<T> buffer) {
+ /** The current connection. */
+ final AtomicReference<ReplayObserver<T>> current;
+
+ ReplayObserver(ReplayBuffer<T> buffer, AtomicReference<ReplayObserver<T>> current) {
  this.buffer = buffer;
+ this.current = current;
 
  this.observers = new AtomicReference<>(EMPTY);
  this.shouldConnect = new AtomicBoolean();
@@ -255,9 +259,7 @@ public boolean isDisposed() {
  @Override
  public void dispose() {
  observers.set(TERMINATED);
- // unlike OperatorPublish, we can't null out the terminated so
- // late observers can still get replay
- // current.compareAndSet(ReplayObserver.this, null);
+ current.compareAndSet(ReplayObserver.this, null);
  // we don't care if it fails because it means the current has
  // been replaced in the meantime
  DisposableHelper.dispose(this);
@@ -1004,7 +1006,7 @@ public void subscribe(Observer<? super T> child) {
  // create a new subscriber to source
  ReplayBuffer<T> buf = bufferFactory.call();
 
- ReplayObserver<T> u = new ReplayObserver<>(buf);
+ ReplayObserver<T> u = new ReplayObserver<>(buf, curr);
  // let's try setting it as the current subscriber-to-source
  if (!curr.compareAndSet(null, u)) {
  // didn't work, maybe someone else did it or the current subscriber

src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowablePublishTest.java

@@ -1776,4 +1776,31 @@ public void onError(Throwable t) {
 
  ts1.assertEmpty();
  }
+
+ @Test
+ public void disposeNoNeedForReset() {
+ PublishProcessor<Integer> pp = PublishProcessor.create();
+
+ ConnectableFlowable<Integer> cf = pp.publish();
+
+ TestSubscriber<Integer> ts = cf.test();
+
+ Disposable d = cf.connect();
+
+ pp.onNext(1);
+
+ d.dispose();
+
+ ts = cf.test();
+
+ ts.assertEmpty();
+
+ cf.connect();
+
+ ts.assertEmpty();
+
+ pp.onNext(2);
+
+ ts.assertValuesOnly(2);
+ }
 }

src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableReplayEagerTruncateTest.java

@@ -1923,19 +1923,6 @@ public ReplayBuffer<Integer> get() throws Exception {
  .assertFailure(TestException.class);
  }
 
- @Test
- public void currentDisposedWhenConnecting() {
- FlowableReplay<Integer> fr = (FlowableReplay<Integer>)FlowableReplay.create(Flowable.<Integer>never(), 16, true);
- fr.connect();
-
- fr.current.get().dispose();
- assertTrue(fr.current.get().isDisposed());
-
- fr.connect();
-
- assertFalse(fr.current.get().isDisposed());
- }
-
  @Test
  public void noBoundedRetentionViaThreadLocal() throws Exception {
  Flowable<byte[]> source = Flowable.range(1, 200)
@@ -2275,4 +2262,85 @@ public void timeAndSizeNoTerminalTruncationOnTimechange() {
  .assertComplete()
  .assertNoErrors();
  }
+
+ @Test
+ public void disposeNoNeedForResetSizeBound() {
+ PublishProcessor<Integer> pp = PublishProcessor.create();
+
+ ConnectableFlowable<Integer> cf = pp.replay(10, true);
+
+ TestSubscriber<Integer> ts = cf.test();
+
+ Disposable d = cf.connect();
+
+ pp.onNext(1);
+
+ d.dispose();
+
+ ts = cf.test();
+
+ ts.assertEmpty();
+
+ cf.connect();
+
+ ts.assertEmpty();
+
+ pp.onNext(2);
+
+ ts.assertValuesOnly(2);
+ }
+
+ @Test
+ public void disposeNoNeedForResetTimeBound() {
+ PublishProcessor<Integer> pp = PublishProcessor.create();
+
+ ConnectableFlowable<Integer> cf = pp.replay(10, TimeUnit.MINUTES, Schedulers.single(), true);
+
+ TestSubscriber<Integer> ts = cf.test();
+
+ Disposable d = cf.connect();
+
+ pp.onNext(1);
+
+ d.dispose();
+
+ ts = cf.test();
+
+ ts.assertEmpty();
+
+ cf.connect();
+
+ ts.assertEmpty();
+
+ pp.onNext(2);
+
+ ts.assertValuesOnly(2);
+ }
+
+ @Test
+ public void disposeNoNeedForResetTimeAndSIzeBound() {
+ PublishProcessor<Integer> pp = PublishProcessor.create();
+
+ ConnectableFlowable<Integer> cf = pp.replay(10, 10, TimeUnit.MINUTES, Schedulers.single(), true);
+
+ TestSubscriber<Integer> ts = cf.test();
+
+ Disposable d = cf.connect();
+
+ pp.onNext(1);
+
+ d.dispose();
+
+ ts = cf.test();
+
+ ts.assertEmpty();
+
+ cf.connect();
+
+ ts.assertEmpty();
+
+ pp.onNext(2);
+
+ ts.assertValuesOnly(2);
+ }
 }

src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableReplayTest.java

@@ -1988,19 +1988,6 @@ public ReplayBuffer<Integer> get() throws Exception {
  .assertFailure(TestException.class);
  }
 
- @Test
- public void currentDisposedWhenConnecting() {
- FlowableReplay<Integer> fr = (FlowableReplay<Integer>)FlowableReplay.create(Flowable.<Integer>never(), 16, false);
- fr.connect();
-
- fr.current.get().dispose();
- assertTrue(fr.current.get().isDisposed());
-
- fr.connect();
-
- assertFalse(fr.current.get().isDisposed());
- }
-
  @Test
  public void noBoundedRetentionViaThreadLocal() throws Exception {
  Flowable<byte[]> source = Flowable.range(1, 200)
@@ -2191,4 +2178,112 @@ public void cancel() {
 
  ts.assertResult();
  }
+
+ @Test
+ public void disposeNoNeedForReset() {
+ PublishProcessor<Integer> pp = PublishProcessor.create();
+
+ ConnectableFlowable<Integer> cf = pp.replay();
+
+ TestSubscriber<Integer> ts = cf.test();
+
+ Disposable d = cf.connect();
+
+ pp.onNext(1);
+
+ d.dispose();
+
+ ts = cf.test();
+
+ ts.assertEmpty();
+
+ cf.connect();
+
+ ts.assertEmpty();
+
+ pp.onNext(2);
+
+ ts.assertValuesOnly(2);
+ }
+
+ @Test
+ public void disposeNoNeedForResetSizeBound() {
+ PublishProcessor<Integer> pp = PublishProcessor.create();
+
+ ConnectableFlowable<Integer> cf = pp.replay(10);
+
+ TestSubscriber<Integer> ts = cf.test();
+
+ Disposable d = cf.connect();
+
+ pp.onNext(1);
+
+ d.dispose();
+
+ ts = cf.test();
+
+ ts.assertEmpty();
+
+ cf.connect();
+
+ ts.assertEmpty();
+
+ pp.onNext(2);
+
+ ts.assertValuesOnly(2);
+ }
+
+ @Test
+ public void disposeNoNeedForResetTimeBound() {
+ PublishProcessor<Integer> pp = PublishProcessor.create();
+
+ ConnectableFlowable<Integer> cf = pp.replay(10, TimeUnit.MINUTES);
+
+ TestSubscriber<Integer> ts = cf.test();
+
+ Disposable d = cf.connect();
+
+ pp.onNext(1);
+
+ d.dispose();
+
+ ts = cf.test();
+
+ ts.assertEmpty();
+
+ cf.connect();
+
+ ts.assertEmpty();
+
+ pp.onNext(2);
+
+ ts.assertValuesOnly(2);
+ }
+
+ @Test
+ public void disposeNoNeedForResetTimeAndSIzeBound() {
+ PublishProcessor<Integer> pp = PublishProcessor.create();
+
+ ConnectableFlowable<Integer> cf = pp.replay(10, 10, TimeUnit.MINUTES);
+
+ TestSubscriber<Integer> ts = cf.test();
+
+ Disposable d = cf.connect();
+
+ pp.onNext(1);
+
+ d.dispose();
+
+ ts = cf.test();
+
+ ts.assertEmpty();
+
+ cf.connect();
+
+ ts.assertEmpty();
+
+ pp.onNext(2);
+
+ ts.assertValuesOnly(2);
+ }
 }

src/test/java/io/reactivex/rxjava3/internal/operators/observable/ObservablePublishTest.java

@@ -866,4 +866,31 @@ public void disposeResets() {
 
  to.assertValuesOnly(1);
  }
+
+ @Test
+ public void disposeNoNeedForReset() {
+ PublishSubject<Integer> ps = PublishSubject.create();
+
+ ConnectableObservable<Integer> co = ps.publish();
+
+ TestObserver<Integer> to = co.test();
+
+ Disposable d = co.connect();
+
+ ps.onNext(1);
+
+ d.dispose();
+
+ to = co.test();
+
+ to.assertEmpty();
+
+ co.connect();
+
+ to.assertEmpty();
+
+ ps.onNext(2);
+
+ to.assertValuesOnly(2);
+ }
 }