RxComboDetector { @VisibleForTesting static Flowable<Integer> detect(Flowable<Integer> clicks, final long maxIntervalMillis, final int minComboTimesCared) { return clicks.timestamp() .scan((lastOne, thisOne) -> { if (thisOne.time() - lastOne.time() <= maxIntervalMillis) { return new Timed<>(lastOne.value() + 1, thisOne.time(), thisOne.unit()); } else { return new Timed<>(1, thisOne.time(), thisOne.unit()); } }) .map(Timed::value) .filter(combo -> combo >= minComboTimesCared); } private  RxComboDetector(View target, long maxIntervalMillis, int minComboTimesCared); Flowable<Integer> start(); @Deprecated rx.Observable<Integer> startAsV1Observable();  }
@Test public void testDetect() { Flowable<Integer> mockClicks = Flowable .create(emitter -> { emitter.onNext(1); sleepSilently(50); emitter.onNext(1); sleepSilently(50); emitter.onNext(1); sleepSilently(50); emitter.onNext(1); sleepSilently(150); emitter.onNext(1); sleepSilently(50); emitter.onNext(1); sleepSilently(150); emitter.onNext(1); sleepSilently(150); emitter.onNext(1); emitter.onComplete(); }, BackpressureStrategy.BUFFER); TestSubscriber<Integer> subscriber = new TestSubscriber<>(); RxComboDetector.detect(mockClicks, 100, 2) .subscribeOn(Schedulers.from(command -> new Thread(command).start())) .subscribe(subscriber); subscriber.awaitTerminalEvent(30, TimeUnit.SECONDS); subscriber.assertValueCount(4); subscriber.assertValues(2, 3, 4, 2); }