@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)); }
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);  }