package rx;

import static org.mockito.Mockito.*;

import java.util.concurrent.TimeUnit;

import org.junit.Test;

import org.mockito.InOrder;

import rx.concurrency.TestScheduler;

import rx.subjects.PublishSubject;

public class ThrottleFirstTests {

@Test

public void testThrottle() {

@SuppressWarnings("unchecked")

Observer<Integer> observer = mock(Observer.class);

TestScheduler s = new TestScheduler();

PublishSubject<Integer> o = PublishSubject.create();

o.throttleFirst(500, TimeUnit.MILLISECONDS, s).subscribe(observer);

// send events with simulated time increments

s.advanceTimeTo(0, TimeUnit.MILLISECONDS);

o.onNext(1); // deliver

o.onNext(2); // skip

s.advanceTimeTo(501, TimeUnit.MILLISECONDS);

o.onNext(3); // deliver

s.advanceTimeTo(600, TimeUnit.MILLISECONDS);

o.onNext(4); // skip

s.advanceTimeTo(700, TimeUnit.MILLISECONDS);

o.onNext(5); // skip

o.onNext(6); // skip

s.advanceTimeTo(1001, TimeUnit.MILLISECONDS);

o.onNext(7); // deliver

s.advanceTimeTo(1501, TimeUnit.MILLISECONDS);

o.onCompleted();

InOrder inOrder = inOrder(observer);

inOrder.verify(observer).onNext(1);

inOrder.verify(observer).onNext(3);

inOrder.verify(observer).onNext(7);

inOrder.verify(observer).onCompleted();

inOrder.verifyNoMoreInteractions();

}

}