OperationDebounce { public static <T> OnSubscribeFunc<T> debounce(Observable<T> items, long timeout, TimeUnit unit) { return debounce(items, timeout, unit, Schedulers.computation()); }  static OnSubscribeFunc<T> debounce(Observable<T> items, long timeout, TimeUnit unit); static OnSubscribeFunc<T> debounce(final Observable<T> items, final long timeout, final TimeUnit unit, final Scheduler scheduler); static OnSubscribeFunc<T> debounceSelector(
            Observable<? extends T> source,
            Func1<? super T, ? extends Observable<U>> debounceSelector);  }
@Test public void testDebounceWithCompleted() { Observable<String> source = Observable.create(new Observable.OnSubscribeFunc<String>() { @Override public Subscription onSubscribe(Observer<? super String> observer) { publishNext(observer, 100, "one"); publishNext(observer, 400, "two"); publishNext(observer, 900, "three"); publishCompleted(observer, 1000); return Subscriptions.empty(); } }); Observable<String> sampled = Observable.create(OperationDebounce.debounce(source, 400, TimeUnit.MILLISECONDS, scheduler)); sampled.subscribe(observer); scheduler.advanceTimeTo(0, TimeUnit.MILLISECONDS); InOrder inOrder = inOrder(observer); scheduler.advanceTimeTo(800, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onNext("two"); scheduler.advanceTimeTo(1000, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onCompleted(); inOrder.verifyNoMoreInteractions(); }
@Test public void testDebounceNeverEmits() { Observable<String> source = Observable.create(new Observable.OnSubscribeFunc<String>() { @Override public Subscription onSubscribe(Observer<? super String> observer) { publishNext(observer, 100, "a"); publishNext(observer, 200, "b"); publishNext(observer, 300, "c"); publishNext(observer, 400, "d"); publishNext(observer, 500, "e"); publishNext(observer, 600, "f"); publishNext(observer, 700, "g"); publishNext(observer, 800, "h"); publishCompleted(observer, 900); return Subscriptions.empty(); } }); Observable<String> sampled = Observable.create(OperationDebounce.debounce(source, 200, TimeUnit.MILLISECONDS, scheduler)); sampled.subscribe(observer); scheduler.advanceTimeTo(0, TimeUnit.MILLISECONDS); InOrder inOrder = inOrder(observer); inOrder.verify(observer, times(0)).onNext(anyString()); scheduler.advanceTimeTo(1000, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onCompleted(); inOrder.verifyNoMoreInteractions(); }
@Test public void testDebounceWithError() { 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, 600, "two"); publishError(observer, 700, error); return Subscriptions.empty(); } }); Observable<String> sampled = Observable.create(OperationDebounce.debounce(source, 400, TimeUnit.MILLISECONDS, scheduler)); sampled.subscribe(observer); scheduler.advanceTimeTo(0, TimeUnit.MILLISECONDS); InOrder inOrder = inOrder(observer); scheduler.advanceTimeTo(500, TimeUnit.MILLISECONDS); inOrder.verify(observer).onNext("one"); scheduler.advanceTimeTo(701, TimeUnit.MILLISECONDS); inOrder.verify(observer).onError(any(TestException.class)); inOrder.verifyNoMoreInteractions(); }
@Test public void debounceSelectorNormal1() { PublishSubject<Integer> source = PublishSubject.create(); final PublishSubject<Integer> debouncer = PublishSubject.create(); Func1<Integer, Observable<Integer>> debounceSel = new Func1<Integer, Observable<Integer>>() { @Override public Observable<Integer> call(Integer t1) { return debouncer; } }; @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); InOrder inOrder = inOrder(o); source.debounce(debounceSel).subscribe(o); source.onNext(1); debouncer.onNext(1); source.onNext(2); source.onNext(3); source.onNext(4); debouncer.onNext(2); source.onNext(5); source.onCompleted(); inOrder.verify(o).onNext(1); inOrder.verify(o).onNext(4); inOrder.verify(o).onNext(5); inOrder.verify(o).onCompleted(); verify(o, never()).onError(any(Throwable.class)); }
@Test public void debounceSelectorFuncThrows() { PublishSubject<Integer> source = PublishSubject.create(); Func1<Integer, Observable<Integer>> debounceSel = new Func1<Integer, Observable<Integer>>() { @Override public Observable<Integer> call(Integer t1) { throw new OperationReduceTest.CustomException(); } }; @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); source.debounce(debounceSel).subscribe(o); source.onNext(1); verify(o, never()).onNext(any()); verify(o, never()).onCompleted(); verify(o).onError(any(OperationReduceTest.CustomException.class)); }
@Test public void debounceSelectorObservableThrows() { PublishSubject<Integer> source = PublishSubject.create(); Func1<Integer, Observable<Integer>> debounceSel = new Func1<Integer, Observable<Integer>>() { @Override public Observable<Integer> call(Integer t1) { return Observable.error(new OperationReduceTest.CustomException()); } }; @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); source.debounce(debounceSel).subscribe(o); source.onNext(1); verify(o, never()).onNext(any()); verify(o, never()).onCompleted(); verify(o).onError(any(OperationReduceTest.CustomException.class)); }