Merge pull request #851 from zsxwing/timeout

Reimplement the timeout operator and fix timeout bugs

Author: benjchristensen

rxjava-core/src/main/java/rx/Observable.java

@@ -87,7 +87,6 @@
 import rx.operators.OperationTakeWhile;
 import rx.operators.OperationThrottleFirst;
 import rx.operators.OperationTimeInterval;
-import rx.operators.OperationTimeout;
 import rx.operators.OperationTimer;
 import rx.operators.OperationToMap;
 import rx.operators.OperationToMultimap;
@@ -106,6 +105,8 @@
 import rx.operators.OperatorRepeat;
 import rx.operators.OperatorSubscribeOn;
 import rx.operators.OperatorTake;
+import rx.operators.OperatorTimeout;
+import rx.operators.OperatorTimeoutWithSelector;
 import rx.operators.OperatorTimestamp;
 import rx.operators.OperatorToObservableList;
 import rx.operators.OperatorToObservableSortedList;
@@ -7755,11 +7756,8 @@ public final Observable<TimeInterval<T>> timeInterval(Scheduler scheduler) {
      * @return an Observable that completes if either the first item or any subsequent item doesn't
      *         arrive within the time windows specified by the timeout selectors
      */
-    public final <U, V> Observable<T> timeout(Func0<? extends Observable<U>> firstTimeoutSelector, Func1<? super T, ? extends Observable<U>> timeoutSelector) {
-        if (firstTimeoutSelector == null) {
-            throw new NullPointerException("firstTimeoutSelector");
-        }
-        return timeout(firstTimeoutSelector, timeoutSelector, Observable.<T> empty());
+    public final <U, V> Observable<T> timeout(Func0<? extends Observable<U>> firstTimeoutSelector, Func1<? super T, ? extends Observable<V>> timeoutSelector) {
+        return timeout(firstTimeoutSelector, timeoutSelector, null);
     }
 
     /**
@@ -7785,14 +7783,11 @@ public final <U, V> Observable<T> timeout(Func0<? extends Observable<U>> firstTi
      * @return an Observable that mirrors the source Observable, but switches to the {@code other} Observable if either the first item emitted by the source Observable or any
      *         subsequent item don't arrive within time windows defined by the timeout selectors
      */
-    public final <U, V> Observable<T> timeout(Func0<? extends Observable<U>> firstTimeoutSelector, Func1<? super T, ? extends Observable<U>> timeoutSelector, Observable<? extends T> other) {
-        if (firstTimeoutSelector == null) {
-            throw new NullPointerException("firstTimeoutSelector");
-        }
-        if (other == null) {
-            throw new NullPointerException("other");
+    public final <U, V> Observable<T> timeout(Func0<? extends Observable<U>> firstTimeoutSelector, Func1<? super T, ? extends Observable<V>> timeoutSelector, Observable<? extends T> other) {
+        if(timeoutSelector == null) {
+            throw new NullPointerException("timeoutSelector is null");
         }
-        return create(OperationTimeout.timeoutSelector(this, firstTimeoutSelector, timeoutSelector, other));
+        return lift(new OperatorTimeoutWithSelector<T, U, V>(firstTimeoutSelector, timeoutSelector, other));
     }
 
     /**
@@ -7814,8 +7809,8 @@ public final <U, V> Observable<T> timeout(Func0<? extends Observable<U>> firstTi
      *         the source Observable takes longer to arrive than the time window defined by the
      *         selector for the previously emitted item
      */
-    public final <U> Observable<T> timeout(Func1<? super T, ? extends Observable<U>> timeoutSelector) {
-        return timeout(timeoutSelector, Observable.<T> empty());
+    public final <V> Observable<T> timeout(Func1<? super T, ? extends Observable<V>> timeoutSelector) {
+        return timeout(null, timeoutSelector, null);
     }
 
     /**
@@ -7839,11 +7834,8 @@ public final <U> Observable<T> timeout(Func1<? super T, ? extends Observable<U>>
      *         fallback Observable if a item emitted by the source Observable takes longer to arrive
      *         than the time window defined by the selector for the previously emitted item
      */
-    public final <U> Observable<T> timeout(Func1<? super T, ? extends Observable<U>> timeoutSelector, Observable<? extends T> other) {
-        if (other == null) {
-            throw new NullPointerException("other");
-        }
-        return create(OperationTimeout.timeoutSelector(this, null, timeoutSelector, other));
+    public final <V> Observable<T> timeout(Func1<? super T, ? extends Observable<V>> timeoutSelector, Observable<? extends T> other) {
+        return timeout(null, timeoutSelector, other);
     }
 
     /**
@@ -7864,7 +7856,7 @@ public final <U> Observable<T> timeout(Func1<? super T, ? extends Observable<U>>
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh244283.aspx">MSDN: Observable.Timeout</a>
      */
     public final Observable<T> timeout(long timeout, TimeUnit timeUnit) {
-        return create(OperationTimeout.timeout(this, timeout, timeUnit));
+        return timeout(timeout, timeUnit, null, Schedulers.computation());
     }
 
     /**
@@ -7887,7 +7879,7 @@ public final Observable<T> timeout(long timeout, TimeUnit timeUnit) {
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229512.aspx">MSDN: Observable.Timeout</a>
      */
     public final Observable<T> timeout(long timeout, TimeUnit timeUnit, Observable<? extends T> other) {
-        return create(OperationTimeout.timeout(this, timeout, timeUnit, other));
+        return timeout(timeout, timeUnit, other, Schedulers.computation());
     }
 
     /**
@@ -7912,7 +7904,7 @@ public final Observable<T> timeout(long timeout, TimeUnit timeUnit, Observable<?
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh211676.aspx">MSDN: Observable.Timeout</a>
      */
     public final Observable<T> timeout(long timeout, TimeUnit timeUnit, Observable<? extends T> other, Scheduler scheduler) {
-        return create(OperationTimeout.timeout(this, timeout, timeUnit, other, scheduler));
+        return lift(new OperatorTimeout<T>(timeout, timeUnit, other, scheduler));
     }
 
     /**
@@ -7935,7 +7927,7 @@ public final Observable<T> timeout(long timeout, TimeUnit timeUnit, Observable<?
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh228946.aspx">MSDN: Observable.Timeout</a>
      */
     public final Observable<T> timeout(long timeout, TimeUnit timeUnit, Scheduler scheduler) {
-        return create(OperationTimeout.timeout(this, timeout, timeUnit, scheduler));
+        return timeout(timeout, timeUnit, null, scheduler);
     }
 
     /**