Merge pull request #3055 from davidmoten/single-subscribe-fix

toSingle() should use unsafeSubscribe

Author: benjchristensen

src/main/java/rx/internal/operators/OnSubscribeSingle.java

@@ -80,7 +80,7 @@ public void onNext(T t) {
             }
         };
         child.add(parent);
-        observable.subscribe(parent);
+        observable.unsafeSubscribe(parent);
     }
 
     public static <T> OnSubscribeSingle<T> create(Observable<T> observable) {

src/test/java/rx/internal/operators/OnSubscribeSingleTest.java

@@ -15,14 +15,19 @@
  */
 package rx.internal.operators;
 
+import static org.junit.Assert.assertFalse;
+
+import java.util.Collections;
+import java.util.NoSuchElementException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
 import org.junit.Test;
+
 import rx.Observable;
 import rx.Single;
+import rx.functions.Action0;
 import rx.observers.TestSubscriber;
 
-import java.util.Collections;
-import java.util.NoSuchElementException;
-
 public class OnSubscribeSingleTest {
 
     @Test
@@ -70,4 +75,19 @@ public void testRepeatObservableThrowsError() {
 
         subscriber.assertError(IllegalArgumentException.class);
     }
+    
+    @Test
+    public void testShouldUseUnsafeSubscribeInternallyNotSubscribe() {
+        TestSubscriber<String> subscriber = TestSubscriber.create();
+        final AtomicBoolean unsubscribed = new AtomicBoolean(false);
+        Single<String> single = Observable.just("Hello World!").doOnUnsubscribe(new Action0() {
+
+            @Override
+            public void call() {
+                unsubscribed.set(true);
+            }}).toSingle();
+        single.unsafeSubscribe(subscriber);
+        subscriber.assertCompleted();
+        assertFalse(unsubscribed.get());
+    }
 }