OnSubscribeAmb implements OnSubscribe<T> { public static <T> OnSubscribe<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); } private  OnSubscribeAmb(Iterable<? extends Observable<? extends T>> sources); static OnSubscribe<T> amb(Observable<? extends T> o1, Observable<? extends T> o2); static OnSubscribe<T> amb(Observable<? extends T> o1, Observable<? extends T> o2, Observable<? extends T> o3); static OnSubscribe<T> amb(Observable<? extends T> o1, Observable<? extends T> o2, Observable<? extends T> o3, Observable<? extends T> o4); static OnSubscribe<T> amb(Observable<? extends T> o1, Observable<? extends T> o2, Observable<? extends T> o3, Observable<? extends T> o4, Observable<? extends T> o5); static OnSubscribe<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 OnSubscribe<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 OnSubscribe<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 OnSubscribe<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 OnSubscribe<T> amb(final Iterable<? extends Observable<? extends T>> sources); @Override void call(final Subscriber<? super T> subscriber);  }
@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 = 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 expectedException = 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, expectedException); 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 = 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(expectedException); 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 = mock(Observer.class); o.subscribe(observer); scheduler.advanceTimeBy(100000, TimeUnit.MILLISECONDS); InOrder inOrder = inOrder(observer); inOrder.verify(observer, times(1)).onCompleted(); inOrder.verifyNoMoreInteractions(); }
@Test public void testProducerRequestThroughAmb() { TestSubscriber<Integer> ts = new TestSubscriber<Integer>(); ts.requestMore(3); final AtomicLong requested1 = new AtomicLong(); final AtomicLong requested2 = new AtomicLong(); Observable<Integer> o1 = Observable.create(new OnSubscribe<Integer>() { @Override public void call(Subscriber<? super Integer> s) { s.setProducer(new Producer() { @Override public void request(long n) { System.out.println("1-requested: " + n); requested1.set(n); } }); } }); Observable<Integer> o2 = Observable.create(new OnSubscribe<Integer>() { @Override public void call(Subscriber<? super Integer> s) { s.setProducer(new Producer() { @Override public void request(long n) { System.out.println("2-requested: " + n); requested2.set(n); } }); } }); Observable.amb(o1, o2).subscribe(ts); assertEquals(3, requested1.get()); assertEquals(3, requested2.get()); }