Implement Scheduler method with dueTime

- added method: schedule(T state, Func2<Scheduler, T, Subscription> action, Date dueTime)

Author: benjchristensen

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

@@ -15,6 +15,7 @@
  */
 package rx;
 
+import java.util.Date;
 import java.util.concurrent.TimeUnit;
 
 import rx.subscriptions.Subscriptions;
@@ -37,7 +38,8 @@
  * <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>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>
  */
@@ -69,6 +71,27 @@ public abstract class Scheduler {
      */
     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.
      * 

rxjava-core/src/test/java/rx/concurrency/TestSchedulers.java

@@ -18,6 +18,7 @@
 import static org.junit.Assert.*;
 import static org.mockito.Mockito.*;
 
+import java.util.Date;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -359,4 +360,39 @@ public void onNext(Integer args) {
         assertTrue(completed.get());
     }
 
+    @Test
+    public void testSchedulingWithDueTime() throws InterruptedException {
+
+        final CountDownLatch latch = new CountDownLatch(5);
+        final AtomicInteger counter = new AtomicInteger();
+
+        long start = System.currentTimeMillis();
+
+        Schedulers.threadPoolForComputation().schedule(null, new Func2<Scheduler, String, Subscription>() {
+
+            @Override
+            public Subscription call(Scheduler scheduler, String state) {
+                System.out.println("doing work");
+                latch.countDown();
+                counter.incrementAndGet();
+                if (latch.getCount() == 0) {
+                    return Subscriptions.empty();
+                } else {
+                    return scheduler.schedule(state, this, new Date(System.currentTimeMillis() + 50));
+                }
+            }
+        }, new Date(System.currentTimeMillis() + 100));
+
+        if (!latch.await(3000, TimeUnit.MILLISECONDS)) {
+            fail("didn't execute ... timed out");
+        }
+
+        long end = System.currentTimeMillis();
+
+        assertEquals(5, counter.get());
+        if ((end - start) < 250) {
+            fail("it should have taken over 250ms since each step was scheduled 50ms in the future");
+        }
+    }
+
 }