@Test public void testSwitchWithError() { Observable<Observable<String>> source = Observable.create(new Observable.OnSubscribeFunc<Observable<String>>() { @Override public Subscription onSubscribe(Observer<? super Observable<String>> observer) { publishNext(observer, 50, Observable.create(new Observable.OnSubscribeFunc<String>() { @Override public Subscription onSubscribe(Observer<? super String> observer) { publishNext(observer, 50, "one"); publishNext(observer, 100, "two"); return Subscriptions.empty(); } })); publishNext(observer, 200, Observable.create(new Observable.OnSubscribeFunc<String>() { @Override public Subscription onSubscribe(Observer<? super String> observer) { publishNext(observer, 0, "three"); publishNext(observer, 100, "four"); return Subscriptions.empty(); } })); publishError(observer, 250, new TestException()); return Subscriptions.empty(); } }); Observable<String> sampled = Observable.create(OperationSwitch.switchDo(source)); sampled.subscribe(observer); InOrder inOrder = inOrder(observer); scheduler.advanceTimeTo(90, TimeUnit.MILLISECONDS); inOrder.verify(observer, never()).onNext(anyString()); verify(observer, never()).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); scheduler.advanceTimeTo(125, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onNext("one"); verify(observer, never()).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); scheduler.advanceTimeTo(175, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onNext("two"); verify(observer, never()).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); scheduler.advanceTimeTo(225, TimeUnit.MILLISECONDS); inOrder.verify(observer, times(1)).onNext("three"); verify(observer, never()).onCompleted(); verify(observer, never()).onError(any(Throwable.class)); scheduler.advanceTimeTo(350, TimeUnit.MILLISECONDS); inOrder.verify(observer, never()).onNext(anyString()); verify(observer, never()).onCompleted(); verify(observer, times(1)).onError(any(TestException.class)); }
public static <T> OnSubscribeFunc<T> switchDo(final Observable<? extends Observable<? extends T>> sequences) { return new OnSubscribeFunc<T>() { @Override public Subscription onSubscribe(Observer<? super T> observer) { return new Switch<T>(sequences).onSubscribe(observer); } }; }
OperationSwitch { public static <T> OnSubscribeFunc<T> switchDo(final Observable<? extends Observable<? extends T>> sequences) { return new OnSubscribeFunc<T>() { @Override public Subscription onSubscribe(Observer<? super T> observer) { return new Switch<T>(sequences).onSubscribe(observer); } }; } }
OperationSwitch { public static <T> OnSubscribeFunc<T> switchDo(final Observable<? extends Observable<? extends T>> sequences) { return new OnSubscribeFunc<T>() { @Override public Subscription onSubscribe(Observer<? super T> observer) { return new Switch<T>(sequences).onSubscribe(observer); } }; }  }
OperationSwitch { public static <T> OnSubscribeFunc<T> switchDo(final Observable<? extends Observable<? extends T>> sequences) { return new OnSubscribeFunc<T>() { @Override public Subscription onSubscribe(Observer<? super T> observer) { return new Switch<T>(sequences).onSubscribe(observer); } }; }  static OnSubscribeFunc<T> switchDo(final Observable<? extends Observable<? extends T>> sequences); }
OperationSwitch { public static <T> OnSubscribeFunc<T> switchDo(final Observable<? extends Observable<? extends T>> sequences) { return new OnSubscribeFunc<T>() { @Override public Subscription onSubscribe(Observer<? super T> observer) { return new Switch<T>(sequences).onSubscribe(observer); } }; }  static OnSubscribeFunc<T> switchDo(final Observable<? extends Observable<? extends T>> sequences);  }