slidingWindow in RxJava

SlidingWindow.java

import fj.P;
import fj.P2;
import rx.Observable;
import rx.Subscriber;

public final class SlidingWindow {

  public static Observable<P2<Integer, Integer>> viaScan(final Observable<Integer> integers) {
    // allocate new Integer outside of integer cache range, so that it can be compared using `==`
    // noinspection UnnecessaryBoxing
    final Integer sentinel = new Integer(133742);
    // noinspection NumberEquality
    return integers
        .scan(P.p(sentinel, sentinel), (previous, x) -> P.p(previous._2(), x))
        .filter(tpl -> tpl._1() != sentinel);
  }

  public static Observable<P2<Integer, Integer>> viaOperator(final Observable<Integer> integers) {
    return integers.<P2<Integer, Integer>>lift(SlidingSubscriber::new);
  }

  public static void main(final String... args) {
    final Observable<Integer> integers = Observable.range(1337, 42);
    viaScan(integers).toBlocking().forEach(System.out::println);
    viaOperator(integers).toBlocking().forEach(System.out::println);
  }

  private static final class SlidingSubscriber extends Subscriber<Integer> {
    private Integer previous;
    private final Subscriber<?super P2<Integer, Integer>> downstream;

    private SlidingSubscriber(final Subscriber<? super P2<Integer, Integer>> downstream) {
      this.downstream = downstream;
    }

    @Override
    public void onCompleted() {
      downstream.onCompleted();
    }

    @Override
    public void onError(Throwable e) {
      downstream.onError(e);
    }

    @Override
    public void onNext(Integer integer) {
      if (previous != null) {
        downstream.onNext(P.p(previous, integer));
      }
      previous = integer;
    }
  }
}

👍 Well done!