Merge Scheduler/AbstractScheduler

- using abstract class for Scheduler for same reason Observable is concrete
- discussed and decided upon at ReactiveX#235

Author: benjchristensen

rxjava-core/src/main/java/rx/Scheduler.java

@@ -17,15 +17,31 @@
 
 import java.util.concurrent.TimeUnit;
 
+import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action0;
 import rx.util.functions.Func0;
 import rx.util.functions.Func1;
 import rx.util.functions.Func2;
 
 /**
  * Represents an object that schedules units of work.
+ * <p>
+ * The methods left to implement are:
+ * <ul>
+ * <li>{@code <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action, long delayTime, TimeUnit unit)}</li>
+ * <li>{@code <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action)}</li>
+ * </ul>
+ * <p>
+ * Why is this an abstract class instead of an interface?
+ * <p>
+ * <ol>
+ * <li>Java doesn't support extension methods and there are many overload methods needing default implementations.</li>
+ * <li>Virtual extension methods aren't available until Java8 which RxJava will not set as a minimum target for a long time.</li>
+ * <li>If only an interface were used Scheduler implementations would then need to extend from an AbstractScheduler pair that gives all of the functionality unless they intend on copy/pasting the functionality.</li>
+ * <li>Without virtual extension methods even additive changes are breaking and thus severely impede library maintenance.</li>
+ * </ol>
  */
-public interface Scheduler {
+public abstract class Scheduler {
 
     /**
      * Schedules a cancelable action to be executed.
@@ -36,7 +52,7 @@ public interface Scheduler {
      *            Action to schedule.
      * @return a subscription to be able to unsubscribe from action.
      */
-    <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action);
+    public abstract <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action);
 
     /**
      * Schedules a cancelable action to be executed in delayTime.
@@ -51,7 +67,7 @@ public interface Scheduler {
      *            Time unit of the delay time.
      * @return a subscription to be able to unsubscribe from action.
      */
-    <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action, long delayTime, TimeUnit unit);
+    public abstract <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action, long delayTime, TimeUnit unit);
 
     /**
      * Schedules a cancelable action to be executed.
@@ -60,7 +76,15 @@ public interface Scheduler {
      *            Action to schedule.
      * @return a subscription to be able to unsubscribe from action.
      */
-    Subscription schedule(Func1<Scheduler, Subscription> action);
+    public Subscription schedule(final Func1<Scheduler, Subscription> action) {
+        return schedule(null, new Func2<Scheduler, Void, Subscription>() {
+
+            @Override
+            public Subscription call(Scheduler scheduler, Void t2) {
+                return action.call(scheduler);
+            }
+        });
+    }
 
     /**
      * Schedules a cancelable action to be executed.
@@ -69,7 +93,15 @@ public interface Scheduler {
      *            action
      * @return a subscription to be able to unsubscribe from action.
      */
-    Subscription schedule(Func0<Subscription> action);
+    public Subscription schedule(final Func0<Subscription> action) {
+        return schedule(null, new Func2<Scheduler, Void, Subscription>() {
+
+            @Override
+            public Subscription call(Scheduler scheduler, Void t2) {
+                return action.call();
+            }
+        });
+    }
 
     /**
      * Schedules an action to be executed.
@@ -78,7 +110,16 @@ public interface Scheduler {
      *            action
      * @return a subscription to be able to unsubscribe from action.
      */
-    Subscription schedule(Action0 action);
+    public Subscription schedule(final Action0 action) {
+        return schedule(null, new Func2<Scheduler, Void, Subscription>() {
+
+            @Override
+            public Subscription call(Scheduler scheduler, Void t2) {
+                action.call();
+                return Subscriptions.empty();
+            }
+        });
+    }
 
     /**
      * Schedules a cancelable action to be executed in delayTime.
@@ -91,7 +132,15 @@ public interface Scheduler {
      *            Time unit of the delay time.
      * @return a subscription to be able to unsubscribe from action.
      */
-    Subscription schedule(Func1<Scheduler, Subscription> action, long delayTime, TimeUnit unit);
+    public Subscription schedule(final Func1<Scheduler, Subscription> action, long delayTime, TimeUnit unit) {
+        return schedule(null, new Func2<Scheduler, Void, Subscription>() {
+
+            @Override
+            public Subscription call(Scheduler scheduler, Void t2) {
+                return action.call(scheduler);
+            }
+        }, delayTime, unit);
+    }
 
     /**
      * Schedules an action to be executed in delayTime.
@@ -100,7 +149,16 @@ public interface Scheduler {
      *            action
      * @return a subscription to be able to unsubscribe from action.
      */
-    Subscription schedule(Action0 action, long delayTime, TimeUnit unit);
+    public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit) {
+        return schedule(null, new Func2<Scheduler, Void, Subscription>() {
+
+            @Override
+            public Subscription call(Scheduler scheduler, Void t2) {
+                action.call();
+                return Subscriptions.empty();
+            }
+        }, delayTime, unit);
+    }
 
     /**
      * Schedules a cancelable action to be executed in delayTime.
@@ -109,11 +167,21 @@ public interface Scheduler {
      *            action
      * @return a subscription to be able to unsubscribe from action.
      */
-    Subscription schedule(Func0<Subscription> action, long delayTime, TimeUnit unit);
+    public Subscription schedule(final Func0<Subscription> action, long delayTime, TimeUnit unit) {
+        return schedule(null, new Func2<Scheduler, Void, Subscription>() {
+
+            @Override
+            public Subscription call(Scheduler scheduler, Void t2) {
+                return action.call();
+            }
+        }, delayTime, unit);
+    }
 
     /**
-     * Returns the scheduler's notion of current time.
+     * Returns the scheduler's notion of current absolute time in milliseconds.
      */
-    long now();
+    public long now() {
+        return System.currentTimeMillis();
+    }
 
 }

rxjava-core/src/main/java/rx/concurrency/AbstractScheduler.java

@@ -32,7 +32,7 @@
 /**
  * Schedules work on the current thread but does not execute immediately. Work is put in a queue and executed after the current unit of work is completed.
  */
-public class CurrentThreadScheduler extends AbstractScheduler {
+public class CurrentThreadScheduler extends Scheduler {
     private static final CurrentThreadScheduler INSTANCE = new CurrentThreadScheduler();
 
     public static CurrentThreadScheduler getInstance() {

rxjava-core/src/main/java/rx/concurrency/CurrentThreadScheduler.java

@@ -33,7 +33,7 @@
  * <p>
  * Note that if an {@link Executor} implementation is used instead of {@link ScheduledExecutorService} then a system-wide Timer will be used to handle delayed events.
  */
-public class ExecutorScheduler extends AbstractScheduler {
+public class ExecutorScheduler extends Scheduler {
     private final Executor executor;
 
     public ExecutorScheduler(Executor executor) {

rxjava-core/src/main/java/rx/concurrency/ExecutorScheduler.java

@@ -30,7 +30,7 @@
 /**
  * Executes work immediately on the current thread.
  */
-public final class ImmediateScheduler extends AbstractScheduler {
+public final class ImmediateScheduler extends Scheduler {
     private static final ImmediateScheduler INSTANCE = new ImmediateScheduler();
 
     public static ImmediateScheduler getInstance() {

rxjava-core/src/main/java/rx/concurrency/ImmediateScheduler.java

@@ -28,7 +28,7 @@
 /**
  * Schedules work on a new thread.
  */
-public class NewThreadScheduler extends AbstractScheduler {
+public class NewThreadScheduler extends Scheduler {
     private static final NewThreadScheduler INSTANCE = new NewThreadScheduler();
 
     public static NewThreadScheduler getInstance() {

rxjava-core/src/main/java/rx/concurrency/NewThreadScheduler.java

@@ -273,7 +273,7 @@ public void onNext(String args)
             }
         }
 
-        public static class ForwardingScheduler implements Scheduler {
+        public static class ForwardingScheduler extends Scheduler {
             private final Scheduler underlying;
 
             public ForwardingScheduler(Scheduler underlying) {