Merge pull request #235 from benjchristensen/schedulers-pull-229-merge

Schedulers Interface (Merging and Adding to Pull Request 229)

Author: benjchristensen

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

@@ -15,15 +15,99 @@
  */
 package rx;
 
+import java.util.Date;
 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.
+     * 
+     * @param state
+     *            State to pass into the action.
+     * @param action
+     *            Action to schedule.
+     * @return a subscription to be able to unsubscribe from action.
+     */
+    public abstract <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.
+     */
+    public abstract <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action, long delayTime, TimeUnit unit);
+
+    /**
+     * Schedules a cancelable action to be executed at dueTime.
+     * 
+     * @param state
+     *            State to pass into the action.
+     * @param action
+     *            Action to schedule.
+     * @param dueTime
+     *            Time the action is to be executed. If in the past it will be executed immediately.
+     * @return a subscription to be able to unsubscribe from action.
+     */
+    public <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action, Date dueTime) {
+        long scheduledTime = dueTime.getTime();
+        long timeInFuture = scheduledTime - now();
+        if (timeInFuture <= 0) {
+            return schedule(state, action);
+        } else {
+            return schedule(state, action, timeInFuture, TimeUnit.MILLISECONDS);
+        }
+    }
+
+    /**
+     * Schedules a cancelable action to be executed.
+     * 
+     * @param action
+     *            Action to schedule.
+     * @return a subscription to be able to unsubscribe from 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.
@@ -32,7 +116,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.
@@ -41,29 +133,78 @@ 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 an action to be executed in dueTime.
+     * Schedules a cancelable action to be executed in delayTime.
+     * 
+     * @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.
+     */
+    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.
      * 
      * @param action
      *            action
      * @return a subscription to be able to unsubscribe from action.
      */
-    Subscription schedule(Action0 action, long dueTime, 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 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);
+    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

@@ -24,51 +24,55 @@
 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.
  */
-public class CurrentThreadScheduler extends AbstractScheduler {
+public class CurrentThreadScheduler extends Scheduler {
     private static final CurrentThreadScheduler INSTANCE = new CurrentThreadScheduler();
 
     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/CurrentThreadScheduler.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();
     }
+
 }