2.x: Add Single.fromObservable(ObservableSource) (#4760)

* 2.x: Add Single.fromObservable(ObservableSource)

* Add Null test

Author: vanniktech

src/main/java/io/reactivex/Single.java

@@ -27,7 +27,7 @@
 import io.reactivex.internal.operators.completable.*;
 import io.reactivex.internal.operators.flowable.*;
 import io.reactivex.internal.operators.maybe.*;
-import io.reactivex.internal.operators.observable.ObservableConcatMap;
+import io.reactivex.internal.operators.observable.*;
 import io.reactivex.internal.operators.single.*;
 import io.reactivex.internal.util.*;
 import io.reactivex.observers.TestObserver;
@@ -573,6 +573,27 @@ public static <T> Single<T> fromPublisher(final Publisher<? extends T> publisher
         return RxJavaPlugins.onAssembly(new SingleFromPublisher<T>(publisher));
     }
 
+    /**
+     * Wraps a specific ObservableSource into a Single and signals its single element or error.
+     * <p>If the ObservableSource is empty, a NoSuchElementException is signalled.
+     * If the source has more than one element, an IndexOutOfBoundsException is signalled.
+     * <p>
+     * <dl>
+     *   <dt><b>Scheduler:</b></dt>
+     *   <dd>{@code fromObservable} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param observableSource the source Observable, not null
+     * @param <T>
+     *         the type of the item emitted by the {@link Single}.
+     * @return the new Single instance
+     */
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public static <T> Single<T> fromObservable(ObservableSource<? extends T> observableSource) {
+        ObjectHelper.requireNonNull(observableSource, "observableSource is null");
+        return RxJavaPlugins.onAssembly(new ObservableSingleSingle<T>(observableSource, null));
+    }
+
     /**
      * Returns a {@code Single} that emits a specified item.
      * <p>

src/main/java/io/reactivex/internal/operators/observable/ObservableSingleSingle.java

@@ -21,11 +21,11 @@
 
 public final class ObservableSingleSingle<T> extends Single<T> {
 
-    final ObservableSource<T> source;
+    final ObservableSource<? extends T> source;
 
     final T defaultValue;
 
-    public ObservableSingleSingle(ObservableSource<T> source, T defaultValue) {
+    public ObservableSingleSingle(ObservableSource<? extends T> source, T defaultValue) {
         this.source = source;
         this.defaultValue = defaultValue;
     }

src/test/java/io/reactivex/single/SingleTest.java

@@ -22,6 +22,7 @@
 import org.junit.*;
 
 import io.reactivex.*;
+import io.reactivex.Observable;
 import io.reactivex.disposables.*;
 import io.reactivex.exceptions.TestException;
 import io.reactivex.functions.*;
@@ -541,5 +542,40 @@ public Integer apply(Single<Integer> v) throws Exception {
             }
         }).intValue());
     }
+
+    @Test(expected = NullPointerException.class)
+    public void fromObservableNull() {
+        Single.fromObservable(null);
+    }
+
+    @Test
+    public void fromObservableEmpty() {
+        Single.fromObservable(Observable.empty())
+            .test()
+            .assertFailure(NoSuchElementException.class);
+    }
+
+    @Test
+    public void fromObservableMoreThan1Elements() {
+        Single.fromObservable(Observable.just(1, 2))
+            .test()
+            .assertFailure(IllegalArgumentException.class)
+            .assertErrorMessage("Sequence contains more than one element!");
+    }
+
+    @Test
+    public void fromObservableOneElement() {
+        Single.fromObservable(Observable.just(1))
+            .test()
+            .assertResult(1);
+    }
+
+    @Test
+    public void fromObservableError() {
+        Single.fromObservable(Observable.error(new RuntimeException("some error")))
+            .test()
+            .assertFailure(RuntimeException.class)
+            .assertErrorMessage("some error");
+    }
 }