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);  }
@Test @SuppressWarnings("unchecked") public void testObserveOn() { Observer<Integer> observer = mock(Observer.class); Observable.create(observeOn(Observable.from(1, 2, 3), Schedulers.immediate())).subscribe(observer); verify(observer, times(1)).onNext(1); verify(observer, times(1)).onNext(2); verify(observer, times(1)).onNext(3); verify(observer, times(1)).onCompleted(); }
@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(); }
@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(); }
@Test public void observeOnTheSameSchedulerTwice() { TestScheduler scheduler = new TestScheduler(); Observable<Integer> o = Observable.from(1, 2, 3); Observable<Integer> o2 = o.observeOn(scheduler); @SuppressWarnings("unchecked") Observer<Object> observer1 = mock(Observer.class); @SuppressWarnings("unchecked") Observer<Object> observer2 = mock(Observer.class); InOrder inOrder1 = inOrder(observer1); InOrder inOrder2 = inOrder(observer2); o2.subscribe(observer1); o2.subscribe(observer2); scheduler.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(); }
@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(); }
@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")); } }); }
@Test public void testObserveOnWithThreadPoolScheduler() { 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.computation()) .toBlockingObservable().forEach(new Action1<Integer>() { @Override public void call(Integer t1) { assertEquals(count.incrementAndGet() * _multiple, t1.intValue()); assertTrue(Thread.currentThread().getName().startsWith("RxComputationThreadPool")); } }); }
@Test public void testObserveOnOrderingConcurrency() { final AtomicInteger count = new AtomicInteger(); final int _multiple = 99; Observable.range(1, 10000).map(new Func1<Integer, Integer>() { @Override public Integer call(Integer t1) { if (randomIntFrom0to100() > 98) { try { Thread.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); } } return t1 * _multiple; } }).observeOn(Schedulers.computation()) .toBlockingObservable().forEach(new Action1<Integer>() { @Override public void call(Integer t1) { assertEquals(count.incrementAndGet() * _multiple, t1.intValue()); assertTrue(Thread.currentThread().getName().startsWith("RxComputationThreadPool")); } }); }