Merge pull request #3418 from artem-zinnatullin/single-from-callable

Add Single.fromCallable()

Author: abersnaze

src/main/java/rx/Single.java

@@ -12,6 +12,7 @@
  */
 package rx;
 
+import java.util.concurrent.Callable;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -605,6 +606,43 @@ public final static <T> Single<T> from(Future<? extends T> future, Scheduler sch
         return new Single<T>(OnSubscribeToObservableFuture.toObservableFuture(future)).subscribeOn(scheduler);
     }
 
+    /**
+     * Returns a {@link Single} that invokes passed function and emits its result for each new Observer that subscribes.
+     * <p>
+     * Allows you to defer execution of passed function until Observer subscribes to the {@link Single}.
+     * It makes passed function "lazy".
+     * Result of the function invocation will be emitted by the {@link Single}.
+     * <dl>
+     *   <dt><b>Scheduler:</b></dt>
+     *   <dd>{@code fromCallable} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param func
+     *         function which execution should be deferred, it will be invoked when Observer will subscribe to the {@link Single}.
+     * @param <T>
+     *         the type of the item emitted by the {@link Single}.
+     * @return a {@link Single} whose {@link Observer}s' subscriptions trigger an invocation of the given function.
+     */
+    @Experimental
+    public static <T> Single<T> fromCallable(final Callable<? extends T> func) {
+        return create(new OnSubscribe<T>() {
+            @Override
+            public void call(SingleSubscriber<? super T> singleSubscriber) {
+                final T value;
+
+                try {
+                    value = func.call();
+                } catch (Throwable t) {
+                    Exceptions.throwIfFatal(t);
+                    singleSubscriber.onError(t);
+                    return;
+                }
+
+                singleSubscriber.onSuccess(value);
+            }
+        });
+    }
+
     /**
      * Returns a {@code Single} that emits a specified item.
      * <p>

src/test/java/rx/SingleTest.java

@@ -20,8 +20,10 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
 
 import java.util.Arrays;
+import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -530,4 +532,42 @@ public void doOnErrorShouldThrowCompositeExceptionIfOnErrorActionThrows() {
 
         verify(action).call(error);
     }
+
+    @Test
+    public void shouldEmitValueFromCallable() throws Exception {
+        Callable<String> callable = mock(Callable.class);
+
+        when(callable.call()).thenReturn("value");
+
+        TestSubscriber<String> testSubscriber = new TestSubscriber<String>();
+
+        Single
+                .fromCallable(callable)
+                .subscribe(testSubscriber);
+
+        testSubscriber.assertValue("value");
+        testSubscriber.assertNoErrors();
+
+        verify(callable).call();
+    }
+
+    @Test
+    public void shouldPassErrorFromCallable() throws Exception {
+        Callable<String> callable = mock(Callable.class);
+
+        Throwable error = new IllegalStateException();
+
+        when(callable.call()).thenThrow(error);
+
+        TestSubscriber<String> testSubscriber = new TestSubscriber<String>();
+
+        Single
+                .fromCallable(callable)
+                .subscribe(testSubscriber);
+
+        testSubscriber.assertNoValues();
+        testSubscriber.assertError(error);
+
+        verify(callable).call();
+    }
 }