@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"); }
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); } }; }
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); } }; } }
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); } }; }  }
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); }
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);  }