@Test public void testTimeoutSelectorFirstThrows() { Observable<Integer> source = Observable.<Integer>never(); final PublishSubject<Integer> timeout = PublishSubject.create(); Func1<Integer, Observable<Integer>> timeoutFunc = new Func1<Integer, Observable<Integer>>() { @Override public Observable<Integer> call(Integer t1) { return timeout; } }; Func0<Observable<Integer>> firstTimeoutFunc = new Func0<Observable<Integer>>() { @Override public Observable<Integer> call() { throw new OperationReduceTest.CustomException(); } }; Observable<Integer> other = Observable.from(Arrays.asList(100)); @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); source.timeout(firstTimeoutFunc, timeoutFunc, other).subscribe(o); verify(o).onError(any(OperationReduceTest.CustomException.class)); verify(o, never()).onNext(any()); verify(o, never()).onCompleted(); }
public static <T> OnSubscribeFunc<T> timeout(Observable<? extends T> source, long timeout, TimeUnit timeUnit) { return new Timeout<T>(source, timeout, timeUnit, null, Schedulers.computation()); }
OperationTimeout { public static <T> OnSubscribeFunc<T> timeout(Observable<? extends T> source, long timeout, TimeUnit timeUnit) { return new Timeout<T>(source, timeout, timeUnit, null, Schedulers.computation()); } }
OperationTimeout { public static <T> OnSubscribeFunc<T> timeout(Observable<? extends T> source, long timeout, TimeUnit timeUnit) { return new Timeout<T>(source, timeout, timeUnit, null, Schedulers.computation()); }  }
OperationTimeout { public static <T> OnSubscribeFunc<T> timeout(Observable<? extends T> source, long timeout, TimeUnit timeUnit) { return new Timeout<T>(source, timeout, timeUnit, null, Schedulers.computation()); }  static OnSubscribeFunc<T> timeout(Observable<? extends T> source, long timeout, TimeUnit timeUnit); static OnSubscribeFunc<T> timeout(Observable<? extends T> sequence, long timeout, TimeUnit timeUnit, Observable<? extends T> other); static OnSubscribeFunc<T> timeout(Observable<? extends T> source, long timeout, TimeUnit timeUnit, Scheduler scheduler); static OnSubscribeFunc<T> timeout(Observable<? extends T> sequence, long timeout, TimeUnit timeUnit, Observable<? extends T> other, Scheduler scheduler); static OnSubscribeFunc<T> timeoutSelector(Observable<? extends T> source, Func0<? extends Observable<U>> firstValueTimeout, Func1<? super T, ? extends Observable<V>> valueTimeout, Observable<? extends T> other); }
OperationTimeout { public static <T> OnSubscribeFunc<T> timeout(Observable<? extends T> source, long timeout, TimeUnit timeUnit) { return new Timeout<T>(source, timeout, timeUnit, null, Schedulers.computation()); }  static OnSubscribeFunc<T> timeout(Observable<? extends T> source, long timeout, TimeUnit timeUnit); static OnSubscribeFunc<T> timeout(Observable<? extends T> sequence, long timeout, TimeUnit timeUnit, Observable<? extends T> other); static OnSubscribeFunc<T> timeout(Observable<? extends T> source, long timeout, TimeUnit timeUnit, Scheduler scheduler); static OnSubscribeFunc<T> timeout(Observable<? extends T> sequence, long timeout, TimeUnit timeUnit, Observable<? extends T> other, Scheduler scheduler); static OnSubscribeFunc<T> timeoutSelector(Observable<? extends T> source, Func0<? extends Observable<U>> firstValueTimeout, Func1<? super T, ? extends Observable<V>> valueTimeout, Observable<? extends T> other);  }