Change Scheduler implementations to use primary methods

Work done in ReactiveX#229 added the following methods:

- Subscription schedule(T state, Func2<Scheduler, T, Subscription> action, long delayTime, TimeUnit unit)}
- Subscription schedule(T state, Func2<Scheduler, T, Subscription> action)}

These are in fact the primary methods from RxNet (http://msdn.microsoft.com/en-us/library/hh211963(v=vs.103).aspx) and the others are just helper overloads.

It seems it is better to set the precedent to use these 2 methods for actual implementation logic while all other methods are just decorating and forwarding from AbstractScheduler to these methods.

I have updated the various implementations to achieve this.

Unit tests are passing … but we don't have enough unit test coverage so I won't be surprised if bugs are found.

Author: benjchristensen

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

@@ -30,20 +30,38 @@ public interface Scheduler {
     /**
      * Schedules a cancelable action to be executed.
      * 
-     * @param state State to pass into the action.
-     * @param action Action to schedule.
+     * @param state
+     *            State to pass into the action.
+     * @param action
+     *            Action to schedule.
      * @return a subscription to be able to unsubscribe from action.
      */
     <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action);
 
+    /**
+     * Schedules a cancelable action to be executed in delayTime.
+     * 
+     * @param state
+     *            State to pass into the action.
+     * @param action
+     *            Action to schedule.
+     * @param delayTime
+     *            Time the action is to be delayed before executing.
+     * @param unit
+     *            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);
+
     /**
      * Schedules a cancelable action to be executed.
      * 
-     * @param action Action to schedule.
+     * @param action
+     *            Action to schedule.
      * @return a subscription to be able to unsubscribe from action.
      */
     Subscription schedule(Func1<Scheduler, Subscription> action);
-    
+
     /**
      * Schedules a cancelable action to be executed.
      * 
@@ -63,43 +81,35 @@ public interface Scheduler {
     Subscription schedule(Action0 action);
 
     /**
-     * Schedules a cancelable action to be executed in dueTime.
+     * Schedules a cancelable action to be executed in delayTime.
      * 
-     * @param state State to pass into the action.
-     * @param action Action to schedule.
-     * @param dueTime Time the action is due for executing.
-     * @param unit Time unit of the due time.
+     * @param action
+     *            Action to schedule.
+     * @param delayTime
+     *            Time the action is to be delayed before executing.
+     * @param unit
+     *            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 dueTime, TimeUnit unit);
+    Subscription schedule(Func1<Scheduler, Subscription> action, long delayTime, TimeUnit unit);
 
     /**
-     * Schedules a cancelable action to be executed in dueTime.
-     * 
-     * @param action Action to schedule.
-     * @param dueTime Time the action is due for executing.
-     * @param unit Time unit of the due time.
-     * @return a subscription to be able to unsubscribe from action.
-     */
-    Subscription schedule(Func1<Scheduler, Subscription> action, long dueTime, TimeUnit unit);
-    
-    /**
-     * Schedules an action to be executed in dueTime.
+     * Schedules an action to be executed in delayTime.
      * 
      * @param action
      *            action
      * @return a subscription to be able to unsubscribe from action.
      */
-    Subscription schedule(Action0 action, long dueTime, TimeUnit unit);
+    Subscription schedule(Action0 action, long delayTime, TimeUnit unit);
 
     /**
-     * Schedules a cancelable action to be executed in dueTime.
+     * Schedules a cancelable action to be executed in delayTime.
      * 
      * @param action
      *            action
      * @return a subscription to be able to unsubscribe from action.
      */
-    Subscription schedule(Func0<Subscription> action, long dueTime, TimeUnit unit);
+    Subscription schedule(Func0<Subscription> action, long delayTime, TimeUnit unit);
 
     /**
      * Returns the scheduler's notion of current time.

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

@@ -25,71 +25,90 @@
 import rx.util.functions.Func1;
 import rx.util.functions.Func2;
 
-/* package */abstract class AbstractScheduler implements Scheduler {
+/**
+ * Default implementations of various convenience overload methods on the Scheduler.
+ * <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>
+ * This is a utility class expected to be used by all {@link Scheduler} implementations since we can't yet rely on Java 8 default methods on the {@link Scheduler}.
+ */
+public abstract class AbstractScheduler implements Scheduler {
 
     @Override
-    public Subscription schedule(Action0 action) {
-        return schedule(asFunc0(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();
+            }
+        });
     }
 
     @Override
     public Subscription schedule(final Func1<Scheduler, Subscription> action) {
-        return schedule(new Func0<Subscription>() {
+        return schedule(null, new Func2<Scheduler, Void, Subscription>() {
+
             @Override
-            public Subscription call() {
-                return action.call(AbstractScheduler.this);
+            public Subscription call(Scheduler scheduler, Void t2) {
+                return action.call(scheduler);
             }
         });
     }
 
     @Override
-    public <T> Subscription schedule(final T state, final Func2<Scheduler, T, Subscription> action) {
-      return schedule(new Func0<Subscription>() {
-          @Override
-          public Subscription call() {
-              return action.call(AbstractScheduler.this, state);
-          }
-      });
+    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);
     }
 
     @Override
-    public Subscription schedule(Action0 action, long dueTime, TimeUnit unit) {
-        return schedule(asFunc0(action), dueTime, 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);
     }
 
     @Override
-    public Subscription schedule(final Func1<Scheduler, Subscription> action, long dueTime, TimeUnit unit) {
-      return schedule(new Func0<Subscription>() {
-          @Override
-          public Subscription call() {
-              return action.call(AbstractScheduler.this);
-          }
-    }, dueTime, unit);
+    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();
+            }
+        });
     }
 
     @Override
-    public <T> Subscription schedule(final T state, final Func2<Scheduler, T, Subscription> action, long dueTime, TimeUnit unit) {
-        return schedule(new Func0<Subscription>() {
+    public Subscription schedule(final Func0<Subscription> action, long delayTime, TimeUnit unit) {
+        return schedule(null, new Func2<Scheduler, Void, Subscription>() {
+
             @Override
-            public Subscription call() {
-                return action.call(AbstractScheduler.this, state);
+            public Subscription call(Scheduler scheduler, Void t2) {
+                return action.call();
             }
-        }, dueTime, unit);
+        }, delayTime, unit);
     }
-    
+
     @Override
     public long now() {
         return System.nanoTime();
     }
 
-    private static Func0<Subscription> asFunc0(final Action0 action) {
-        return new Func0<Subscription>() {
-            @Override
-            public Subscription call() {
-                action.call();
-                return Subscriptions.empty();
-            }
-        };
-    }
-
 }

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

@@ -24,9 +24,10 @@
 import org.junit.Test;
 import org.mockito.InOrder;
 
+import rx.Scheduler;
 import rx.Subscription;
 import rx.util.functions.Action0;
-import rx.util.functions.Func0;
+import rx.util.functions.Func2;
 
 /**
  * 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.
@@ -38,37 +39,40 @@ public static CurrentThreadScheduler getInstance() {
         return INSTANCE;
     }
 
-    private static final ThreadLocal<Queue<DiscardableAction>> QUEUE = new ThreadLocal<Queue<DiscardableAction>>();
+    private static final ThreadLocal<Queue<DiscardableAction<?>>> QUEUE = new ThreadLocal<Queue<DiscardableAction<?>>>();
 
     private CurrentThreadScheduler() {
     }
 
     @Override
-    public Subscription schedule(Func0<Subscription> action) {
-        DiscardableAction discardableAction = new DiscardableAction(action);
+    public <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action) {
+        DiscardableAction<T> discardableAction = new DiscardableAction<T>(state, action);
         enqueue(discardableAction);
         return discardableAction;
     }
 
     @Override
-    public Subscription schedule(Func0<Subscription> action, long dueTime, TimeUnit unit) {
-        return schedule(new SleepingAction(action, this, dueTime, unit));
+    public <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action, long dueTime, TimeUnit unit) {
+        // since we are executing immediately on this thread we must cause this thread to sleep
+        // TODO right now the 'enqueue' does not take delay into account so if another task is enqueued after this it will 
+        // wait behind the sleeping action ... should that be the case or should it be allowed to proceed ahead of the delayed action?
+        return schedule(state, new SleepingAction<T>(action, this, dueTime, unit));
     }
 
-    private void enqueue(DiscardableAction action) {
-        Queue<DiscardableAction> queue = QUEUE.get();
+    private void enqueue(DiscardableAction<?> action) {
+        Queue<DiscardableAction<?>> queue = QUEUE.get();
         boolean exec = queue == null;
 
         if (exec) {
-            queue = new LinkedList<DiscardableAction>();
+            queue = new LinkedList<DiscardableAction<?>>();
             QUEUE.set(queue);
         }
 
         queue.add(action);
 
         if (exec) {
             while (!queue.isEmpty()) {
-                queue.poll().call();
+                queue.poll().call(this);
             }
 
             QUEUE.set(null);
@@ -143,4 +147,5 @@ public void testSequenceOfActions() {
         }
 
     }
+
 }

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

@@ -17,27 +17,31 @@
 
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import rx.Scheduler;
 import rx.Subscription;
 import rx.util.AtomicObservableSubscription;
-import rx.util.functions.Func0;
+import rx.util.functions.Func1;
+import rx.util.functions.Func2;
 
 /**
  * Combines standard {@link Subscription#unsubscribe()} functionality with ability to skip execution if an unsubscribe occurs before the {@link #call()} method is invoked.
  */
-/* package */class DiscardableAction implements Func0<Subscription>, Subscription {
-    private final Func0<Subscription> underlying;
+/* package */class DiscardableAction<T> implements Func1<Scheduler, Subscription>, Subscription {
+    private final Func2<Scheduler, T, Subscription> underlying;
+    private final T state;
 
     private final AtomicObservableSubscription wrapper = new AtomicObservableSubscription();
     private final AtomicBoolean ready = new AtomicBoolean(true);
 
-    public DiscardableAction(Func0<Subscription> underlying) {
+    public DiscardableAction(T state, Func2<Scheduler, T, Subscription> underlying) {
+        this.state = state;
         this.underlying = underlying;
     }
 
     @Override
-    public Subscription call() {
+    public Subscription call(Scheduler scheduler) {
         if (ready.compareAndSet(true, false)) {
-            Subscription subscription = underlying.call();
+            Subscription subscription = underlying.call(scheduler, state);
             wrapper.wrap(subscription);
             return subscription;
         }
@@ -49,4 +53,5 @@ public void unsubscribe() {
         ready.set(false);
         wrapper.unsubscribe();
     }
+
 }