@SuppressWarnings("unchecked") @Test public void testNestedAsyncConcat() throws Throwable { Observer<String> observer = mock(Observer.class); final TestObservable<String> o1 = new TestObservable<String>("one", "two", "three"); final TestObservable<String> o2 = new TestObservable<String>("four", "five", "six"); final TestObservable<String> o3 = new TestObservable<String>("seven", "eight", "nine"); final CountDownLatch allowThird = new CountDownLatch(1); final AtomicReference<Thread> parent = new AtomicReference<Thread>(); final CountDownLatch parentHasStarted = new CountDownLatch(1); Observable<Observable<String>> observableOfObservables = Observable.create(new Observable.OnSubscribeFunc<Observable<String>>() { @Override public Subscription onSubscribe(final Observer<? super Observable<String>> observer) { final BooleanSubscription s = new BooleanSubscription(); parent.set(new Thread(new Runnable() { @Override public void run() { try { if (!s.isUnsubscribed()) { System.out.println("Emit o1"); observer.onNext(Observable.create(o1)); } if (!s.isUnsubscribed()) { System.out.println("Emit o2"); observer.onNext(Observable.create(o2)); } try { allowThird.await(); } catch (InterruptedException e) { observer.onError(e); } if (!s.isUnsubscribed()) { System.out.println("Emit o3"); observer.onNext(Observable.create(o3)); } } catch (Throwable e) { observer.onError(e); } finally { System.out.println("Done parent Observable"); observer.onCompleted(); } } })); parent.get().start(); parentHasStarted.countDown(); return s; } }); Observable.create(concat(observableOfObservables)).subscribe(observer); parentHasStarted.await(); try { System.out.println("Thread1 is starting ... waiting for it to complete ..."); o1.waitForThreadDone(); System.out.println("Thread2 is starting ... waiting for it to complete ..."); o2.waitForThreadDone(); } catch (Throwable e) { throw new RuntimeException("failed waiting on threads", e); } InOrder inOrder = inOrder(observer); inOrder.verify(observer, times(1)).onNext("one"); inOrder.verify(observer, times(1)).onNext("two"); inOrder.verify(observer, times(1)).onNext("three"); inOrder.verify(observer, times(1)).onNext("four"); inOrder.verify(observer, times(1)).onNext("five"); inOrder.verify(observer, times(1)).onNext("six"); inOrder.verify(observer, never()).onNext("seven"); inOrder.verify(observer, never()).onNext("eight"); inOrder.verify(observer, never()).onNext("nine"); verify(observer, never()).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); allowThird.countDown(); try { o3.waitForThreadDone(); } catch (Throwable e) { throw new RuntimeException("failed waiting on threads", e); } inOrder.verify(observer, times(1)).onNext("seven"); inOrder.verify(observer, times(1)).onNext("eight"); inOrder.verify(observer, times(1)).onNext("nine"); inOrder.verify(observer, times(1)).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); }
public static <T> OnSubscribeFunc<T> concat(final Observable<? extends T>... sequences) { return concat(Observable.from(sequences)); }
OperationConcat { public static <T> OnSubscribeFunc<T> concat(final Observable<? extends T>... sequences) { return concat(Observable.from(sequences)); } }
OperationConcat { public static <T> OnSubscribeFunc<T> concat(final Observable<? extends T>... sequences) { return concat(Observable.from(sequences)); }  }
OperationConcat { public static <T> OnSubscribeFunc<T> concat(final Observable<? extends T>... sequences) { return concat(Observable.from(sequences)); }  static OnSubscribeFunc<T> concat(final Observable<? extends T>... sequences); static OnSubscribeFunc<T> concat(final Iterable<? extends Observable<? extends T>> sequences); static OnSubscribeFunc<T> concat(final Observable<? extends Observable<? extends T>> sequences); }
OperationConcat { public static <T> OnSubscribeFunc<T> concat(final Observable<? extends T>... sequences) { return concat(Observable.from(sequences)); }  static OnSubscribeFunc<T> concat(final Observable<? extends T>... sequences); static OnSubscribeFunc<T> concat(final Iterable<? extends Observable<? extends T>> sequences); static OnSubscribeFunc<T> concat(final Observable<? extends Observable<? extends T>> sequences);  }