Merge pull request ReactiveX#1903 from benjchristensen/testscheduler-1790

Fix TestScheduler Handling of Immediate vs Virtual Time

Author: benjchristensen

src/main/java/rx/schedulers/TestScheduler.java

@@ -111,7 +111,8 @@ private void triggerActions(long targetTimeInNanos) {
             if (current.time > targetTimeInNanos) {
                 break;
             }
-            time = current.time;
+            // if scheduled time is 0 (immediate) use current virtual time
+            time = current.time == 0 ? time : current.time;
             queue.remove();
 
             // Only execute if not unsubscribed
@@ -129,7 +130,7 @@ public Worker createWorker() {
 
     private final class InnerTestScheduler extends Worker {
 
-        private BooleanSubscription s = new BooleanSubscription();
+        private final BooleanSubscription s = new BooleanSubscription();
 
         @Override
         public void unsubscribe() {

src/test/java/rx/schedulers/TestSchedulerTest.java

@@ -29,7 +29,10 @@
 import org.mockito.InOrder;
 import org.mockito.Mockito;
 
+import rx.Observable;
+import rx.Observable.OnSubscribe;
 import rx.Scheduler;
+import rx.Subscriber;
 import rx.Subscription;
 import rx.functions.Action0;
 import rx.functions.Func1;
@@ -160,4 +163,42 @@ public void call() {
         assertEquals(0, counter.get());
     }
 
+    @Test
+    public final void testNestedSchedule() {
+        final TestScheduler scheduler = new TestScheduler();
+        final Scheduler.Worker inner = scheduler.createWorker();
+        final Action0 calledOp = mock(Action0.class);
+
+        Observable<Object> poller;
+        poller = Observable.create(new OnSubscribe<Object>() {
+            @Override
+            public void call(final Subscriber<? super Object> aSubscriber) {
+                inner.schedule(new Action0() {
+                    @Override
+                    public void call() {
+                        if (!aSubscriber.isUnsubscribed()) {
+                            calledOp.call();
+                            inner.schedule(this, 5, TimeUnit.SECONDS);
+                        }
+                    }
+                });
+            }
+        });
+
+        InOrder inOrder = Mockito.inOrder(calledOp);
+
+        Subscription sub;
+        sub = poller.subscribe();
+
+        scheduler.advanceTimeTo(6, TimeUnit.SECONDS);
+        inOrder.verify(calledOp, times(2)).call();
+
+        sub.unsubscribe();
+        scheduler.advanceTimeTo(11, TimeUnit.SECONDS);
+        inOrder.verify(calledOp, never()).call();
+
+        sub = poller.subscribe();
+        scheduler.advanceTimeTo(12, TimeUnit.SECONDS);
+        inOrder.verify(calledOp, times(1)).call();
+    }
 }