Prevent some additional memory leaks

I took a look at merging the `collect()` and `collectValuesInMemory()`
methods into a single method at the prompting of akarnokd. This made me
realize the following:
1) There are more memory leaks. I updated the setNext() method to fix some of these.
2) We can't avoid all these memory leaks without introducing schduling
because of the TimedBuffer. I don't believe engineers should have any
expectation that the TimedBuffer values will be eagerly cleaned up after
they timeout. So I left the seperate `collect()` and
`collectValuesInMemory()` methods. This allows us to test memory for
memory leaks only when we think it is appropriate.

Author: Brian Attwell

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

@@ -616,31 +616,22 @@ final void removeFirst() {
             size--;
             // can't just move the head because it would retain the very first value
             // can't null out the head's value because of late replayers would see null
+            setFirst(next);
+
+        }
+        /**
+         * Arranges the given node is the new head from now on.
+         * @param next the Node instance to set as first
+         */
+        final void setFirst(Node next) {
             Node newHead = new Node(null);
             Node newNext = next.get();
             newHead.set(newNext);
-            setFirst(newHead);
+            set(newHead);
             if (newNext == null) {
                 tail = newHead;
             }
         }
-        /* test */ final void removeSome(int n) {
-            Node head = get();
-            while (n > 0) {
-                head = head.get();
-                n--;
-                size--;
-            }
-
-            setFirst(head);
-        }
-        /**
-         * Arranges the given node is the new head from now on.
-         * @param n the Node instance to set as first
-         */
-        final void setFirst(Node n) {
-            set(n);
-        }
 
         @Override
         public final void next(T value) {
@@ -740,41 +731,45 @@ Object leaveTransform(Object value) {
         void truncateFinal() {
 
         }
-        /* test */ final  void collect(Collection<? super T> output) {
-            Node n = getHead();
+        /**
+         * Collect all values in memory even if they will be ignored by subsequent calls to replay(). This is useful
+         * for tracking memory leaks.
+         * @param output all values.
+         */
+        /* test */ final void collectValuesInMemory(Collection<? super T> output) {
+            Node next = getHead();
             for (;;) {
-                Node next = n.get();
                 if (next != null) {
                     Object o = next.value;
                     Object v = leaveTransform(o);
                     if (NotificationLite.isComplete(v) || NotificationLite.isError(v)) {
                         break;
                     }
-                    output.add(NotificationLite.<T>getValue(v));
-                    n = next;
+                    T value = NotificationLite.<T>getValue(v);
+                    if (value != null) {
+                        output.add(value);
+                    }
                 } else {
                     break;
                 }
+                next = next.get();
             }
         }
-        /* test */ final void collectObjectsInMemory(Collection<? super T> output) {
-            Node next = getHead();
+        /* test */ final void collect(Collection<? super T> output) {
+            Node n = getHead();
             for (;;) {
+                Node next = n.get();
                 if (next != null) {
                     Object o = next.value;
                     Object v = leaveTransform(o);
                     if (NotificationLite.isComplete(v) || NotificationLite.isError(v)) {
                         break;
                     }
-                    T value = NotificationLite.<T>getValue(v);
-                    if (value != null) {
-                        output.add(value);
-                    }
-                    next = next;
+                    output.add(NotificationLite.<T>getValue(v));
+                    n = next;
                 } else {
                     break;
                 }
-                next = next.get();
             }
         }
         /* test */ boolean hasError() {

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

@@ -731,60 +731,24 @@ void truncate() {
         buf.addLast(new Node(5));
 
         List<Integer> values = new ArrayList<Integer>();
-        buf.collect(values);
+        collectAndAssertNoLeaks(buf, values);
 
         Assert.assertEquals(Arrays.asList(1, 2, 3, 4, 5), values);
 
-        buf.removeSome(2);
-        buf.removeFirst();
-        buf.removeSome(2);
-
-        values.clear();
-        buf.collect(values);
-        Assert.assertTrue(values.isEmpty());
-
-        buf.addLast(new Node(5));
-        buf.addLast(new Node(6));
-        buf.collect(values);
-
-        Assert.assertEquals(Arrays.asList(5, 6), values);
-    }
-
-    @Test
-    public void testBoundedReplayBufferForMemoryLeaks() {
-        BoundedReplayBuffer<Integer> buf = new BoundedReplayBuffer<Integer>() {
-            private static final long serialVersionUID = -5182053207244406872L;
-
-            @Override
-            void truncate() {
-            }
-        };
-        buf.addLast(new Node(1));
-        buf.addLast(new Node(2));
-        buf.addLast(new Node(3));
-        buf.addLast(new Node(4));
-        buf.addLast(new Node(5));
-
-        List<Integer> valuesInMemory = new ArrayList<Integer>();
-        buf.collectObjectsInMemory(valuesInMemory);
-
-        Assert.assertEquals(Arrays.asList(1, 2, 3, 4, 5), valuesInMemory);
-
         buf.removeFirst();
         buf.removeFirst();
         buf.removeFirst();
         buf.removeFirst();
         buf.removeFirst();
 
-        valuesInMemory.clear();
-        buf.collectObjectsInMemory(valuesInMemory);
-        Assert.assertTrue(valuesInMemory.isEmpty());
+        collectAndAssertNoLeaks(buf, values);
+        Assert.assertTrue(values.isEmpty());
 
         buf.addLast(new Node(5));
         buf.addLast(new Node(6));
-        buf.collectObjectsInMemory(valuesInMemory);
+        collectAndAssertNoLeaks(buf, values);
 
-        Assert.assertEquals(Arrays.asList(5, 6), valuesInMemory);
+        Assert.assertEquals(Arrays.asList(5, 6), values);
     }
 
     @Test
@@ -877,26 +841,23 @@ public void testSizedTruncation() {
 
         buf.next(1);
         buf.next(2);
-        buf.collect(values);
+        collectAndAssertNoLeaks(buf, values);
         Assert.assertEquals(Arrays.asList(1, 2), values);
 
         buf.next(3);
         buf.next(4);
-        values.clear();
-        buf.collect(values);
+        collectAndAssertNoLeaks(buf, values);
         Assert.assertEquals(Arrays.asList(3, 4), values);
 
         buf.next(5);
 
-        values.clear();
-        buf.collect(values);
+        collectAndAssertNoLeaks(buf, values);
         Assert.assertEquals(Arrays.asList(4, 5), values);
         Assert.assertFalse(buf.hasCompleted());
 
         buf.complete();
 
-        values.clear();
-        buf.collect(values);
+        collectAndAssertNoLeaks(buf, values);
         Assert.assertEquals(Arrays.asList(4, 5), values);
 
         Assert.assertEquals(3, buf.size);
@@ -1564,4 +1525,12 @@ public void timedNoOutdatedData() {
 
         source.test().assertResult();
     }
+
+    private <T> void collectAndAssertNoLeaks(BoundedReplayBuffer<T> buffer, Collection<T> collection) {
+        Collection<T> allObjects = new ArrayList<T>();
+        collection.clear();
+        buffer.collectValuesInMemory(allObjects);
+        buffer.collect(collection);
+        assertEquals(allObjects, collection);
+    }
 }