Merge branch 'master' of https://github.com/akarnokd/RxJava2Extension…

…s.git
akarnokd · May 10, 2018 · 4de5e2d · 4de5e2d
2 parents 0ecd7d3 + 97f5c10
commit 4de5e2d
Show file tree

Hide file tree

Showing 2 changed files with 73 additions and 3 deletions.
diff --git a/src/main/java/hu/akarnokd/rxjava2/operators/FlowableRefCountTimeout.java b/src/main/java/hu/akarnokd/rxjava2/operators/FlowableRefCountTimeout.java
@@ -127,6 +127,9 @@ void terminated(RefConnection rc) {
         synchronized (this) {
             if (!rc.terminated) {
                 rc.terminated = true;
+                if (source instanceof Disposable) {
+                    ((Disposable)source).dispose();
+                }
                 connection = null;
             }
         }
@@ -200,18 +203,18 @@ public void onNext(T t) {
 
         @Override
         public void onError(Throwable t) {
-            actual.onError(t);
             if (compareAndSet(false, true)) {
                 parent.terminated(connection);
             }
+            actual.onError(t);
         }
 
         @Override
         public void onComplete() {
-            actual.onComplete();
             if (compareAndSet(false, true)) {
                 parent.terminated(connection);
             }
+            actual.onComplete();
         }
 
         @Override

diff --git a/src/test/java/hu/akarnokd/rxjava2/operators/FlowableRefCountTimeoutTest.java b/src/test/java/hu/akarnokd/rxjava2/operators/FlowableRefCountTimeoutTest.java
@@ -16,14 +16,18 @@
 
 package hu.akarnokd.rxjava2.operators;
 
+import static org.junit.Assert.assertTrue;
+
 import java.io.IOException;
-import java.util.concurrent.TimeUnit;
+import java.lang.management.ManagementFactory;
+import java.util.concurrent.*;
 
 import org.junit.*;
 import org.reactivestreams.Subscription;
 
 import hu.akarnokd.rxjava2.test.TestHelper;
 import io.reactivex.Flowable;
+import io.reactivex.disposables.Disposable;
 import io.reactivex.functions.Consumer;
 import io.reactivex.processors.PublishProcessor;
 import io.reactivex.schedulers.Schedulers;
@@ -205,4 +209,67 @@ public void run() {
             .assertResult(1, 2, 3, 4, 5);
         }
     }
+
+    Flowable<Object> source;
+
+    @Test
+    public void replayNoLeak() throws Exception {
+        System.gc();
+        Thread.sleep(100);
+
+        long start = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed();
+
+        source = Flowable.fromCallable(new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+                return new byte[100 * 1000 * 1000];
+            }
+        })
+        .replay(1)
+        .compose(FlowableTransformers.<Object>refCount(1));
+
+        source.subscribe();
+
+        System.gc();
+        Thread.sleep(100);
+
+        long after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed();
+
+        source = null;
+        assertTrue(String.format("%,3d -> %,3d%n", start, after), start + 20 * 1000 * 1000 > after);
+    }
+
+    @Test
+    public void replayNoLeak2() throws Exception {
+        System.gc();
+        Thread.sleep(100);
+
+        long start = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed();
+
+        source = Flowable.fromCallable(new Callable<Object>() {
+            @Override
+            public Object call() throws Exception {
+                return new byte[100 * 1000 * 1000];
+            }
+        }).concatWith(Flowable.never())
+        .replay(1)
+        .compose(FlowableTransformers.<Object>refCount(1));
+
+        Disposable s1 = source.subscribe();
+        Disposable s2 = source.subscribe();
+
+        s1.dispose();
+        s2.dispose();
+
+        s1 = null;
+        s2 = null;
+
+        System.gc();
+        Thread.sleep(100);
+
+        long after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed();
+
+        source = null;
+        assertTrue(String.format("%,3d -> %,3d%n", start, after), start + 20 * 1000 * 1000 > after);
+    }
 }