Operation: throttle

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

@@ -65,6 +65,7 @@
 import rx.operators.OperationTakeLast;
 import rx.operators.OperationTakeUntil;
 import rx.operators.OperationTakeWhile;
+import rx.operators.OperationThrottle;
 import rx.operators.OperationTimestamp;
 import rx.operators.OperationToFuture;
 import rx.operators.OperationToIterator;
@@ -2095,6 +2096,29 @@ public Boolean call(T t, Integer integer)
         }));
     }
 
+    /**
+     * Throttles the {@link Observable} by dropping values which are followed by newer values before the timer has expired.
+     * 
+     * @param timeout    The time each value has to be 'the most recent' of the {@link Observable} to ensure that it's not dropped.
+     * @param unit       The {@link TimeUnit} for the timeout.
+     * @return An {@link Observable} which filters out values which are too quickly followed up with never values.
+     */
+    public Observable<T> throttle(long timeout, TimeUnit unit) {
+        return create(OperationThrottle.throttle(this, timeout, unit));
+    }
+
+    /**
+     * Throttles the {@link Observable} by dropping values which are followed by newer values before the timer has expired.
+     * 
+     * @param timeout    The time each value has to be 'the most recent' of the {@link Observable} to ensure that it's not dropped.
+     * @param unit       The {@link TimeUnit} for the timeout.
+     * @param scheduler  The {@link Scheduler} to use when timing incoming values.
+     * @return An {@link Observable} which filters out values which are too quickly followed up with never values.
+     */
+    public Observable<T> throttle(long timeout, TimeUnit unit, Scheduler scheduler) {
+        return create(OperationThrottle.throttle(this, timeout, unit, scheduler));
+    }
+
     /**
      * Adds a timestamp to each item emitted by this observable.
      * @return An observable sequence of timestamped items.

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

@@ -19,20 +19,22 @@
 import java.util.PriorityQueue;
 import java.util.Queue;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import rx.Scheduler;
 import rx.Subscription;
-import rx.subscriptions.Subscriptions;
 import rx.util.functions.Func2;
 
 public class TestScheduler extends Scheduler {
     private final Queue<TimedAction<?>> queue = new PriorityQueue<TimedAction<?>>(11, new CompareActionsByTime());
 
     private static class TimedAction<T> {
+
         private final long time;
         private final Func2<Scheduler, T, Subscription> action;
         private final T state;
         private final TestScheduler scheduler;
+        private final AtomicBoolean isCancelled = new AtomicBoolean(false);
 
         private TimedAction(TestScheduler scheduler, long time, Func2<Scheduler, T, Subscription> action, T state) {
             this.time = time;
@@ -41,6 +43,10 @@ private TimedAction(TestScheduler scheduler, long time, Func2<Scheduler, T, Subs
             this.scheduler = scheduler;
         }
 
+        public void cancel() {
+            isCancelled.set(true);
+        }
+
         @Override
         public String toString() {
             return String.format("TimedAction(time = %d, action = %s)", time, action.toString());
@@ -85,8 +91,12 @@ private void triggerActions(long targetTimeInNanos) {
             }
             time = current.time;
             queue.remove();
-            // because the queue can have wildcards we have to ignore the type T for the state
-            ((Func2<Scheduler, Object, Subscription>) current.action).call(current.scheduler, current.state);
+
+            // Only execute if the TimedAction has not yet been cancelled
+            if (!current.isCancelled.get()) {
+                // because the queue can have wildcards we have to ignore the type T for the state
+                ((Func2<Scheduler, Object, Subscription>) current.action).call(current.scheduler, current.state);
+            }
         }
     }
 
@@ -97,7 +107,14 @@ public <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> acti
 
     @Override
     public <T> Subscription schedule(T state, Func2<Scheduler, T, Subscription> action, long delayTime, TimeUnit unit) {
-        queue.add(new TimedAction<T>(this, time + unit.toNanos(delayTime), action, state));
-        return Subscriptions.empty();
+        final TimedAction<T> timedAction = new TimedAction<T>(this, time + unit.toNanos(delayTime), action, state);
+        queue.add(timedAction);
+
+        return new Subscription() {
+            @Override
+            public void unsubscribe() {
+                timedAction.cancel();
+            }
+        };
     }
 }