@Test public void testObserveOnWithNewThreadScheduler() { final AtomicInteger count = new AtomicInteger(); final int _multiple = 99; Observable.range(1, 100000).map(new Func1<Integer, Integer>() { @Override public Integer call(Integer t1) { return t1 * _multiple; } }).observeOn(Schedulers.newThread()) .toBlockingObservable().forEach(new Action1<Integer>() { @Override public void call(Integer t1) { assertEquals(count.incrementAndGet() * _multiple, t1.intValue()); assertTrue(Thread.currentThread().getName().startsWith("RxNewThreadScheduler")); } }); }
public static <T> OnSubscribeFunc<T> observeOn(Observable<? extends T> source, Scheduler scheduler) { return new ObserveOn<T>(source, scheduler); }
OperationObserveOn { public static <T> OnSubscribeFunc<T> observeOn(Observable<? extends T> source, Scheduler scheduler) { return new ObserveOn<T>(source, scheduler); } }
OperationObserveOn { public static <T> OnSubscribeFunc<T> observeOn(Observable<? extends T> source, Scheduler scheduler) { return new ObserveOn<T>(source, scheduler); }  }
OperationObserveOn { public static <T> OnSubscribeFunc<T> observeOn(Observable<? extends T> source, Scheduler scheduler) { return new ObserveOn<T>(source, scheduler); }  static OnSubscribeFunc<T> observeOn(Observable<? extends T> source, Scheduler scheduler); }
OperationObserveOn { public static <T> OnSubscribeFunc<T> observeOn(Observable<? extends T> source, Scheduler scheduler) { return new ObserveOn<T>(source, scheduler); }  static OnSubscribeFunc<T> observeOn(Observable<? extends T> source, Scheduler scheduler);  }