Reputation: 20278
Suppose I have List<CharSequence> observableList
, which contains random CharSequence
's which have length from 1 to 10.
I have such Observable
:
Observable.from(observableList)
.flatMap(new Func1<CharSequence, Observable<CharSequence>>() {
@Override
public Observable<CharSequence> call(CharSequence charSequence) {
if (charSequence.length() == 1) {
return Observable.error(new RuntimeException("Too short"));
} else {
return Observable.just(charSequence);
}
}
}).retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() {
@Override
public Observable<?> call(final Observable<? extends Throwable> observable) {
return observable.flatMap(new Func1<Throwable, Observable<?>>() {
@Override
public Observable<?> call(Throwable throwable) {
if (backoffStrategy.isApplicable(throwable)) {
Log.d(MainActivity.class.getSimpleName(), "Appropriate throwable is thrown!");
return backoffStrategy.call(observable);
}
return observable;
}
});
}
})
The observable throws exception, when the sequence length is 1. Inside retryWhen
's Func1
I would like to check error type (RuntimeException
is just an example) and choose appropriate retry strategy.
This is the backoff strategy:
public class BaseBackoffStrategy implements BackoffStrategy {
@Override
public Observable<?> call(Observable<? extends Throwable> attempts) {
return attempts.zipWith(Observable.range(1, 3 + 1), new ObservableUtils.RxPair<Throwable, Integer>())
.flatMap(new Func1<Pair<Throwable, Integer>, Observable<?>>() {
@Override
public Observable<?> call(Pair<Throwable, Integer> ti) {
if (ti.second <= 3) {
System.out.println(new Date().toGMTString() + " : " + ti.second + " retry");
return Observable.timer((long) Math.pow(2, ti.second), TimeUnit.SECONDS);
} else {
return Observable.error(ti.first);
}
}
});
}
@Override
public boolean isApplicable(Throwable throwable) {
return RuntimeException.class.isInstance(throwable);
}
}
private interface BackoffStrategy extends Func1<Observable<? extends Throwable>, Observable<?>> {
boolean isApplicable(Throwable throwable);
}
And the function just returning a Pair
of objects:
public class ObservableUtils {
public static class RxPair<T1, T2> implements Func2<T1, T2, Pair<T1, T2>> {
@Override
public Pair<T1, T2> call(T1 t1, T2 t2) {
return Pair.of(t1, t2);
}
}
}
The ouput from the console is:
D/MainActivity: Appropriate throwable is thrown!
I/System.out: 6 Oct 2016 07:34:34 GMT : 1 retry
D/MainActivity: Appropriate throwable is thrown!
I/System.out: 6 Oct 2016 07:34:36 GMT : 1 retry
I/System.out: 6 Oct 2016 07:34:36 GMT : 2 retry
D/MainActivity: Appropriate throwable is thrown!
I/System.out: 6 Oct 2016 07:34:38 GMT : 1 retry
I/System.out: 6 Oct 2016 07:34:38 GMT : 3 retry
I/System.out: 6 Oct 2016 07:34:38 GMT : 2 retry
D/MainActivity: Appropriate throwable is thrown!
I/System.out: 6 Oct 2016 07:34:40 GMT : 1 retry
I/System.out: 6 Oct 2016 07:34:40 GMT : 3 retry
I/System.out: 6 Oct 2016 07:34:40 GMT : 2 retry
D/MainActivity: Appropriate throwable is thrown!
I/System.out: 6 Oct 2016 07:34:40 GMT : 1 retry
I/System.out: 6 Oct 2016 07:34:40 GMT : 3 retry
I/System.out: 6 Oct 2016 07:34:40 GMT : 2 retry
D/MainActivity: onError
But I would like the Observable
to retry after the specified time. The time should become longer, when the next exception is thrown. The output from console should look like:
D/MainActivity: Appropriate throwable is thrown!
I/System.out: 6 Oct 2016 07:34:34 GMT : 1 retry
D/MainActivity: Appropriate throwable is thrown!
I/System.out: 6 Oct 2016 07:34:36 GMT : 2 retry
D/MainActivity: Appropriate throwable is thrown!
I/System.out: 6 Oct 2016 07:34:38 GMT : 3 retry
D/MainActivity: Appropriate throwable is thrown!
D/MainActivity: onError
My question is: what am I doing wrong here? Can I call other
Func1
fromFunc1
?
Upvotes: 0
Views: 2023
Reputation: 1176
When using RxJava it is better use lamdas rather inner classes. You code become more readable.
I think you have a mistake here return backoffStrategy.call(observable);
Every time when retryWhen
emits an exception you recreate zip + timer. That's why you have single, double, triple and etc. messages like
I/System.out: 6 Oct 2016 07:34:40 GMT : 1 retry
This is bad design
private interface BackoffStrategy
extends Func1<Observable<? extends Throwable>, Observable<?>> {
Why do you need implement rx interfaces in the class which just creates another sequence?
I rewrite your code a little bit using lambdas and move code from BackoffStrategy.call
into sequence
BackoffStrategy backoffStrategy = new BaseBackoffStrategy();
List<CharSequence> observableList = Arrays.asList("a");
Observable.from(observableList)
.flatMap(charSequence -> {
if (charSequence.length() == 1) {
return Observable.error(new RuntimeException("Too short"));
} else {
return Observable.just(charSequence);
}
})
.retryWhen(observable ->
observable
.filter(backoffStrategy::isApplicable)
.doOnNext(next -> System.out.println("Appropriate throwable is thrown!"))
.zipWith(Observable.range(1, 3 + 1), Tuple::new)
.flatMap(tuple -> {
Integer attempts = tuple.getRight();
if (attempts <= 3) {
System.out.println(new Date().toGMTString() + " : " + attempts + " retry");
return Observable.timer((long) Math.pow(2, attempts), TimeUnit.SECONDS);
} else {
return Observable.error(tuple.getLeft());
}
})
)
.toBlocking()
.subscribe(
next -> System.out.println("Next: " + next),
error -> System.out.println("Error: " + error),
() -> System.out.println("Completed")
);
The output is
Appropriate throwable is thrown!
7 Oct 2016 12:24:24 GMT : 1 retry
Appropriate throwable is thrown!
7 Oct 2016 12:24:26 GMT : 2 retry
Appropriate throwable is thrown!
7 Oct 2016 12:24:30 GMT : 3 retry
Appropriate throwable is thrown!
Error: java.lang.RuntimeException: Too short
The Tuple
class
private static class Tuple<T, V> {
private final T left;
private final V right;
public Tuple(T left, V right) {
this.left = left;
this.right = right;
}
public T getLeft() {
return left;
}
public V getRight() {
return right;
}
}
Upvotes: 2