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);  }
@Test public void testConcat() { @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); final String[] o = { "1", "3", "5", "7" }; final String[] e = { "2", "4", "6" }; final Observable<String> odds = Observable.from(o); final Observable<String> even = Observable.from(e); @SuppressWarnings("unchecked") Observable<String> concat = Observable.create(concat(odds, even)); concat.subscribe(observer); verify(observer, times(7)).onNext(anyString()); }
@Test public void testConcatWithList() { @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); final String[] o = { "1", "3", "5", "7" }; final String[] e = { "2", "4", "6" }; final Observable<String> odds = Observable.from(o); final Observable<String> even = Observable.from(e); final List<Observable<String>> list = new ArrayList<Observable<String>>(); list.add(odds); list.add(even); Observable<String> concat = Observable.create(concat(list)); concat.subscribe(observer); verify(observer, times(7)).onNext(anyString()); }
@Test public void testConcatObservableOfObservables() { @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); final String[] o = { "1", "3", "5", "7" }; final String[] e = { "2", "4", "6" }; final Observable<String> odds = Observable.from(o); final Observable<String> even = Observable.from(e); Observable<Observable<String>> observableOfObservables = Observable.create(new Observable.OnSubscribeFunc<Observable<String>>() { @Override public Subscription onSubscribe(Observer<? super Observable<String>> observer) { observer.onNext(odds); observer.onNext(even); observer.onCompleted(); return Subscriptions.empty(); } }); Observable<String> concat = Observable.create(concat(observableOfObservables)); concat.subscribe(observer); verify(observer, times(7)).onNext(anyString()); }
@SuppressWarnings("unchecked") @Test public void testSimpleAsyncConcat() { Observer<String> observer = mock(Observer.class); TestObservable<String> o1 = new TestObservable<String>("one", "two", "three"); TestObservable<String> o2 = new TestObservable<String>("four", "five", "six"); Observable.concat(Observable.create(o1), Observable.create(o2)).subscribe(observer); try { o1.t.join(); o2.t.join(); } catch (Throwable e) { throw new RuntimeException("failed waiting on threads"); } 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"); }
@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)); }
@SuppressWarnings("unchecked") @Test public void testBlockedObservableOfObservables() { Observer<String> observer = mock(Observer.class); final String[] o = { "1", "3", "5", "7" }; final String[] e = { "2", "4", "6" }; final Observable<String> odds = Observable.from(o); final Observable<String> even = Observable.from(e); final CountDownLatch callOnce = new CountDownLatch(1); final CountDownLatch okToContinue = new CountDownLatch(1); TestObservable<Observable<String>> observableOfObservables = new TestObservable<Observable<String>>(callOnce, okToContinue, odds, even); Observable.OnSubscribeFunc<String> concatF = concat(Observable.create(observableOfObservables)); Observable<String> concat = Observable.create(concatF); concat.subscribe(observer); try { callOnce.await(); } catch (Throwable ex) { ex.printStackTrace(); fail(ex.getMessage()); } verify(observer, times(1)).onNext("1"); verify(observer, times(1)).onNext("3"); verify(observer, times(1)).onNext("5"); verify(observer, times(1)).onNext("7"); try { okToContinue.countDown(); observableOfObservables.t.join(); } catch (Throwable ex) { ex.printStackTrace(); fail(ex.getMessage()); } verify(observer, times(1)).onNext("2"); verify(observer, times(1)).onNext("4"); verify(observer, times(1)).onNext("6"); }
@Test public void testConcatConcurrentWithInfinity() { final TestObservable<String> w1 = new TestObservable<String>("one", "two", "three"); final TestObservable<String> w2 = new TestObservable<String>("hello", Integer.MAX_VALUE); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); @SuppressWarnings("unchecked") TestObservable<Observable<String>> observableOfObservables = new TestObservable<Observable<String>>(Observable.create(w1), Observable.create(w2)); Observable.OnSubscribeFunc<String> concatF = concat(Observable.create(observableOfObservables)); Observable<String> concat = Observable.create(concatF); concat.take(50).subscribe(observer); try { w1.waitForThreadDone(); w2.waitForThreadDone(); } catch (InterruptedException e) { e.printStackTrace(); } 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(47)).onNext("hello"); verify(observer, times(1)).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); }
@Test public void testConcatNonBlockingObservables() { final CountDownLatch okToContinueW1 = new CountDownLatch(1); final CountDownLatch okToContinueW2 = new CountDownLatch(1); final TestObservable<String> w1 = new TestObservable<String>(null, okToContinueW1, "one", "two", "three"); final TestObservable<String> w2 = new TestObservable<String>(null, okToContinueW2, "four", "five", "six"); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); Observable<Observable<String>> observableOfObservables = Observable.create(new Observable.OnSubscribeFunc<Observable<String>>() { @Override public Subscription onSubscribe(Observer<? super Observable<String>> observer) { observer.onNext(Observable.create(w1)); observer.onNext(Observable.create(w2)); observer.onCompleted(); return Subscriptions.empty(); } }); Observable<String> concat = Observable.create(concat(observableOfObservables)); concat.subscribe(observer); verify(observer, times(0)).onCompleted(); try { okToContinueW1.countDown(); okToContinueW2.countDown(); w1.t.join(); w2.t.join(); } catch (InterruptedException e) { e.printStackTrace(); } 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"); verify(observer, times(1)).onCompleted(); }
@Test public void testConcatUnsubscribe() { final CountDownLatch callOnce = new CountDownLatch(1); final CountDownLatch okToContinue = new CountDownLatch(1); final TestObservable<String> w1 = new TestObservable<String>("one", "two", "three"); final TestObservable<String> w2 = new TestObservable<String>(callOnce, okToContinue, "four", "five", "six"); @SuppressWarnings("unchecked") final Observer<String> observer = mock(Observer.class); @SuppressWarnings("unchecked") final Observable<String> concat = Observable.create(concat(Observable.create(w1), Observable.create(w2))); try { Subscription s1 = concat.subscribe(observer); callOnce.await(); s1.unsubscribe(); okToContinue.countDown(); w1.t.join(); w2.t.join(); } catch (Throwable e) { e.printStackTrace(); fail(e.getMessage()); } 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, never()).onNext("five"); inOrder.verify(observer, never()).onNext("six"); inOrder.verify(observer, never()).onCompleted(); }
@Test public void testConcatUnsubscribeConcurrent() { final CountDownLatch callOnce = new CountDownLatch(1); final CountDownLatch okToContinue = new CountDownLatch(1); final TestObservable<String> w1 = new TestObservable<String>("one", "two", "three"); final TestObservable<String> w2 = new TestObservable<String>(callOnce, okToContinue, "four", "five", "six"); @SuppressWarnings("unchecked") Observer<String> observer = mock(Observer.class); @SuppressWarnings("unchecked") TestObservable<Observable<String>> observableOfObservables = new TestObservable<Observable<String>>(Observable.create(w1), Observable.create(w2)); Observable.OnSubscribeFunc<String> concatF = concat(Observable.create(observableOfObservables)); Observable<String> concat = Observable.create(concatF); Subscription s1 = concat.subscribe(observer); try { callOnce.await(); s1.unsubscribe(); okToContinue.countDown(); w1.t.join(); w2.t.join(); } catch (Throwable e) { e.printStackTrace(); fail(e.getMessage()); } 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, never()).onNext("five"); inOrder.verify(observer, never()).onNext("six"); verify(observer, never()).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); }
@Test public void testMultipleObservers() { Observer<Object> o1 = mock(Observer.class); Observer<Object> o2 = mock(Observer.class); TestScheduler s = new TestScheduler(); Observable<Long> timer = Observable.interval(500, TimeUnit.MILLISECONDS, s).take(2); Observable<Long> o = Observable.concat(timer, timer); o.subscribe(o1); o.subscribe(o2); InOrder inOrder1 = inOrder(o1); InOrder inOrder2 = inOrder(o2); s.advanceTimeBy(500, TimeUnit.MILLISECONDS); inOrder1.verify(o1, times(1)).onNext(0L); inOrder2.verify(o2, times(1)).onNext(0L); s.advanceTimeBy(500, TimeUnit.MILLISECONDS); inOrder1.verify(o1, times(1)).onNext(1L); inOrder2.verify(o2, times(1)).onNext(1L); s.advanceTimeBy(500, TimeUnit.MILLISECONDS); inOrder1.verify(o1, times(1)).onNext(0L); inOrder2.verify(o2, times(1)).onNext(0L); s.advanceTimeBy(500, TimeUnit.MILLISECONDS); inOrder1.verify(o1, times(1)).onNext(1L); inOrder2.verify(o2, times(1)).onNext(1L); inOrder1.verify(o1, times(1)).onCompleted(); inOrder2.verify(o2, times(1)).onCompleted(); verify(o1, never()).onError(any(Throwable.class)); verify(o2, never()).onError(any(Throwable.class)); }