Merge pull request #1938 from akarnokd/OperatorAnyFix

Any/All should not unsubscribe downstream.
ReactiveX · Dec 13, 2014 · 162e042 · 162e042
2 parents e4239b5 + a40a659
commit 162e042
Show file tree

Hide file tree

Showing 4 changed files with 44 additions and 18 deletions.
diff --git a/src/main/java/rx/internal/operators/OperatorAll.java b/src/main/java/rx/internal/operators/OperatorAll.java
@@ -34,7 +34,7 @@ public OperatorAll(Func1<? super T, Boolean> predicate) {
 
     @Override
     public Subscriber<? super T> call(final Subscriber<? super Boolean> child) {
-        return new Subscriber<T>(child) {
+        Subscriber<T> s = new Subscriber<T>() {
             boolean done;
 
             @Override
@@ -65,5 +65,7 @@ public void onCompleted() {
                 }
             }
         };
+        child.add(s);
+        return s;
     }
 }
diff --git a/src/main/java/rx/internal/operators/OperatorAny.java b/src/main/java/rx/internal/operators/OperatorAny.java
@@ -36,7 +36,7 @@ public OperatorAny(Func1<? super T, Boolean> predicate, boolean returnOnEmpty) {
 
     @Override
     public Subscriber<? super T> call(final Subscriber<? super Boolean> child) {
-        return new Subscriber<T>(child) {
+        Subscriber<T> s = new Subscriber<T>() {
             boolean hasElements;
             boolean done;
 
@@ -74,5 +74,7 @@ public void onCompleted() {
             }
 
         };
+        child.add(s);
+        return s;
     }
 }
diff --git a/src/test/java/rx/internal/operators/OperatorAllTest.java b/src/test/java/rx/internal/operators/OperatorAllTest.java
@@ -15,19 +15,17 @@
  */
 package rx.internal.operators;
 
-import static org.junit.Assert.assertFalse;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 
-import rx.Observable;
-import rx.Observer;
+import rx.*;
 import rx.functions.Func1;
 
-import java.util.Arrays;
-
 public class OperatorAllTest {
 
     @Test
@@ -113,4 +111,21 @@ public Boolean call(Integer i) {
         });
         assertFalse(allOdd.toBlocking().first());
     }
+    @Test(timeout = 5000)
+    public void testIssue1935NoUnsubscribeDownstream() {
+        Observable<Integer> source = Observable.just(1)
+            .all(new Func1<Object, Boolean>() {
+                @Override
+                public Boolean call(Object t1) {
+                    return false;
+                }
+            })
+            .flatMap(new Func1<Boolean, Observable<Integer>>() {
+                @Override
+                public Observable<Integer> call(Boolean t1) {
+                    return Observable.just(2).delay(500, TimeUnit.MILLISECONDS);
+                }
+        });
+        assertEquals((Object)2, source.toBlocking().first());
+    }
 }
diff --git a/src/test/java/rx/internal/operators/OperatorAnyTest.java b/src/test/java/rx/internal/operators/OperatorAnyTest.java
@@ -15,20 +15,16 @@
  */
 package rx.internal.operators;
 
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
 
 import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 
-import rx.Observable;
-import rx.Observer;
+import rx.*;
 import rx.functions.Func1;
-import rx.functions.Functions;
 import rx.internal.util.UtilityFunctions;
 
 public class OperatorAnyTest {
@@ -213,4 +209,15 @@ public Boolean call(Integer i) {
         });
         assertTrue(anyEven.toBlocking().first());
     }
+    @Test(timeout = 5000)
+    public void testIssue1935NoUnsubscribeDownstream() {
+        Observable<Integer> source = Observable.just(1).isEmpty()
+            .flatMap(new Func1<Boolean, Observable<Integer>>() {
+                @Override
+                public Observable<Integer> call(Boolean t1) {
+                    return Observable.just(2).delay(500, TimeUnit.MILLISECONDS);
+                }
+        });
+        assertEquals((Object)2, source.toBlocking().first());
+    }
 }