OperatorThrottleFirst

Author: zsxwing

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

@@ -71,7 +71,7 @@
 import rx.operators.OperationTakeTimed;
 import rx.operators.OperationTakeUntil;
 import rx.operators.OperationTakeWhile;
-import rx.operators.OperationThrottleFirst;
+import rx.operators.OperatorThrottleFirst;
 import rx.operators.OperationTimeInterval;
 import rx.operators.OperationTimer;
 import rx.operators.OperationToMap;
@@ -6733,7 +6733,7 @@ public final Observable<T> takeWhileWithIndex(final Func2<? super T, ? super Int
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Filtering-Observables#wiki-throttlefirst">RxJava Wiki: throttleFirst()</a>
      */
     public final Observable<T> throttleFirst(long windowDuration, TimeUnit unit) {
-        return create(OperationThrottleFirst.throttleFirst(this, windowDuration, unit));
+        return lift(new OperatorThrottleFirst<T>(windowDuration, unit, Schedulers.computation()));
     }
 
     /**
@@ -6755,7 +6755,7 @@ public final Observable<T> throttleFirst(long windowDuration, TimeUnit unit) {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Filtering-Observables#wiki-throttlefirst">RxJava Wiki: throttleFirst()</a>
      */
     public final Observable<T> throttleFirst(long skipDuration, TimeUnit unit, Scheduler scheduler) {
-        return create(OperationThrottleFirst.throttleFirst(this, skipDuration, unit, scheduler));
+        return lift(new OperatorThrottleFirst<T>(skipDuration, unit, scheduler));
     }
 
     /**

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

@@ -0,0 +1,63 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import java.util.concurrent.TimeUnit;
+
+import rx.*;
+import rx.Observable.Operator;
+
+/**
+ * Throttle by windowing a stream and returning the first value in each window.
+ */
+public final class OperatorThrottleFirst<T> implements Operator<T, T> {
+
+    private final long timeInMilliseconds;
+    private final Scheduler scheduler;
+
+    public OperatorThrottleFirst(long windowDuration, TimeUnit unit, Scheduler scheduler) {
+        this.timeInMilliseconds = unit.toMillis(windowDuration);
+        this.scheduler = scheduler;
+    }
+
+    @Override
+    public Subscriber<? super T> call(final Subscriber<? super T> subscriber) {
+        return new Subscriber<T>(subscriber) {
+
+            private long lastOnNext = 0;
+
+            @Override
+            public void onNext(T v) {
+                long now = scheduler.now();
+                if (lastOnNext == 0 || now - lastOnNext >= timeInMilliseconds) {
+                    lastOnNext = now;
+                    subscriber.onNext(v);
+                }
+            }
+
+            @Override
+            public void onCompleted() {
+                subscriber.onCompleted();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                subscriber.onError(e);
+            }
+
+        };
+    }
+}

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

@@ -27,14 +27,15 @@
 import org.mockito.InOrder;
 
 import rx.Observable;
+import rx.Observable.OnSubscribe;
 import rx.Observer;
 import rx.Scheduler;
-import rx.Subscription;
+import rx.Subscriber;
 import rx.functions.Action0;
 import rx.schedulers.TestScheduler;
-import rx.subscriptions.Subscriptions;
+import rx.subjects.PublishSubject;
 
-public class OperationThrottleFirstTest {
+public class OperatorThrottleFirstTest {
 
     private TestScheduler scheduler;
     private Scheduler.Worker innerScheduler;
@@ -50,20 +51,18 @@ public void before() {
 
     @Test
     public void testThrottlingWithCompleted() {
-        Observable<String> source = Observable.create(new Observable.OnSubscribeFunc<String>() {
+        Observable<String> source = Observable.create(new OnSubscribe<String>() {
             @Override
-            public Subscription onSubscribe(Observer<? super String> observer) {
+            public void call(Subscriber<? super String> observer) {
                 publishNext(observer, 100, "one");    // publish as it's first
                 publishNext(observer, 300, "two");    // skip as it's last within the first 400
                 publishNext(observer, 900, "three");   // publish
                 publishNext(observer, 905, "four");   // skip
                 publishCompleted(observer, 1000);     // Should be published as soon as the timeout expires.
-
-                return Subscriptions.empty();
             }
         });
 
-        Observable<String> sampled = Observable.create(OperationThrottleFirst.throttleFirst(source, 400, TimeUnit.MILLISECONDS, scheduler));
+        Observable<String> sampled = source.throttleFirst(400, TimeUnit.MILLISECONDS, scheduler);
         sampled.subscribe(observer);
 
         InOrder inOrder = inOrder(observer);
@@ -79,19 +78,17 @@ public Subscription onSubscribe(Observer<? super String> observer) {
 
     @Test
     public void testThrottlingWithError() {
-        Observable<String> source = Observable.create(new Observable.OnSubscribeFunc<String>() {
+        Observable<String> source = Observable.create(new OnSubscribe<String>() {
             @Override
-            public Subscription onSubscribe(Observer<? super String> observer) {
+            public void call(Subscriber<? super String> observer) {
                 Exception error = new TestException();
                 publishNext(observer, 100, "one");    // Should be published since it is first
                 publishNext(observer, 200, "two");    // Should be skipped since onError will arrive before the timeout expires
                 publishError(observer, 300, error);   // Should be published as soon as the timeout expires.
-
-                return Subscriptions.empty();
             }
         });
 
-        Observable<String> sampled = Observable.create(OperationThrottleFirst.throttleFirst(source, 400, TimeUnit.MILLISECONDS, scheduler));
+        Observable<String> sampled = source.throttleFirst(400, TimeUnit.MILLISECONDS, scheduler);
         sampled.subscribe(observer);
 
         InOrder inOrder = inOrder(observer);
@@ -132,4 +129,36 @@ public void call() {
     @SuppressWarnings("serial")
     private class TestException extends Exception {
     }
+
+    @Test
+    public void testThrottle() {
+        @SuppressWarnings("unchecked")
+        Observer<Integer> observer = mock(Observer.class);
+        TestScheduler s = new TestScheduler();
+        PublishSubject<Integer> o = PublishSubject.create();
+        o.throttleFirst(500, TimeUnit.MILLISECONDS, s).subscribe(observer);
+
+        // send events with simulated time increments
+        s.advanceTimeTo(0, TimeUnit.MILLISECONDS);
+        o.onNext(1); // deliver
+        o.onNext(2); // skip
+        s.advanceTimeTo(501, TimeUnit.MILLISECONDS);
+        o.onNext(3); // deliver
+        s.advanceTimeTo(600, TimeUnit.MILLISECONDS);
+        o.onNext(4); // skip
+        s.advanceTimeTo(700, TimeUnit.MILLISECONDS);
+        o.onNext(5); // skip
+        o.onNext(6); // skip
+        s.advanceTimeTo(1001, TimeUnit.MILLISECONDS);
+        o.onNext(7); // deliver
+        s.advanceTimeTo(1501, TimeUnit.MILLISECONDS);
+        o.onCompleted();
+
+        InOrder inOrder = inOrder(observer);
+        inOrder.verify(observer).onNext(1);
+        inOrder.verify(observer).onNext(3);
+        inOrder.verify(observer).onNext(7);
+        inOrder.verify(observer).onCompleted();
+        inOrder.verifyNoMoreInteractions();
+    }
 }