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

Closes  #3940.

Author: akarnokd

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);
             }
         });
     }

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());
+    }
+
 }