@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");    // 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));
    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();
}