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