Debounce vs Throttle (RxJava)

SchedulersTest.java

package net.lamtran.rxjava2;

import org.junit.Before;
import org.junit.Test;

import java.util.List;
import java.util.concurrent.TimeUnit;

import rx.observers.TestSubscriber;
import rx.schedulers.TestScheduler;
import rx.subjects.PublishSubject;

import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.core.IsEqual.equalTo;

/**
 * Created by lam on 9/19/16.
 */

public class SchedulersTest {

  private static final int DEFAULT_TIMEOUT = 5000;

  private static final int INTERVAL_TIME = 500;

  private TestScheduler mScheduler;

  private PublishSubject<Integer> mSubject;

  @Before
  public void before() {
    mScheduler = new TestScheduler();
    mSubject = PublishSubject.create();
  }

  @Test(timeout = DEFAULT_TIMEOUT)
  public void testDebounce() throws Exception {
    TestSubscriber<Integer> subscriber = new TestSubscriber<>();
    mSubject.debounce(INTERVAL_TIME, TimeUnit.MILLISECONDS, mScheduler).subscribe(subscriber);

    mScheduler.advanceTimeTo(0, TimeUnit.MILLISECONDS);
    mSubject.onNext(1); // skip
    mSubject.onNext(2); // deliver
    mScheduler.advanceTimeTo(501, TimeUnit.MILLISECONDS);
    mSubject.onNext(3); // skip
    mScheduler.advanceTimeTo(600, TimeUnit.MILLISECONDS);
    mSubject.onNext(4); // skip
    mScheduler.advanceTimeTo(700, TimeUnit.MILLISECONDS);
    mSubject.onNext(5); // skip
    mSubject.onNext(6); // skip
    mScheduler.advanceTimeTo(1001, TimeUnit.MILLISECONDS);
    mSubject.onNext(7); // deliver
    mScheduler.advanceTimeTo(1501, TimeUnit.MILLISECONDS);
    mSubject.onCompleted();

    subscriber.assertNoErrors();
    subscriber.assertCompleted();
    List<Integer> list = subscriber.getOnNextEvents();
    assertThat(list.size(), equalTo(2));
    assertThat(list.get(0), equalTo(2));
    assertThat(list.get(1), equalTo(7));
  }

  @Test
  public void testThrottleLast() throws Exception {
    TestSubscriber<Integer> subscriber = new TestSubscriber<>();
    mSubject.throttleLast(INTERVAL_TIME, TimeUnit.MILLISECONDS, mScheduler).subscribe(subscriber);

    mScheduler.advanceTimeTo(0, TimeUnit.MILLISECONDS);
    mSubject.onNext(1); // skip
    mSubject.onNext(2); // deliver
    mScheduler.advanceTimeTo(501, TimeUnit.MILLISECONDS);
    mSubject.onNext(3); // skip
    mScheduler.advanceTimeTo(600, TimeUnit.MILLISECONDS);
    mSubject.onNext(4); // skip
    mScheduler.advanceTimeTo(700, TimeUnit.MILLISECONDS);
    mSubject.onNext(5); // skip
    mSubject.onNext(6); // deliver
    mScheduler.advanceTimeTo(1001, TimeUnit.MILLISECONDS);
    mSubject.onNext(7); // deliver
    mScheduler.advanceTimeTo(1501, TimeUnit.MILLISECONDS);
    mSubject.onCompleted();

    subscriber.assertNoErrors();
    subscriber.assertCompleted();
    List<Integer> list = subscriber.getOnNextEvents();
    assertThat(list.size(), equalTo(3));
    assertThat(list.get(0), equalTo(2));
    assertThat(list.get(1), equalTo(6));
    assertThat(list.get(2), equalTo(7));
  }

  @Test
  public void testSample() throws Exception {
    TestSubscriber<Integer> subscriber = new TestSubscriber<>();
    mSubject.sample(INTERVAL_TIME, TimeUnit.MILLISECONDS, mScheduler).subscribe(subscriber);

    mScheduler.advanceTimeTo(0, TimeUnit.MILLISECONDS);
    mSubject.onNext(1); // skip
    mSubject.onNext(2); // deliver
    mScheduler.advanceTimeTo(501, TimeUnit.MILLISECONDS);
    mSubject.onNext(3); // skip
    mScheduler.advanceTimeTo(600, TimeUnit.MILLISECONDS);
    mSubject.onNext(4); // skip
    mScheduler.advanceTimeTo(700, TimeUnit.MILLISECONDS);
    mSubject.onNext(5); // skip
    mSubject.onNext(6); // deliver
    mScheduler.advanceTimeTo(1001, TimeUnit.MILLISECONDS);
    mSubject.onNext(7); // deliver
    mScheduler.advanceTimeTo(1501, TimeUnit.MILLISECONDS);
    mSubject.onCompleted();

    subscriber.assertNoErrors();
    subscriber.assertCompleted();
    List<Integer> list = subscriber.getOnNextEvents();
    assertThat(list.size(), equalTo(3));
    assertThat(list.get(0), equalTo(2));
    assertThat(list.get(1), equalTo(6));
    assertThat(list.get(2), equalTo(7));
  }

  @Test(timeout = DEFAULT_TIMEOUT)
  public void testThrottleWithTimeout() throws Exception {
    TestSubscriber<Integer> subscriber = new TestSubscriber<>();
    mSubject.throttleWithTimeout(INTERVAL_TIME, TimeUnit.MILLISECONDS, mScheduler).subscribe(subscriber);

    mScheduler.advanceTimeTo(0, TimeUnit.MILLISECONDS);
    mSubject.onNext(1); // skip
    mSubject.onNext(2); // deliver
    mScheduler.advanceTimeTo(501, TimeUnit.MILLISECONDS);
    mSubject.onNext(3); // skip
    mScheduler.advanceTimeTo(600, TimeUnit.MILLISECONDS);
    mSubject.onNext(4); // skip
    mScheduler.advanceTimeTo(700, TimeUnit.MILLISECONDS);
    mSubject.onNext(5); // skip
    mSubject.onNext(6); // skip
    mScheduler.advanceTimeTo(1001, TimeUnit.MILLISECONDS);
    mSubject.onNext(7); // deliver
    mScheduler.advanceTimeTo(1501, TimeUnit.MILLISECONDS);
    mSubject.onCompleted();

    subscriber.assertNoErrors();
    subscriber.assertCompleted();
    List<Integer> list = subscriber.getOnNextEvents();
    assertThat(list.size(), equalTo(2));
    assertThat(list.get(0), equalTo(2));
    assertThat(list.get(1), equalTo(7));
  }
}