@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 {
                        // emit first
                        if (!s.isUnsubscribed()) {
                            System.out.println("Emit o1");
                            observer.onNext(Observable.create(o1));
                        }
                        // emit second
                        if (!s.isUnsubscribed()) {
                            System.out.println("Emit o2");
                            observer.onNext(Observable.create(o2));
                        }
                        // wait until sometime later and emit third
                        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);
    // wait for parent to start
    parentHasStarted.await();
    try {
        // wait for first 2 async observables to complete
        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");
    // we shouldn't have the following 3 yet
    inOrder.verify(observer, never()).onNext("seven");
    inOrder.verify(observer, never()).onNext("eight");
    inOrder.verify(observer, never()).onNext("nine");
    // we should not be completed yet
    verify(observer, never()).onCompleted();
    verify(observer, never()).onError(any(Throwable.class));
    // now allow the third
    allowThird.countDown();
    try {
        // wait for 3rd to complete
        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));
}