Func0 can transparently implement java.util.concurrent.Callable.

This change doesn't change the API at all for users of Func0, but it makes all Func0 objects immediately reusable with any JDK API that accepts Callables. For example, a Func0 can now be submitted directly to an ExecutorService for asynchronous execution. It also allows the elimination of a small amount of redundant code within RxJava itself.

Author: rickbw

rxjava-contrib/rxjava-async-util/src/main/java/rx/util/async/Async.java

@@ -1596,9 +1596,13 @@ public static <R> Observable<R> fromAction(Action0 action, R result) {
      * @see #start(rx.functions.Func0) 
      * @see #fromCallable(java.util.concurrent.Callable) 
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Async-Operators#fromfunc0">RxJava Wiki: fromFunc0()</a>
+     *
+     * @deprecated  Unnecessary now that Func0 extends Callable. Just call
+     *              {@link #fromCallable(Callable)} instead.
      */
+    @Deprecated
     public static <R> Observable<R> fromFunc0(Func0<? extends R> function) {
-        return fromFunc0(function, Schedulers.computation());
+        return fromCallable(function);
     }
 
     /**
@@ -1674,9 +1678,13 @@ public static <R> Observable<R> fromAction(Action0 action, R result, Scheduler s
      * @see #start(rx.functions.Func0) 
      * @see #fromCallable(java.util.concurrent.Callable) 
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Async-Operators#fromfunc0">RxJava Wiki: fromFunc0()</a>
+     *
+     * @deprecated  Unnecessary now that Func0 extends Callable. Just call
+     *              {@link #fromCallable(Callable, Scheduler)} instead.
      */
+    @Deprecated
     public static <R> Observable<R> fromFunc0(Func0<? extends R> function, Scheduler scheduler) {
-        return Observable.create(OperationFromFunctionals.fromFunc0(function)).subscribeOn(scheduler);
+        return fromCallable(function, scheduler);
     }
 
     /**

rxjava-contrib/rxjava-async-util/src/main/java/rx/util/async/operators/OperationFromFunctionals.java

@@ -38,22 +38,28 @@ public final class OperationFromFunctionals {
     public static <R> OnSubscribeFunc<R> fromAction(Action0 action, R result) {
         return new InvokeAsync<R>(Actions.toFunc(action, result));
     }
-    
-    /** Subscriber function that invokes a function and returns its value. */
+
+    /**
+     * Subscriber function that invokes a function and returns its value.
+     *
+     * @deprecated  Unnecessary now that Func0 extends Callable. Just call
+     *              {@link #fromCallable(Callable)} instead.
+     */
+    @Deprecated
     public static <R> OnSubscribeFunc<R> fromFunc0(Func0<? extends R> function) {
-        return new InvokeAsync<R>(function);
+        return fromCallable(function);
     }
 
     /** 
      * Subscriber function that invokes the callable and returns its value or
      * propagates its checked exception.
      */
     public static <R> OnSubscribeFunc<R> fromCallable(Callable<? extends R> callable) {
-        return new InvokeAsyncCallable<R>(callable);
+        return new InvokeAsync<R>(callable);
     }
     /** Subscriber function that invokes a runnable and returns the given result. */
     public static <R> OnSubscribeFunc<R> fromRunnable(final Runnable run, final R result) {
-        return new InvokeAsync(new Func0<R>() {
+        return new InvokeAsync<R>(new Func0<R>() {
             @Override
             public R call() {
                 run.run();
@@ -62,38 +68,13 @@ public R call() {
         });
     }
 
-    /**
-     * Invokes a function when an observer subscribes.
-     * @param <R> the return type
-     */
-    static final class InvokeAsync<R> implements OnSubscribeFunc<R> {
-        final Func0<? extends R> function;
-        public InvokeAsync(Func0<? extends R> function) {
-            if (function == null) {
-                throw new NullPointerException("function");
-            }
-            this.function = function;
-        }
-        @Override
-        public Subscription onSubscribe(Observer<? super R> t1) {
-            Subscription s = Subscriptions.empty();
-            try {
-                t1.onNext(function.call());
-            } catch (Throwable t) {
-                t1.onError(t);
-                return s;
-            }
-            t1.onCompleted();
-            return s;
-        }
-    }
     /**
      * Invokes a java.util.concurrent.Callable when an observer subscribes.
      * @param <R> the return type
      */
-    static final class InvokeAsyncCallable<R> implements OnSubscribeFunc<R> {
+    static final class InvokeAsync<R> implements OnSubscribeFunc<R> {
         final Callable<? extends R> callable;
-        public InvokeAsyncCallable(Callable<? extends R> callable) {
+        public InvokeAsync(Callable<? extends R> callable) {
             if (callable == null) {
                 throw new NullPointerException("function");
             }

rxjava-contrib/rxjava-async-util/src/test/java/rx/util/async/operators/OperationFromFunctionalsTest.java

@@ -110,6 +110,13 @@ public void call() {
 
         testRunShouldThrow(source, RuntimeException.class);
     }
+
+    /**
+     * @deprecated  {@link Func0} now extends {@link Callable}, so
+     *              {@link Async#fromFunc0(Func0)} is unnecessary. Once it's
+     *              removed, this test can be removed as well.
+     */
+    @Deprecated
     @Test
     public void testFromFunc0() {
         Func0<Integer> func = new Func0<Integer>() {
@@ -139,7 +146,14 @@ public Integer call() {
             verify(observer, never()).onError(any(Throwable.class));
         }
     }
-    
+
+    /**
+     * @deprecated  {@link Func0} now extends {@link Callable}, so
+     *              {@link Async#fromFunc0(Func0, rx.Scheduler)} is
+     *              unnecessary. Once it's removed, this test can be removed
+     *              as well.
+     */
+    @Deprecated
     @Test
     public void testFromFunc0Throws() {
         Func0<Integer> func = new Func0<Integer>() {

rxjava-core/src/main/java/rx/functions/Func0.java

@@ -15,6 +15,9 @@
  */
 package rx.functions;
 
-public interface Func0<R> extends Function {
+import java.util.concurrent.Callable;
+
+public interface Func0<R> extends Function, Callable<R> {
+    @Override
     public R call();
-}
+}