@Test public void observeSameOnMultipleSchedulers() { TestScheduler scheduler1 = new TestScheduler(); TestScheduler scheduler2 = new TestScheduler(); Observable<Integer> o = Observable.from(1, 2, 3); Observable<Integer> o1 = o.observeOn(scheduler1); Observable<Integer> o2 = o.observeOn(scheduler2); @SuppressWarnings("unchecked") Observer<Object> observer1 = mock(Observer.class); @SuppressWarnings("unchecked") Observer<Object> observer2 = mock(Observer.class); InOrder inOrder1 = inOrder(observer1); InOrder inOrder2 = inOrder(observer2); o1.subscribe(observer1); o2.subscribe(observer2); scheduler1.advanceTimeBy(1, TimeUnit.SECONDS); scheduler2.advanceTimeBy(1, TimeUnit.SECONDS); inOrder1.verify(observer1, times(1)).onNext(1); inOrder1.verify(observer1, times(1)).onNext(2); inOrder1.verify(observer1, times(1)).onNext(3); inOrder1.verify(observer1, times(1)).onCompleted(); verify(observer1, never()).onError(any(Throwable.class)); inOrder1.verifyNoMoreInteractions(); inOrder2.verify(observer2, times(1)).onNext(1); inOrder2.verify(observer2, times(1)).onNext(2); inOrder2.verify(observer2, times(1)).onNext(3); inOrder2.verify(observer2, times(1)).onCompleted(); verify(observer2, never()).onError(any(Throwable.class)); inOrder2.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);  }