1.x: fix Single.flatMap not composing subscription through (ReactiveX#3941)

akarnokd · zsxwing · commit 4a6441ce8d49 · 2016-05-17T12:57:36.000-07:00
Closes ReactiveX#3940.
diff --git a/src/main/java/rx/Single.java b/src/main/java/rx/Single.java
@@ -671,7 +671,7 @@ public static <T> Single<T> merge(final Single<? extends Single<? extends T>> so
 
             @Override
             public void call(final SingleSubscriber<? super T> child) {
-                source.subscribe(new SingleSubscriber<Single<? extends T>>() {
+                SingleSubscriber<Single<? extends T>> parent = new SingleSubscriber<Single<? extends T>>() {
 
                     @Override
                     public void onSuccess(Single<? extends T> innerSingle) {
@@ -683,7 +683,9 @@ public void onError(Throwable error) {
                         child.onError(error);
                     }
 
-                });
+                };
+                child.add(parent);
+                source.subscribe(parent);
             }
         });
     }
diff --git a/src/test/java/rx/SingleTest.java b/src/test/java/rx/SingleTest.java
@@ -1914,4 +1914,44 @@ public void subscribeWithNullObserver() {
             assertEquals("observer is null", ex.getMessage());
         }
     }
+
+    @Test
+    public void unsubscribeComposesThrough() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+        
+        Subscription s = ps.toSingle()
+        .flatMap(new Func1<Integer, Single<Integer>>() {
+            @Override
+            public Single<Integer> call(Integer v) {
+                return Single.just(1);
+            }
+        })
+        .subscribe();
+        
+        s.unsubscribe();
+        
+        assertFalse("Observers present?!", ps.hasObservers());
+    }
+
+    @Test(timeout = 1000)
+    public void unsubscribeComposesThroughAsync() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+        
+        Subscription s = ps.toSingle()
+        .subscribeOn(Schedulers.io())
+        .flatMap(new Func1<Integer, Single<Integer>>() {
+            @Override
+            public Single<Integer> call(Integer v) {
+                return Single.just(1);
+            }
+        })
+        .subscribe();
+        
+        while (!ps.hasObservers() && !Thread.currentThread().isInterrupted()) ;
+        
+        s.unsubscribe();
+        
+        assertFalse("Observers present?!", ps.hasObservers());
+    }
+
 }

Original file line number	Diff line number	Diff line change
`@@ -671,7 +671,7 @@ public static <T> Single<T> merge(final Single<? extends Single<? extends T>> so`
`671`	`671`
`672`	`672`	`@Override`
`673`	`673`	`public void call(final SingleSubscriber<? super T> child) {`
`674`		`- source.subscribe(new SingleSubscriber<Single<? extends T>>() {`
	`674`	`+ SingleSubscriber<Single<? extends T>> parent = new SingleSubscriber<Single<? extends T>>() {`
`675`	`675`
`676`	`676`	`@Override`
`677`	`677`	`public void onSuccess(Single<? extends T> innerSingle) {`
`@@ -683,7 +683,9 @@ public void onError(Throwable error) {`
`683`	`683`	`child.onError(error);`
`684`	`684`	`}`
`685`	`685`
`686`		`- });`
	`686`	`+ };`
	`687`	`+ child.add(parent);`
	`688`	`+ source.subscribe(parent);`
`687`	`689`	`}`
`688`	`690`	`});`
`689`	`691`	`}`