Merge branch 'master' into sleepingfix

Conflicts:
	rxjava-core/src/main/java/rx/concurrency/SleepingAction.java

Author: mairbek

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

@@ -41,6 +41,7 @@
 import rx.operators.OperationConcat;
 import rx.operators.OperationDefer;
 import rx.operators.OperationDematerialize;
+import rx.operators.OperationGroupBy;
 import rx.operators.OperationFilter;
 import rx.operators.OperationFinally;
 import rx.operators.OperationMap;
@@ -59,16 +60,15 @@
 import rx.operators.OperationSynchronize;
 import rx.operators.OperationTake;
 import rx.operators.OperationTakeLast;
+import rx.operators.OperationTakeUntil;
 import rx.operators.OperationTakeWhile;
+import rx.operators.OperationToIterator;
 import rx.operators.OperationToObservableFuture;
 import rx.operators.OperationToObservableIterable;
 import rx.operators.OperationToObservableList;
 import rx.operators.OperationToObservableSortedList;
 import rx.operators.OperationWhere;
 import rx.operators.OperationZip;
-import rx.operators.OperatorGroupBy;
-import rx.operators.OperatorTakeUntil;
-import rx.operators.OperatorToIterator;
 import rx.plugins.RxJavaErrorHandler;
 import rx.plugins.RxJavaObservableExecutionHook;
 import rx.plugins.RxJavaPlugins;
@@ -1303,7 +1303,7 @@ public static <T> Observable<T> merge(Observable<T>... source) {
      * @return An observable sequence containing the elements of the source sequence up to the point the other sequence interrupted further propagation.
      */
     public static <T, E> Observable<T> takeUntil(final Observable<T> source, final Observable<E> other) {
-        return OperatorTakeUntil.takeUntil(source, other);
+        return OperationTakeUntil.takeUntil(source, other);
     }
 
     /**
@@ -1355,7 +1355,7 @@ public static <T> Observable<T> finallyDo(Observable<T> source, Action0 action)
      * @return an observable of observable groups, each of which corresponds to a unique key value, containing all elements that share that same key value.
      */
     public static <K, T, R> Observable<GroupedObservable<K, R>> groupBy(Observable<T> source, final Func1<T, K> keySelector, final Func1<T, R> elementSelector) {
-        return create(OperatorGroupBy.groupBy(source, keySelector, elementSelector));
+        return create(OperationGroupBy.groupBy(source, keySelector, elementSelector));
     }
 
     /**
@@ -1372,7 +1372,7 @@ public static <K, T, R> Observable<GroupedObservable<K, R>> groupBy(Observable<T
      * @return an observable of observable groups, each of which corresponds to a unique key value, containing all elements that share that same key value.
      */
     public static <K, T> Observable<GroupedObservable<K, T>> groupBy(Observable<T> source, final Func1<T, K> keySelector) {
-        return create(OperatorGroupBy.groupBy(source, keySelector));
+        return create(OperationGroupBy.groupBy(source, keySelector));
     }
 
     /**
@@ -2041,7 +2041,7 @@ public Iterator<T> iterator() {
      * @return the iterator that could be used to iterate over the elements of the observable.
      */
     public static <T> Iterator<T> getIterator(Observable<T> that) {
-        return OperatorToIterator.toIterator(that);
+        return OperationToIterator.toIterator(that);
     }
 
     /**

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() {
         }
 
     }
+
 }