@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)); }
public static <T> OnSubscribeFunc<T> sample(final Observable<? extends T> source, long period, TimeUnit unit) { return new Sample<T>(source, period, unit, Schedulers.computation()); }
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()); } }
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()); }  }
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); }
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);  }