BoundedReplayBuffer should not leak memory.

Problem: When `BoundedReplayBuffer` evicts an item to make room for a new item it still contains a strong reference to the evicted reference.

This diff adds a new unit test to verify this behavior. Given this unit test needs to look under the hood of `BoundedReplayBuffer` in order to ensure no memory leaks exist, writing this test requires a new package private method.

This diff fixes the behavior.

Author: Brian Attwell

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

@@ -616,7 +616,13 @@ 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);
+            Node newHead = new Node(null);
+            Node newNext = next.get();
+            newHead.set(newNext);
+            setFirst(newHead);
+            if (newNext == null) {
+                tail = newHead;
+            }
         }
         /* test */ final void removeSome(int n) {
             Node head = get();
@@ -751,6 +757,26 @@ void truncateFinal() {
                 }
             }
         }
+        /* test */ final void collectObjectsInMemory(Collection<? super T> output) {
+            Node next = getHead();
+            for (;;) {
+                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;
+                } else {
+                    break;
+                }
+                next = next.get();
+            }
+        }
         /* test */ boolean hasError() {
             return tail.value != null && NotificationLite.isError(leaveTransform(tail.value));
         }

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

@@ -748,7 +748,43 @@ void truncate() {
         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());
+
+        buf.addLast(new Node(5));
+        buf.addLast(new Node(6));
+        buf.collectObjectsInMemory(valuesInMemory);
 
+        Assert.assertEquals(Arrays.asList(5, 6), valuesInMemory);
     }
 
     @Test