@Test @SuppressWarnings("unchecked") public void testSubscribeOn() { Observable<Integer> w = Observable.from(1, 2, 3); Scheduler scheduler = spy(OperatorTester.forwardingScheduler(Schedulers.immediate())); Observer<Integer> observer = mock(Observer.class); Subscription subscription = Observable.create(subscribeOn(w, scheduler)).subscribe(observer); verify(scheduler, times(1)).schedule(any(Action1.class)); subscription.unsubscribe(); verify(scheduler, times(1)).schedule(any(Action1.class)); verifyNoMoreInteractions(scheduler); verify(observer, times(1)).onNext(1); verify(observer, times(1)).onNext(2); verify(observer, times(1)).onNext(3); verify(observer, times(1)).onCompleted(); }
public static <T> OnSubscribeFunc<T> subscribeOn(Observable<? extends T> source, Scheduler scheduler) { return new SubscribeOn<T>(source, scheduler); }
OperationSubscribeOn { public static <T> OnSubscribeFunc<T> subscribeOn(Observable<? extends T> source, Scheduler scheduler) { return new SubscribeOn<T>(source, scheduler); } }
OperationSubscribeOn { public static <T> OnSubscribeFunc<T> subscribeOn(Observable<? extends T> source, Scheduler scheduler) { return new SubscribeOn<T>(source, scheduler); }  }
OperationSubscribeOn { public static <T> OnSubscribeFunc<T> subscribeOn(Observable<? extends T> source, Scheduler scheduler) { return new SubscribeOn<T>(source, scheduler); }  static OnSubscribeFunc<T> subscribeOn(Observable<? extends T> source, Scheduler scheduler); }
OperationSubscribeOn { public static <T> OnSubscribeFunc<T> subscribeOn(Observable<? extends T> source, Scheduler scheduler) { return new SubscribeOn<T>(source, scheduler); }  static OnSubscribeFunc<T> subscribeOn(Observable<? extends T> source, Scheduler scheduler);  }