@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");    // Should be published since "two" will arrive after the timeout expires.
            publishNext(observer, 600, "two");    // Should be skipped since onError will arrive before the timeout expires.
            publishError(observer, 700, error);   // Should be published as soon as the timeout expires.
            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);
    // 100 + 400 means it triggers at 500
    scheduler.advanceTimeTo(500, TimeUnit.MILLISECONDS);
    inOrder.verify(observer).onNext("one");
    scheduler.advanceTimeTo(701, TimeUnit.MILLISECONDS);
    inOrder.verify(observer).onError(any(TestException.class));
    inOrder.verifyNoMoreInteractions();
}