OperationAmb { public static <T> OnSubscribeFunc<T> amb(Observable<? extends T> o1, Observable<? extends T> o2) { List<Observable<? extends T>> sources = new ArrayList<Observable<? extends T>>(); sources.add(o1); sources.add(o2); return amb(sources); }  static OnSubscribeFunc<T> amb(Observable<? extends T> o1, Observable<? extends T> o2); static OnSubscribeFunc<T> amb(Observable<? extends T> o1, Observable<? extends T> o2, Observable<? extends T> o3); static OnSubscribeFunc<T> amb(Observable<? extends T> o1, Observable<? extends T> o2, Observable<? extends T> o3, Observable<? extends T> o4); static OnSubscribeFunc<T> amb(Observable<? extends T> o1, Observable<? extends T> o2, Observable<? extends T> o3, Observable<? extends T> o4, Observable<? extends T> o5); static OnSubscribeFunc<T> amb(Observable<? extends T> o1, Observable<? extends T> o2, Observable<? extends T> o3, Observable<? extends T> o4, Observable<? extends T> o5, Observable<? extends T> o6); static OnSubscribeFunc<T> amb(Observable<? extends T> o1, Observable<? extends T> o2, Observable<? extends T> o3, Observable<? extends T> o4, Observable<? extends T> o5, Observable<? extends T> o6, Observable<? extends T> o7); static OnSubscribeFunc<T> amb(Observable<? extends T> o1, Observable<? extends T> o2, Observable<? extends T> o3, Observable<? extends T> o4, Observable<? extends T> o5, Observable<? extends T> o6, Observable<? extends T> o7, Observable<? extends T> o8); static OnSubscribeFunc<T> amb(Observable<? extends T> o1, Observable<? extends T> o2, Observable<? extends T> o3, Observable<? extends T> o4, Observable<? extends T> o5, Observable<? extends T> o6, Observable<? extends T> o7, Observable<? extends T> o8, Observable<? extends T> o9); static OnSubscribeFunc<T> amb(
            final Iterable<? extends Observable<? extends T>> sources);  }
@Test public void testAmb() { Observable<String> observable1 = createObservable(new String[] { "1", "11", "111", "1111" }, 2000, null); Observable<String> observable2 = createObservable(new String[] { "2", "22", "222", "2222" }, 1000, null); Observable<String> observable3 = createObservable(new String[] { "3", "33", "333", "3333" }, 3000, null); Observable<String> o = Observable.create(amb(observable1, observable2, observable3)); @SuppressWarnings("unchecked") Observer<String> observer = (Observer<String>) mock(Observer.class); o.subscribe(observer); scheduler.advanceTimeBy(100000, TimeUnit.MILLISECONDS); InOrder inOrder = inOrder(observer); inOrder.verify(observer, times(1)).onNext("2"); inOrder.verify(observer, times(1)).onNext("22"); inOrder.verify(observer, times(1)).onNext("222"); inOrder.verify(observer, times(1)).onNext("2222"); inOrder.verify(observer, times(1)).onCompleted(); inOrder.verifyNoMoreInteractions(); }
@Test public void testAmb2() { IOException needHappenedException = new IOException( "fake exception"); Observable<String> observable1 = createObservable(new String[] {}, 2000, new IOException("fake exception")); Observable<String> observable2 = createObservable(new String[] { "2", "22", "222", "2222" }, 1000, needHappenedException); Observable<String> observable3 = createObservable(new String[] {}, 3000, new IOException("fake exception")); Observable<String> o = Observable.create(amb(observable1, observable2, observable3)); @SuppressWarnings("unchecked") Observer<String> observer = (Observer<String>) mock(Observer.class); o.subscribe(observer); scheduler.advanceTimeBy(100000, TimeUnit.MILLISECONDS); InOrder inOrder = inOrder(observer); inOrder.verify(observer, times(1)).onNext("2"); inOrder.verify(observer, times(1)).onNext("22"); inOrder.verify(observer, times(1)).onNext("222"); inOrder.verify(observer, times(1)).onNext("2222"); inOrder.verify(observer, times(1)).onError(needHappenedException); inOrder.verifyNoMoreInteractions(); }
@Test public void testAmb3() { Observable<String> observable1 = createObservable(new String[] { "1" }, 2000, null); Observable<String> observable2 = createObservable(new String[] {}, 1000, null); Observable<String> observable3 = createObservable(new String[] { "3" }, 3000, null); Observable<String> o = Observable.create(amb(observable1, observable2, observable3)); @SuppressWarnings("unchecked") Observer<String> observer = (Observer<String>) mock(Observer.class); o.subscribe(observer); scheduler.advanceTimeBy(100000, TimeUnit.MILLISECONDS); InOrder inOrder = inOrder(observer); inOrder.verify(observer, times(1)).onCompleted(); inOrder.verifyNoMoreInteractions(); }