Reactive Java Programming by Unknown

Author:Unknown
Language: eng
Format: epub

public void onError(Throwable e) {

System.err.println("error! " + e.toString());

}

@Override

public void onNext(Integer item) {

System.out.println("next item is: " + item);

}

});

Whenever an error occurs, retry() will resubscribe to the source, restarting with the emission and encountering the error again and again. As you may have guessed, this code will run forever, emitting the following output:

...

next item is: 1

next item is: 2

next item is: 1

next item is: 2

next item is: 1

...

CHAPTER 3 ■ SUBSCRIPTION LIFECYCLE

You can use retry(1) instead of retry() , so the retry mechanism will be executed only one time (the argument passed in determines the number of retries): Observable.just("1", "2", "a", "3", "4")

.map(new Func1<String, Integer>() {

@Override

public Integer call(String s) {

return Integer.parseInt(s);

}

})

.retry(1)

.subscribe(new Subscriber<Integer>() {

@Override

public void onCompleted() {

System.out.println("sequence completed!");

}

@Override

public void onError(Throwable e) {

System.err.println("error! " + e.toString());

}

@Override

public void onNext(Integer item) {

System.out.println("next item is: " + item);

}

});

In this case, the emission will be retried just once; after that, if the error condition persists, the error event will be propagated. The output is

next item is: 1

next item is: 2

next item is: 1

next item is: 2

error! java.lang.NumberFormatException : For input string: "a" If you want to retry one time after 5 seconds, you can use retryWhen() :

.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() { @Override

public Observable<?> call(Observable<? extends Throwable> observable) { return Observable.timer(5, TimeUnit.SECONDS);

}

})

After the specified timeout (5 seconds in this example), the sequence will be retried. If an error occurs during the retry, the error event will not be notified. Instead the complete event will be notified. The output is

CHAPTER 3 ■ SUBSCRIPTION LIFECYCLE

next item is: 1

next item is: 2

next item is: 1

next item is: 2

sequence completed!

In the following more complex example, you combine different operators to get three retries, one every 5 seconds:

.retryWhen(new Func1<Observable<? extends Throwable>, Observable<?>>() { @Override

public Observable<?> call(Observable<? extends Throwable> observable) { return observable.zipWith(Observable.range(1, 3),

new Func2<Throwable, Integer, Integer>() {

@Override

public Integer call(Throwable throwable,

Integer retryCount) {

System.out.println("retry #" + retryCount);

return retryCount;

}

}).flatMap(new Func1<Integer, Observable<?>>() {

@Override

public Observable<?> call(Integer integer) {

return Observable.timer(5, TimeUnit.SECONDS);

}

});

}

})

The idea here is to combine a sequence of three items (because you want three retries) with a delay of 5 seconds for each retry. The sequence of three items is generated by Observable.range(1,3) (also Observable.just(1,2,3) would work). For the delay, you still use Observable.timer(5, TimeUnit.SECONDS) . For every integer emitted, a timer is launched, and this timer triggers the retry mechanism. In addition, you print the current retry count.

The Observable.zipWith() operator is applied to the source observable and takes two parameters:

• An observable ( Observable.range(1, 3) in this example) that is the observable that you want to zip with the source observable (as with the Observable.zip() operator).

• An instance of Func2<Throwable, Integer, Integer> . A Func2 is required because you have two inputs: the Throwable emitted by the retryWhen operator and an Integer emitted by the Observable.range(1,3) .

Then you use flatMap to transform the items emitted by the zipWith operator into a 5-second delay. The resulting observable is used by retryWhen to apply the retry logic to the source observable.

CHAPTER 3 ■ SUBSCRIPTION LIFECYCLE

If you apply this retry mechanism to the previous example, you get the following output:

next

Download

Copyright Disclaimer:
This site does not store any files on its server. We only index and link to content provided by other sites. Please contact the content providers to delete copyright contents if any and email us, we'll remove relevant links or contents immediately.