Skip to content

Commit

Permalink
Merge pull request #740 from akarnokd/TimeoutWithSelector
Browse files Browse the repository at this point in the history 
Timeout with selector overloads
  • Loading branch information
@benjchristensen
benjchristensen committed Jan 14, 2014
2 parents 4bce409 + 34c5638 commit 57376d0
Show file tree
Hide file tree
Showing 3 changed files with 474 additions and 0 deletions.
80 changes: 80 additions & 0 deletions rxjava-core/src/main/java/rx/Observable.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7047,6 +7047,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.
Expand Down
158 changes: 158 additions & 0 deletions rxjava-core/src/main/java/rx/operators/OperationTimeout.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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();
}
}
}
}
Loading

0 comments on commit 57376d0

Please sign in to comment.