3.x: Fix truncation bugs in replay() and ReplaySubject/Processor

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

@@ -767,6 +767,11 @@ final void removeFirst() {
             }
 
             setFirst(head);
+            // correct the tail if all items have been removed
+            head = get();
+            if (head.get() == null) {
+                tail = head;
+            }
         }
         /**
          * Arranges the given node is the new head from now on.
@@ -775,35 +780,31 @@ final void removeFirst() {
         final void setFirst(Node n) {
             if (eagerTruncate) {
                 Node m = new Node(null, n.index);
-                Node nextNode = n.get();
-                if (nextNode == null) {
-                    tail = m;
-                }
-                m.lazySet(nextNode);
+                m.lazySet(n.get());
                 n = m;
             }
             set(n);
         }
 
         @Override
         public final void next(T value) {
-            Object o = enterTransform(NotificationLite.next(value));
+            Object o = enterTransform(NotificationLite.next(value), false);
             Node n = new Node(o, ++index);
             addLast(n);
             truncate();
         }
 
         @Override
         public final void error(Throwable e) {
-            Object o = enterTransform(NotificationLite.error(e));
+            Object o = enterTransform(NotificationLite.error(e), true);
             Node n = new Node(o, ++index);
             addLast(n);
             truncateFinal();
         }
 
         @Override
         public final void complete() {
-            Object o = enterTransform(NotificationLite.complete());
+            Object o = enterTransform(NotificationLite.complete(), true);
             Node n = new Node(o, ++index);
             addLast(n);
             truncateFinal();
@@ -897,9 +898,10 @@ public final void replay(InnerSubscription<T> output) {
          * Override this to wrap the NotificationLite object into a
          * container to be used later by truncate.
          * @param value the value to transform into the internal representation
+         * @param terminal is this a terminal value?
          * @return the transformed value
          */
-        Object enterTransform(Object value) {
+        Object enterTransform(Object value, boolean terminal) {
             return value;
         }
         /**
@@ -1001,8 +1003,8 @@ static final class SizeAndTimeBoundReplayBuffer<T> extends BoundedReplayBuffer<T
         }
 
         @Override
-        Object enterTransform(Object value) {
-            return new Timed<Object>(value, scheduler.now(unit), unit);
+        Object enterTransform(Object value, boolean terminal) {
+            return new Timed<Object>(value, terminal ? Long.MAX_VALUE : scheduler.now(unit), unit);
         }
 
         @Override
@@ -1019,7 +1021,7 @@ void truncate() {
 
             int e = 0;
             for (;;) {
-                if (next != null) {
+                if (next != null && size > 1) { // never truncate the very last item just added
                     if (size > limit) {
                         e++;
                         size--;

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

@@ -633,6 +633,11 @@ final void trimHead() {
             }
 
             setFirst(head);
+            // correct the tail if all items have been removed
+            head = get();
+            if (head.get() == null) {
+                tail = head;
+            }
         }
         /**
          * Arranges the given node is the new head from now on.
@@ -641,11 +646,7 @@ final void trimHead() {
         final void setFirst(Node n) {
             if (eagerTruncate) {
                 Node m = new Node(null);
-                Node nextNode = n.get();
-                if (nextNode == null) {
-                    tail = m;
-                }
-                m.lazySet(nextNode);
+                m.lazySet(n.get());
                 n = m;
             }
             set(n);
@@ -845,7 +846,7 @@ void truncate() {
 
             int e = 0;
             for (;;) {
-                if (next != null) {
+                if (next != null && size > 1) { // never truncate the very last item just added
                     if (size > limit) {
                         e++;
                         size--;

src/main/java/io/reactivex/processors/ReplayProcessor.java

@@ -1070,6 +1070,10 @@ void trim() {
             TimedNode<T> h = head;
 
             for (;;) {
+                if (size <= 1) {
+                    head = h;
+                    break;
+                }
                 TimedNode<T> next = h.get();
                 if (next == null) {
                     head = h;
@@ -1082,6 +1086,7 @@ void trim() {
                 }
 
                 h = next;
+                size--;
             }
 
         }

src/main/java/io/reactivex/subjects/ReplaySubject.java

@@ -1071,6 +1071,10 @@ void trim() {
             TimedNode<Object> h = head;
 
             for (;;) {
+                if (size <= 1) {
+                    head = h;
+                    break;
+                }
                 TimedNode<Object> next = h.get();
                 if (next == null) {
                     head = h;
@@ -1083,6 +1087,7 @@ void trim() {
                 }
 
                 h = next;
+                size--;
             }
 
         }

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

@@ -28,7 +28,7 @@
 import io.reactivex.*;
 import io.reactivex.Scheduler.Worker;
 import io.reactivex.annotations.NonNull;
-import io.reactivex.disposables.Disposable;
+import io.reactivex.disposables.*;
 import io.reactivex.exceptions.TestException;
 import io.reactivex.flowables.ConnectableFlowable;
 import io.reactivex.functions.*;
@@ -2258,4 +2258,13 @@ public void timeAndSizeBoundSelectorEagerTruncate() throws Exception {
                     + " -> " + after / 1024.0 / 1024.0);
         }
     }
+
+    @Test
+    public void timeAndSizeNoTerminalTruncationOnTimechange() {
+        Flowable.just(1).replay(1, 1, TimeUnit.SECONDS, new TimesteppingScheduler(), true)
+        .autoConnect()
+        .test()
+        .assertComplete()
+        .assertNoErrors();
+    }
 }

src/test/java/io/reactivex/internal/operators/observable/ObservableReplayEagerTruncateTest.java

@@ -14,6 +14,7 @@
 package io.reactivex.internal.operators.observable;
 
 import static org.junit.Assert.*;
+import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 
 import java.lang.management.*;
@@ -1997,4 +1998,12 @@ public void timeAndSizeSelectorBoundEagerTruncate() throws Exception {
         }
     }
 
+    @Test
+    public void timeAndSizeNoTerminalTruncationOnTimechange() {
+        Observable.just(1).replay(1, 1, TimeUnit.SECONDS, new TimesteppingScheduler(), true)
+        .autoConnect()
+        .test()
+        .assertComplete()
+        .assertNoErrors();
+    }
 }

src/test/java/io/reactivex/processors/ReplayProcessorTest.java

@@ -14,6 +14,7 @@
 package io.reactivex.processors;
 
 import static org.junit.Assert.*;
+import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 
 import java.lang.management.*;
@@ -25,15 +26,15 @@
 import org.mockito.*;
 import org.reactivestreams.*;
 
-import io.reactivex.*;
+import io.reactivex.Flowable;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.TestException;
 import io.reactivex.functions.*;
 import io.reactivex.internal.subscriptions.BooleanSubscription;
 import io.reactivex.processors.ReplayProcessor.*;
 import io.reactivex.schedulers.*;
 import io.reactivex.subscribers.*;
-import io.reactivex.testsupport.TestHelper;
+import io.reactivex.testsupport.*;
 
 public class ReplayProcessorTest extends FlowableProcessorTest<Object> {
 
@@ -1751,4 +1752,79 @@ public void accept(byte[] v) throws Exception {
                     + " -> " + after.get() / 1024.0 / 1024.0);
         }
     }
+
+    @Test
+    public void timeAndSizeNoTerminalTruncationOnTimechange() {
+        ReplayProcessor<Integer> rp = ReplayProcessor.createWithTimeAndSize(1, TimeUnit.SECONDS, new TimesteppingScheduler(), 1);
+
+        TestSubscriber<Integer> ts = rp.test();
+
+        rp.onNext(1);
+        rp.cleanupBuffer();
+        rp.onComplete();
+
+        ts.assertNoErrors()
+        .assertComplete();
+    }
+
+    @Test
+    public void timeAndSizeNoTerminalTruncationOnTimechange2() {
+        ReplayProcessor<Integer> rp = ReplayProcessor.createWithTimeAndSize(1, TimeUnit.SECONDS, new TimesteppingScheduler(), 1);
+
+        TestSubscriber<Integer> ts = rp.test();
+
+        rp.onNext(1);
+        rp.cleanupBuffer();
+        rp.onNext(2);
+        rp.cleanupBuffer();
+        rp.onComplete();
+
+        ts.assertNoErrors()
+        .assertComplete();
+    }
+
+    @Test
+    public void timeAndSizeNoTerminalTruncationOnTimechange3() {
+        ReplayProcessor<Integer> rp = ReplayProcessor.createWithTimeAndSize(1, TimeUnit.SECONDS, new TimesteppingScheduler(), 1);
+
+        TestSubscriber<Integer> ts = rp.test();
+
+        rp.onNext(1);
+        rp.onNext(2);
+        rp.onComplete();
+
+        ts.assertNoErrors()
+        .assertComplete();
+    }
+
+    @Test
+    public void timeAndSizeNoTerminalTruncationOnTimechange4() {
+        ReplayProcessor<Integer> rp = ReplayProcessor.createWithTimeAndSize(1, TimeUnit.SECONDS, new TimesteppingScheduler(), 10);
+
+        TestSubscriber<Integer> ts = rp.test();
+
+        rp.onNext(1);
+        rp.onNext(2);
+        rp.onComplete();
+
+        ts.assertNoErrors()
+        .assertComplete();
+    }
+
+    @Test
+    public void timeAndSizeRemoveCorrectNumberOfOld() {
+        TestScheduler scheduler = new TestScheduler();
+        ReplayProcessor<Integer> rp = ReplayProcessor.createWithTimeAndSize(1, TimeUnit.SECONDS, scheduler, 2);
+
+        rp.onNext(1);
+        rp.onNext(2);
+        rp.onNext(3);
+
+        scheduler.advanceTimeBy(2, TimeUnit.SECONDS);
+
+        rp.onNext(4);
+        rp.onNext(5);
+
+        rp.test().assertValuesOnly(4, 5);
+    }
 }