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);  }
@Test(timeout = 2000) public void testTimeoutSelectorNormal1() { PublishSubject<Integer> source = PublishSubject.create(); 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() { return timeout; } }; Observable<Integer> other = Observable.from(Arrays.asList(100)); @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); InOrder inOrder = inOrder(o); source.timeout(firstTimeoutFunc, timeoutFunc, other).subscribe(o); source.onNext(1); source.onNext(2); source.onNext(3); timeout.onNext(1); inOrder.verify(o).onNext(1); inOrder.verify(o).onNext(2); inOrder.verify(o).onNext(3); inOrder.verify(o).onNext(100); inOrder.verify(o).onCompleted(); verify(o, never()).onError(any(Throwable.class)); }
@Test public void testTimeoutSelectorTimeoutFirst() throws InterruptedException { 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() { return timeout; } }; Observable<Integer> other = Observable.from(Arrays.asList(100)); @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); InOrder inOrder = inOrder(o); source.timeout(firstTimeoutFunc, timeoutFunc, other).subscribe(o); timeout.onNext(1); inOrder.verify(o).onNext(100); inOrder.verify(o).onCompleted(); verify(o, never()).onError(any(Throwable.class)); }
@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(); }
@Test public void testTimeoutSelectorSubsequentThrows() { PublishSubject<Integer> source = PublishSubject.create(); final PublishSubject<Integer> timeout = PublishSubject.create(); Func1<Integer, Observable<Integer>> timeoutFunc = new Func1<Integer, Observable<Integer>>() { @Override public Observable<Integer> call(Integer t1) { throw new OperationReduceTest.CustomException(); } }; Func0<Observable<Integer>> firstTimeoutFunc = new Func0<Observable<Integer>>() { @Override public Observable<Integer> call() { return timeout; } }; Observable<Integer> other = Observable.from(Arrays.asList(100)); @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); InOrder inOrder = inOrder(o); source.timeout(firstTimeoutFunc, timeoutFunc, other).subscribe(o); source.onNext(1); inOrder.verify(o).onNext(1); inOrder.verify(o).onError(any(OperationReduceTest.CustomException.class)); verify(o, never()).onCompleted(); }
@Test public void testTimeoutSelectorFirstObservableThrows() { PublishSubject<Integer> source = PublishSubject.create(); 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() { return Observable.<Integer> error(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(); }
@Test public void testTimeoutSelectorSubsequentObservableThrows() { PublishSubject<Integer> source = PublishSubject.create(); final PublishSubject<Integer> timeout = PublishSubject.create(); Func1<Integer, Observable<Integer>> timeoutFunc = new Func1<Integer, Observable<Integer>>() { @Override public Observable<Integer> call(Integer t1) { return Observable.<Integer> error(new OperationReduceTest.CustomException()); } }; Func0<Observable<Integer>> firstTimeoutFunc = new Func0<Observable<Integer>>() { @Override public Observable<Integer> call() { return timeout; } }; Observable<Integer> other = Observable.from(Arrays.asList(100)); @SuppressWarnings("unchecked") Observer<Object> o = mock(Observer.class); InOrder inOrder = inOrder(o); source.timeout(firstTimeoutFunc, timeoutFunc, other).subscribe(o); source.onNext(1); inOrder.verify(o).onNext(1); inOrder.verify(o).onError(any(OperationReduceTest.CustomException.class)); verify(o, never()).onCompleted(); }