OperationMergeDelayError { public static <T> OnSubscribeFunc<T> mergeDelayError(final Observable<? extends Observable<? extends T>> sequences) { return new OnSubscribeFunc<T>() { @Override public Subscription onSubscribe(Observer<? super T> observer) { return new MergeDelayErrorObservable<T>(sequences).onSubscribe(observer); } }; }  static OnSubscribeFunc<T> mergeDelayError(final Observable<? extends Observable<? extends T>> sequences); static OnSubscribeFunc<T> mergeDelayError(final Observable<? extends T>... sequences); static OnSubscribeFunc<T> mergeDelayError(final List<? extends Observable<? extends T>> sequences);  }
@Test public void testErrorDelayed1() { final Observable<String> o1 = Observable.create(new TestErrorObservable("four", null, "six")); final Observable<String> o2 = Observable.create(new TestErrorObservable("one", "two", "three")); @SuppressWarnings("unchecked") Observable<String> m = Observable.create(mergeDelayError(o1, o2)); m.subscribe(stringObserver); verify(stringObserver, times(1)).onError(any(NullPointerException.class)); verify(stringObserver, never()).onCompleted(); verify(stringObserver, times(1)).onNext("one"); verify(stringObserver, times(1)).onNext("two"); verify(stringObserver, times(1)).onNext("three"); verify(stringObserver, times(1)).onNext("four"); verify(stringObserver, times(0)).onNext("five"); verify(stringObserver, times(0)).onNext("six"); }
@Test public void testErrorDelayed2() { final Observable<String> o1 = Observable.create(new TestErrorObservable("one", "two", "three")); final Observable<String> o2 = Observable.create(new TestErrorObservable("four", null, "six")); final Observable<String> o3 = Observable.create(new TestErrorObservable("seven", "eight", null)); final Observable<String> o4 = Observable.create(new TestErrorObservable("nine")); @SuppressWarnings("unchecked") Observable<String> m = Observable.create(mergeDelayError(o1, o2, o3, o4)); m.subscribe(stringObserver); verify(stringObserver, times(1)).onError(any(NullPointerException.class)); verify(stringObserver, never()).onCompleted(); verify(stringObserver, times(1)).onNext("one"); verify(stringObserver, times(1)).onNext("two"); verify(stringObserver, times(1)).onNext("three"); verify(stringObserver, times(1)).onNext("four"); verify(stringObserver, times(0)).onNext("five"); verify(stringObserver, times(0)).onNext("six"); verify(stringObserver, times(1)).onNext("seven"); verify(stringObserver, times(1)).onNext("eight"); verify(stringObserver, times(1)).onNext("nine"); }
@Test public void testErrorDelayed3() { final Observable<String> o1 = Observable.create(new TestErrorObservable("one", "two", "three")); final Observable<String> o2 = Observable.create(new TestErrorObservable("four", "five", "six")); final Observable<String> o3 = Observable.create(new TestErrorObservable("seven", "eight", null)); final Observable<String> o4 = Observable.create(new TestErrorObservable("nine")); @SuppressWarnings("unchecked") Observable<String> m = Observable.create(mergeDelayError(o1, o2, o3, o4)); m.subscribe(stringObserver); verify(stringObserver, times(1)).onError(any(NullPointerException.class)); verify(stringObserver, never()).onCompleted(); verify(stringObserver, times(1)).onNext("one"); verify(stringObserver, times(1)).onNext("two"); verify(stringObserver, times(1)).onNext("three"); verify(stringObserver, times(1)).onNext("four"); verify(stringObserver, times(1)).onNext("five"); verify(stringObserver, times(1)).onNext("six"); verify(stringObserver, times(1)).onNext("seven"); verify(stringObserver, times(1)).onNext("eight"); verify(stringObserver, times(1)).onNext("nine"); }
@Test public void testErrorDelayed4() { final Observable<String> o1 = Observable.create(new TestErrorObservable("one", "two", "three")); final Observable<String> o2 = Observable.create(new TestErrorObservable("four", "five", "six")); final Observable<String> o3 = Observable.create(new TestErrorObservable("seven", "eight")); final Observable<String> o4 = Observable.create(new TestErrorObservable("nine", null)); @SuppressWarnings("unchecked") Observable<String> m = Observable.create(mergeDelayError(o1, o2, o3, o4)); m.subscribe(stringObserver); verify(stringObserver, times(1)).onError(any(NullPointerException.class)); verify(stringObserver, never()).onCompleted(); verify(stringObserver, times(1)).onNext("one"); verify(stringObserver, times(1)).onNext("two"); verify(stringObserver, times(1)).onNext("three"); verify(stringObserver, times(1)).onNext("four"); verify(stringObserver, times(1)).onNext("five"); verify(stringObserver, times(1)).onNext("six"); verify(stringObserver, times(1)).onNext("seven"); verify(stringObserver, times(1)).onNext("eight"); verify(stringObserver, times(1)).onNext("nine"); }
@Test public void testErrorDelayed4WithThreading() { final TestAsyncErrorObservable o1 = new TestAsyncErrorObservable("one", "two", "three"); final TestAsyncErrorObservable o2 = new TestAsyncErrorObservable("four", "five", "six"); final TestAsyncErrorObservable o3 = new TestAsyncErrorObservable("seven", "eight"); final TestAsyncErrorObservable o4 = new TestAsyncErrorObservable("nine", null); @SuppressWarnings("unchecked") Observable<String> m = Observable.create(mergeDelayError(Observable.create(o1), Observable.create(o2), Observable.create(o3), Observable.create(o4))); m.subscribe(stringObserver); try { o1.t.join(); o2.t.join(); o3.t.join(); o4.t.join(); } catch (InterruptedException e) { throw new RuntimeException(e); } verify(stringObserver, times(1)).onError(any(NullPointerException.class)); verify(stringObserver, never()).onCompleted(); verify(stringObserver, times(1)).onNext("one"); verify(stringObserver, times(1)).onNext("two"); verify(stringObserver, times(1)).onNext("three"); verify(stringObserver, times(1)).onNext("four"); verify(stringObserver, times(1)).onNext("five"); verify(stringObserver, times(1)).onNext("six"); verify(stringObserver, times(1)).onNext("seven"); verify(stringObserver, times(1)).onNext("eight"); verify(stringObserver, times(1)).onNext("nine"); }
@Test public void testCompositeErrorDelayed1() { final Observable<String> o1 = Observable.create(new TestErrorObservable("four", null, "six")); final Observable<String> o2 = Observable.create(new TestErrorObservable("one", "two", null)); @SuppressWarnings("unchecked") Observable<String> m = Observable.create(mergeDelayError(o1, o2)); m.subscribe(stringObserver); verify(stringObserver, times(1)).onError(any(CompositeException.class)); verify(stringObserver, never()).onCompleted(); verify(stringObserver, times(1)).onNext("one"); verify(stringObserver, times(1)).onNext("two"); verify(stringObserver, times(0)).onNext("three"); verify(stringObserver, times(1)).onNext("four"); verify(stringObserver, times(0)).onNext("five"); verify(stringObserver, times(0)).onNext("six"); }
@Test public void testCompositeErrorDelayed2() { final Observable<String> o1 = Observable.create(new TestErrorObservable("four", null, "six")); final Observable<String> o2 = Observable.create(new TestErrorObservable("one", "two", null)); @SuppressWarnings("unchecked") Observable<String> m = Observable.create(mergeDelayError(o1, o2)); CaptureObserver w = new CaptureObserver(); m.subscribe(w); assertNotNull(w.e); if (w.e instanceof CompositeException) { assertEquals(2, ((CompositeException) w.e).getExceptions().size()); w.e.printStackTrace(); } else { fail("Expecting CompositeException"); } }
@Test public void testMergeObservableOfObservables() { final Observable<String> o1 = Observable.create(new TestSynchronousObservable()); final Observable<String> o2 = Observable.create(new TestSynchronousObservable()); Observable<Observable<String>> observableOfObservables = Observable.create(new Observable.OnSubscribeFunc<Observable<String>>() { @Override public Subscription onSubscribe(Observer<? super Observable<String>> observer) { observer.onNext(o1); observer.onNext(o2); observer.onCompleted(); return Subscriptions.empty(); } }); Observable<String> m = Observable.create(mergeDelayError(observableOfObservables)); m.subscribe(stringObserver); verify(stringObserver, never()).onError(any(Throwable.class)); verify(stringObserver, times(1)).onCompleted(); verify(stringObserver, times(2)).onNext("hello"); }
@Test public void testMergeArray() { final Observable<String> o1 = Observable.create(new TestSynchronousObservable()); final Observable<String> o2 = Observable.create(new TestSynchronousObservable()); @SuppressWarnings("unchecked") Observable<String> m = Observable.create(mergeDelayError(o1, o2)); m.subscribe(stringObserver); verify(stringObserver, never()).onError(any(Throwable.class)); verify(stringObserver, times(2)).onNext("hello"); verify(stringObserver, times(1)).onCompleted(); }
@Test public void testMergeList() { final Observable<String> o1 = Observable.create(new TestSynchronousObservable()); final Observable<String> o2 = Observable.create(new TestSynchronousObservable()); List<Observable<String>> listOfObservables = new ArrayList<Observable<String>>(); listOfObservables.add(o1); listOfObservables.add(o2); Observable<String> m = Observable.create(mergeDelayError(listOfObservables)); m.subscribe(stringObserver); verify(stringObserver, never()).onError(any(Throwable.class)); verify(stringObserver, times(1)).onCompleted(); verify(stringObserver, times(2)).onNext("hello"); }
@Test public void testUnSubscribe() { TestObservable tA = new TestObservable(); TestObservable tB = new TestObservable(); @SuppressWarnings("unchecked") Observable<String> m = Observable.create(mergeDelayError(Observable.create(tA), Observable.create(tB))); Subscription s = m.subscribe(stringObserver); tA.sendOnNext("Aone"); tB.sendOnNext("Bone"); s.unsubscribe(); tA.sendOnNext("Atwo"); tB.sendOnNext("Btwo"); tA.sendOnCompleted(); tB.sendOnCompleted(); verify(stringObserver, never()).onError(any(Throwable.class)); verify(stringObserver, times(1)).onNext("Aone"); verify(stringObserver, times(1)).onNext("Bone"); assertTrue(tA.unsubscribed); assertTrue(tB.unsubscribed); verify(stringObserver, never()).onNext("Atwo"); verify(stringObserver, never()).onNext("Btwo"); verify(stringObserver, never()).onCompleted(); }
@Test public void testMergeArrayWithThreading() { final TestASynchronousObservable o1 = new TestASynchronousObservable(); final TestASynchronousObservable o2 = new TestASynchronousObservable(); @SuppressWarnings("unchecked") Observable<String> m = Observable.create(mergeDelayError(Observable.create(o1), Observable.create(o2))); m.subscribe(stringObserver); try { o1.t.join(); o2.t.join(); } catch (InterruptedException e) { throw new RuntimeException(e); } verify(stringObserver, never()).onError(any(Throwable.class)); verify(stringObserver, times(2)).onNext("hello"); verify(stringObserver, times(1)).onCompleted(); }