OperationSample { public static <T> OnSubscribeFunc<T> sample(final Observable<? extends T> source, long period, TimeUnit unit) { return new Sample<T>(source, period, unit, Schedulers.computation()); }  static OnSubscribeFunc<T> sample(final Observable<? extends T> source, long period, TimeUnit unit); static OnSubscribeFunc<T> sample(final Observable<? extends T> source, long period, TimeUnit unit, Scheduler scheduler);  }
@Test public void testSample() { Observable<Long> source = Observable.create(new Observable.OnSubscribeFunc<Long>() { @Override public Subscription onSubscribe(final Observer<? super Long> observer1) { scheduler.schedule(new Action1<Inner>() { @Override public void call(Inner inner) { observer1.onNext(1L); } }, 1, TimeUnit.SECONDS); scheduler.schedule(new Action1<Inner>() { @Override public void call(Inner inner) { observer1.onNext(2L); } }, 2, TimeUnit.SECONDS); scheduler.schedule(new Action1<Inner>() { @Override public void call(Inner inner) { observer1.onCompleted(); } }, 3, TimeUnit.SECONDS); return Subscriptions.empty(); } }); Observable<Long> sampled = Observable.create(OperationSample.sample(source, 400L, TimeUnit.MILLISECONDS, scheduler)); sampled.subscribe(observer); InOrder inOrder = inOrder(observer); scheduler.advanceTimeTo(800L, TimeUnit.MILLISECONDS); verify(observer, never()).onNext(any(Long.class)); verify(observer, never()).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); scheduler.advanceTimeTo(1200L, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onNext(1L); verify(observer, never()).onNext(2L); verify(observer, never()).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); scheduler.advanceTimeTo(1600L, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onNext(1L); verify(observer, never()).onNext(2L); verify(observer, never()).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); scheduler.advanceTimeTo(2000L, TimeUnit.MILLISECONDS); inOrder.verify(observer, never()).onNext(1L); inOrder.verify(observer, times(1)).onNext(2L); verify(observer, never()).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); scheduler.advanceTimeTo(3000L, TimeUnit.MILLISECONDS); inOrder.verify(observer, never()).onNext(1L); inOrder.verify(observer, times(2)).onNext(2L); verify(observer, times(1)).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); }
@Test public void sampleWithSamplerNormal() { PublishSubject<Integer> source = PublishSubject.create(); PublishSubject<Integer> sampler = PublishSubject.create(); Observable<Integer> m = source.sample(sampler); m.subscribe(observer2); source.onNext(1); source.onNext(2); sampler.onNext(1); source.onNext(3); source.onNext(4); sampler.onNext(2); source.onCompleted(); sampler.onNext(3); InOrder inOrder = inOrder(observer2); inOrder.verify(observer2, never()).onNext(1); inOrder.verify(observer2, times(1)).onNext(2); inOrder.verify(observer2, never()).onNext(3); inOrder.verify(observer2, times(1)).onNext(4); inOrder.verify(observer2, times(1)).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); }
@Test public void sampleWithSamplerNoDuplicates() { PublishSubject<Integer> source = PublishSubject.create(); PublishSubject<Integer> sampler = PublishSubject.create(); Observable<Integer> m = source.sample(sampler); m.subscribe(observer2); source.onNext(1); source.onNext(2); sampler.onNext(1); sampler.onNext(1); source.onNext(3); source.onNext(4); sampler.onNext(2); sampler.onNext(2); source.onCompleted(); sampler.onNext(3); InOrder inOrder = inOrder(observer2); inOrder.verify(observer2, never()).onNext(1); inOrder.verify(observer2, times(1)).onNext(2); inOrder.verify(observer2, never()).onNext(3); inOrder.verify(observer2, times(1)).onNext(4); inOrder.verify(observer2, times(1)).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); }
@Test public void sampleWithSamplerTerminatingEarly() { PublishSubject<Integer> source = PublishSubject.create(); PublishSubject<Integer> sampler = PublishSubject.create(); Observable<Integer> m = source.sample(sampler); m.subscribe(observer2); source.onNext(1); source.onNext(2); sampler.onNext(1); sampler.onCompleted(); source.onNext(3); source.onNext(4); InOrder inOrder = inOrder(observer2); inOrder.verify(observer2, never()).onNext(1); inOrder.verify(observer2, times(1)).onNext(2); inOrder.verify(observer2, times(1)).onCompleted(); inOrder.verify(observer2, never()).onNext(any()); verify(observer, never()).onError(any(Throwable.class)); }
@Test public void sampleWithSamplerEmitAndTerminate() { PublishSubject<Integer> source = PublishSubject.create(); PublishSubject<Integer> sampler = PublishSubject.create(); Observable<Integer> m = source.sample(sampler); m.subscribe(observer2); source.onNext(1); source.onNext(2); sampler.onNext(1); source.onNext(3); source.onCompleted(); sampler.onNext(2); sampler.onCompleted(); InOrder inOrder = inOrder(observer2); inOrder.verify(observer2, never()).onNext(1); inOrder.verify(observer2, times(1)).onNext(2); inOrder.verify(observer2, never()).onNext(3); inOrder.verify(observer2, times(1)).onCompleted(); inOrder.verify(observer2, never()).onNext(any()); verify(observer, never()).onError(any(Throwable.class)); }
@Test public void sampleWithSamplerEmptySource() { PublishSubject<Integer> source = PublishSubject.create(); PublishSubject<Integer> sampler = PublishSubject.create(); Observable<Integer> m = source.sample(sampler); m.subscribe(observer2); source.onCompleted(); sampler.onNext(1); InOrder inOrder = inOrder(observer2); inOrder.verify(observer2, times(1)).onCompleted(); verify(observer2, never()).onNext(any()); verify(observer, never()).onError(any(Throwable.class)); }
@Test public void sampleWithSamplerSourceThrows() { PublishSubject<Integer> source = PublishSubject.create(); PublishSubject<Integer> sampler = PublishSubject.create(); Observable<Integer> m = source.sample(sampler); m.subscribe(observer2); source.onNext(1); source.onError(new RuntimeException("Forced failure!")); sampler.onNext(1); InOrder inOrder = inOrder(observer2); inOrder.verify(observer2, times(1)).onError(any(Throwable.class)); verify(observer2, never()).onNext(any()); verify(observer, never()).onCompleted(); }
@Test public void sampleWithSamplerThrows() { PublishSubject<Integer> source = PublishSubject.create(); PublishSubject<Integer> sampler = PublishSubject.create(); Observable<Integer> m = source.sample(sampler); m.subscribe(observer2); source.onNext(1); sampler.onNext(1); sampler.onError(new RuntimeException("Forced failure!")); InOrder inOrder = inOrder(observer2); inOrder.verify(observer2, times(1)).onNext(1); inOrder.verify(observer2, times(1)).onError(any(RuntimeException.class)); verify(observer, never()).onCompleted(); }