OperationThrottleFirst { public static <T> OnSubscribeFunc<T> throttleFirst(Observable<T> items, long windowDuration, TimeUnit unit) { return throttleFirst(items, windowDuration, unit, Schedulers.computation()); }  static OnSubscribeFunc<T> throttleFirst(Observable<T> items, long windowDuration, TimeUnit unit); static OnSubscribeFunc<T> throttleFirst(final Observable<T> items, final long windowDuration, final TimeUnit unit, final Scheduler scheduler);  }
@Test public void testThrottlingWithCompleted() { Observable<String> source = Observable.create(new Observable.OnSubscribeFunc<String>() { @Override public Subscription onSubscribe(Observer<? super String> observer) { publishNext(observer, 100, "one"); publishNext(observer, 300, "two"); publishNext(observer, 900, "three"); publishNext(observer, 905, "four"); publishCompleted(observer, 1000); return Subscriptions.empty(); } }); Observable<String> sampled = Observable.create(OperationThrottleFirst.throttleFirst(source, 400, TimeUnit.MILLISECONDS, scheduler)); sampled.subscribe(observer); InOrder inOrder = inOrder(observer); scheduler.advanceTimeTo(1000, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onNext("one"); inOrder.verify(observer, times(0)).onNext("two"); inOrder.verify(observer, times(1)).onNext("three"); inOrder.verify(observer, times(0)).onNext("four"); inOrder.verify(observer, times(1)).onCompleted(); inOrder.verifyNoMoreInteractions(); }
@Test public void testThrottlingWithError() { Observable<String> source = Observable.create(new Observable.OnSubscribeFunc<String>() { @Override public Subscription onSubscribe(Observer<? super String> observer) { Exception error = new TestException(); publishNext(observer, 100, "one"); publishNext(observer, 200, "two"); publishError(observer, 300, error); return Subscriptions.empty(); } }); Observable<String> sampled = Observable.create(OperationThrottleFirst.throttleFirst(source, 400, TimeUnit.MILLISECONDS, scheduler)); sampled.subscribe(observer); InOrder inOrder = inOrder(observer); scheduler.advanceTimeTo(400, TimeUnit.MILLISECONDS); inOrder.verify(observer).onNext("one"); inOrder.verify(observer).onError(any(TestException.class)); inOrder.verifyNoMoreInteractions(); }