@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(); }
public static <T> OnSubscribeFunc<T> throttleFirst(Observable<T> items, long windowDuration, TimeUnit unit) { return throttleFirst(items, windowDuration, unit, Schedulers.computation()); }
OperationThrottleFirst { public static <T> OnSubscribeFunc<T> throttleFirst(Observable<T> items, long windowDuration, TimeUnit unit) { return throttleFirst(items, windowDuration, unit, Schedulers.computation()); } }
OperationThrottleFirst { public static <T> OnSubscribeFunc<T> throttleFirst(Observable<T> items, long windowDuration, TimeUnit unit) { return throttleFirst(items, windowDuration, unit, Schedulers.computation()); }  }
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); }
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);  }