@Test @SuppressWarnings("unchecked") public void testThreadName() throws InterruptedException { System.out.println("Main Thread: " + Thread.currentThread().getName()); Observable<String> obs = Observable.from("one", null, "two", "three", "four"); Observer<String> observer = mock(Observer.class); final String parentThreadName = Thread.currentThread().getName(); final CountDownLatch completedLatch = new CountDownLatch(1); obs = obs.doOnNext(new Action1<String>() { @Override public void call(String s) { String threadName = Thread.currentThread().getName(); System.out.println("Source ThreadName: " + threadName + " Expected => " + parentThreadName); assertEquals(parentThreadName, threadName); } }); obs.observeOn(Schedulers.newThread()).doOnNext(new Action1<String>() { @Override public void call(String t1) { String threadName = Thread.currentThread().getName(); boolean correctThreadName = threadName.startsWith("RxNewThreadScheduler"); System.out.println("ObserveOn ThreadName: " + threadName + " Correct => " + correctThreadName); assertTrue(correctThreadName); } }).finallyDo(new Action0() { @Override public void call() { completedLatch.countDown(); } }).subscribe(observer); if (!completedLatch.await(1000, TimeUnit.MILLISECONDS)) { fail("timed out waiting"); } verify(observer, never()).onError(any(Throwable.class)); verify(observer, times(5)).onNext(any(String.class)); verify(observer, times(1)).onCompleted(); }
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);  }