switched OperationInterval implementation over to use periodic scheduling

Author: jmhofer

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

@@ -15,11 +15,8 @@
  */
 package rx;
 
-import static org.mockito.Matchers.anyLong;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
+import static org.mockito.Matchers.*;
+import static org.mockito.Mockito.*;
 
 import java.util.Date;
 import java.util.concurrent.TimeUnit;

rxjava-core/src/main/java/rx/operators/OperationInterval.java

@@ -20,7 +20,6 @@
 
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -57,47 +56,35 @@ public static Func1<Observer<Long>, Subscription> interval(long interval, TimeUn
     }
 
     private static class Interval implements Func1<Observer<Long>, Subscription> {
-        private final long interval;
+        private final long period;
         private final TimeUnit unit;
         private final Scheduler scheduler;
 
         private long currentValue;
-        private final AtomicBoolean complete = new AtomicBoolean();
 
-        private Interval(long interval, TimeUnit unit, Scheduler scheduler) {
-            this.interval = interval;
+        private Interval(long period, TimeUnit unit, Scheduler scheduler) {
+            this.period = period;
             this.unit = unit;
             this.scheduler = scheduler;
         }
 
         @Override
         public Subscription call(final Observer<Long> observer) {
-            scheduler.schedule(new IntervalAction(observer), interval, unit);
-            return Subscriptions.create(new Action0() {
+            final Subscription wrapped = scheduler.schedulePeriodically(new Action0() {
                 @Override
                 public void call() {
-                    complete.set(true);
+                    observer.onNext(currentValue);
+                    currentValue++;
                 }
-            });
-        }
-        
-        private class IntervalAction implements Action0 {
-            private final Observer<Long> observer;
-            
-            private IntervalAction(Observer<Long> observer) {
-                this.observer = observer;
-            }
+            }, period, period, unit);
 
-            @Override
-            public void call() {
-                if (complete.get()) {
+            return Subscriptions.create(new Action0() {
+                @Override
+                public void call() {
+                    wrapped.unsubscribe();
                     observer.onCompleted();
-                } else {
-                    observer.onNext(currentValue);
-                    currentValue++;
-                    scheduler.schedule(this, interval, unit);
                 }
-            }
+            });
         }
     }