@Test @SuppressWarnings("unchecked") public void testOrdering() throws InterruptedException { Observable<String> obs = Observable.from("one", null, "two", "three", "four"); Observer<String> observer = mock(Observer.class); InOrder inOrder = inOrder(observer); final CountDownLatch completedLatch = new CountDownLatch(1); doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocation) throws Throwable { completedLatch.countDown(); return null; } }).when(observer).onCompleted(); obs.observeOn(Schedulers.computation()).subscribe(observer); if (!completedLatch.await(1000, TimeUnit.MILLISECONDS)) { fail("timed out waiting"); } inOrder.verify(observer, times(1)).onNext("one"); inOrder.verify(observer, times(1)).onNext(null); inOrder.verify(observer, times(1)).onNext("two"); inOrder.verify(observer, times(1)).onNext("three"); inOrder.verify(observer, times(1)).onNext("four"); inOrder.verify(observer, times(1)).onCompleted(); inOrder.verifyNoMoreInteractions(); }
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);  }