Timeout with selector overloads

Author: akarnokd

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

@@ -6691,6 +6691,86 @@ public Observable<T> timeout(long timeout, TimeUnit timeUnit, Observable<? exten
         return create(OperationTimeout.timeout(this, timeout, timeUnit, other, scheduler));
     }
 
+    /**
+     * Create an observable which completes if a source item doesn't arrive after the 
+     * previous one in the time window specified by the per-item observable.
+     * <p>
+     * The arrival of the first source item is not timed out.
+     * @param <U> the timeout value type (ignored)
+     * @param timeoutSelector function that returns an observable for each source item
+     * which determines the timeout window for the subsequent source item
+     * @return an observable which completes if a source item doesn't arrive after the 
+     * previous one in the time window specified by the per-item observable.
+     */
+    public <U> Observable<T> timeout(Func1<? super T, ? extends Observable<U>> timeoutSelector) {
+        return timeout(timeoutSelector, Observable.<T>empty());
+    }
+
+    /**
+     * Create an observable which switches to the other Observable if a source 
+     * item doesn't arrive after the 
+     * previous one in the time window specified by the per-item observable.
+     * <p>
+     * The arrival of the first source item is not timed out.
+     * @param <U> the timeout value type (ignored)
+     * @param timeoutSelector function that returns an observable for each source item
+     * which determines the timeout window for the subsequent source item
+     * @param other the other observable to switch to if the source times out
+     * @return an observable which switches to the other Observable if a source 
+     * item doesn't arrive after the 
+     * previous one in the time window specified by the per-item observable
+     */
+    public <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));
+    }
+
+    /**
+     * Create an Observable which completes if either the first item or any subsequent item
+     * doesn't arrive within the time window specified by the timeout selectors' Observable.
+     * @param <U> the first timeout value type (ignored)
+     * @param <V> the subsequent timeout value type (ignored)
+     * @param firstTimeoutSelector function that returns an observable which determines
+     * the timeout window for the first source item
+     * @param timeoutSelector function that returns an observable for each source item
+     * which determines the timeout window for the subsequent source item
+     * @return an Observable which completes if either the first item or any subsequent item
+     * doesn't arrive within the time window specified by the timeout selectors' Observable.
+     */
+    public <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());
+    }
+
+    /**
+     * Create an Observable which switches to another Observable
+     * if either the first item or any subsequent item
+     * doesn't arrive within the time window specified by the timeout selectors' Observable.
+     * @param <U> the first timeout value type (ignored)
+     * @param <V> the subsequent timeout value type (ignored)
+     * @param firstTimeoutSelector function that returns an observable which determines
+     * the timeout window for the first source item
+     * @param timeoutSelector function that returns an observable for each source item
+     * which determines the timeout window for the subsequent source item
+     * @param other the other observable to switch to if the source times out
+     * @return an Observable which switches to another Observable
+     * if either the first item or any subsequent item
+     * doesn't arrive within the time window specified by the timeout selectors' Observable
+     */
+    public <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");
+        }
+        return create(OperationTimeout.timeoutSelector(this, firstTimeoutSelector, timeoutSelector, other));
+    }
+
     /**
      * Records the time interval between consecutive items emitted by an
      * Observable.

rxjava-core/src/main/java/rx/operators/OperationTimeout.java

@@ -28,8 +28,10 @@
 import rx.schedulers.Schedulers;
 import rx.subscriptions.CompositeSubscription;
 import rx.subscriptions.SerialSubscription;
+import rx.subscriptions.Subscriptions;
 import rx.util.functions.Action0;
 import rx.util.functions.Func0;
+import rx.util.functions.Func1;
 
 /**
  * Applies a timeout policy for each element in the observable sequence, using
@@ -154,4 +156,160 @@ public void onCompleted() {
             return composite;
         }
     }
+    
+    /** Timeout using a per-item observable sequence. */
+    public static <T, U, V> OnSubscribeFunc<T> timeoutSelector(Observable<? extends T> source, Func0<? extends Observable<U>> firstValueTimeout, Func1<? super T, ? extends Observable<V>> valueTimeout, Observable<? extends T> other) {
+        return new TimeoutSelector<T, U, V>(source, firstValueTimeout, valueTimeout, other);
+    }
+    
+    /** Timeout using a per-item observable sequence. */
+    private static final class TimeoutSelector<T, U, V> implements OnSubscribeFunc<T> {
+        final Observable<? extends T> source;
+        final Func0<? extends Observable<U>> firstValueTimeout;
+        final Func1<? super T, ? extends Observable<V>> valueTimeout;
+        final Observable<? extends T> other;
+
+        public TimeoutSelector(Observable<? extends T> source, Func0<? extends Observable<U>> firstValueTimeout, Func1<? super T, ? extends Observable<V>> valueTimeout, Observable<? extends T> other) {
+            this.source = source;
+            this.firstValueTimeout = firstValueTimeout;
+            this.valueTimeout = valueTimeout;
+            this.other = other;
+        }
+
+        @Override
+        public Subscription onSubscribe(Observer<? super T> t1) {
+            CompositeSubscription csub = new CompositeSubscription();
+
+            SourceObserver<T, V> so = new SourceObserver<T, V>(t1, valueTimeout, other, csub);
+            if (firstValueTimeout != null) {
+                Observable<U> o;
+                try {
+                    o = firstValueTimeout.call();
+                } catch (Throwable t) {
+                    t1.onError(t);
+                    return Subscriptions.empty();
+                }
+                
+                csub.add(o.subscribe(new TimeoutObserver<U>(so)));
+            }
+            csub.add(source.subscribe(so));
+            return csub;
+        }
+        
+        /** Observe the source. */
+        private static final class SourceObserver<T, V> implements Observer<T>, TimeoutCallback {
+            final Observer<? super T> observer;
+            final Func1<? super T, ? extends Observable<V>> valueTimeout;
+            final Observable<? extends T> other;
+            final CompositeSubscription cancel;
+            final Object guard;
+            boolean done;
+            final SerialSubscription tsub;
+            final TimeoutObserver<V> to;
+
+            public SourceObserver(Observer<? super T> observer, Func1<? super T, ? extends Observable<V>> valueTimeout, Observable<? extends T> other, CompositeSubscription cancel) {
+                this.observer = observer;
+                this.valueTimeout = valueTimeout;
+                this.other = other;
+                this.cancel = cancel;
+                this.guard = new Object();
+                this.tsub = new SerialSubscription();
+                this.cancel.add(tsub);
+                this.to = new TimeoutObserver<V>(this);
+            }
+
+            @Override
+            public void onNext(T args) {
+                tsub.set(Subscriptions.empty());
+                
+                synchronized (guard) {
+                    if (done) {
+                        return;
+                    }
+                    observer.onNext(args);
+                }
+                
+                Observable<V> o;
+                try {
+                    o = valueTimeout.call(args);
+                } catch (Throwable t) {
+                    onError(t);
+                    return;
+                }
+                
+                SerialSubscription osub = new SerialSubscription();
+                tsub.set(osub);
+                
+                osub.set(o.subscribe(to));
+            }
+            @Override
+            public void onError(Throwable e) {
+                synchronized (guard) {
+                    if (done) {
+                        return;
+                    }
+                    done = true;
+                    observer.onError(e);
+                }
+                cancel.unsubscribe();
+            }
+
+            @Override
+            public void onCompleted() {
+                synchronized (guard) {
+                    if (done) {
+                        return;
+                    }
+                    done = true;
+                    observer.onCompleted();
+                }
+                cancel.unsubscribe();
+            }
+            @Override
+            public void timeout() {
+                if (other != null) {
+                    synchronized (guard) {
+                        if (done) {
+                            return;
+                        }
+                        done = true;
+                    }
+                    cancel.clear();
+                    cancel.add(other.subscribe(observer));
+                } else {
+                    onCompleted();
+                }
+            }
+        }
+        
+        /** The timeout callback. */
+        private interface TimeoutCallback {
+            void timeout();
+            void onError(Throwable t);
+        }
+        
+        /** Observe the timeout. */
+        private static final class TimeoutObserver<V> implements Observer<V> {
+            final TimeoutCallback parent;
+
+            public TimeoutObserver(TimeoutCallback parent) {
+                this.parent = parent;
+            }
+            
+            @Override
+            public void onNext(V args) {
+                parent.timeout();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                parent.onError(e);
+            }
+
+            @Override
+            public void onCompleted() {
+                parent.timeout();
+            }
+        }
+    }
 }