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