ExecutorScheduler now uses scheduleAtFixedRate instead of recursion when possible.

Author: jmhofer

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

@@ -19,13 +19,15 @@
 import static org.mockito.Mockito.*;
 
 import java.util.Date;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.junit.Test;
 import org.mockito.InOrder;
 import org.mockito.Mockito;
 
+import rx.concurrency.Schedulers;
 import rx.concurrency.TestScheduler;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action0;

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

@@ -44,6 +44,27 @@ public ExecutorScheduler(ScheduledExecutorService executor) {
         this.executor = executor;
     }
 
+    @Override
+    public <T> Subscription schedulePeriodically(final T state, final Func2<Scheduler, T, Subscription> action, long initialDelay, long period, TimeUnit unit) {
+        if (executor instanceof ScheduledExecutorService) {
+            final CompositeSubscription subscriptions = new CompositeSubscription();
+            
+            ScheduledFuture<?> f = ((ScheduledExecutorService) executor).scheduleAtFixedRate(new Runnable() {
+                @Override
+                public void run() {
+                    Subscription s = action.call(ExecutorScheduler.this, state);
+                    subscriptions.add(s);
+                }
+            }, initialDelay, period, unit);
+            
+            subscriptions.add(Subscriptions.create(f));
+            return subscriptions;
+            
+        } else {
+            return super.schedulePeriodically(state, action, initialDelay, period, unit);
+        }
+    }
+    
     @Override
     public <T> Subscription schedule(final T state, final Func2<Scheduler, T, Subscription> action, long delayTime, TimeUnit unit) {
         final DiscardableAction<T> discardableAction = new DiscardableAction<T>(state, action);