RxJava retry not working with Observer

Question

It is a very simple sample：

public static void main(String[] args) {
    Observable.from(ListUtils.asList(1, 2, 3)).retry(3).subscribe(new Observer<Integer>() {
        @Override
        public void onCompleted() {
            System.out.println("onCompleted");
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("onError");
        }

        @Override
        public void onNext(Integer integer) {
            System.out.println(integer);
            if (integer == 3)
                throw new RuntimeException("onNext exception");

        }
    });
}

The console output is: 1,2,3,onError. But my Expected: 1,2,3,1,2,3,1,2,3,onError.

Answer 1

Once than an error happens the subscriber unsubscriber from the observable, and in case you use operator retry it will retry only if the operator is not used in the main pipeline but in a flatMap operator

This one since the retry is after a flatMap it will work

@Test
public void retryInFlatMap() {
    Observable.from(Arrays.asList(1, 2, 3, 4))
            .flatMap(number -> Observable.just(number)
                    .doOnNext(n -> {
                        if (n == 2) {
                            throw new NullPointerException();
                        }
                    }))
                    .retry(3)
            .subscribe(n-> System.out.println("number:"+n));
}

This one, since is after a map it wont

int cont=0;
@Test
public void retryInMap() {
    Observable.from(Arrays.asList(1, 2, 3, 4))
            .map(number ->{
                        if (cont == 2) {
                            throw new NullPointerException();
                        }
                        cont++;
                        return number;
                    })
            .retry(3)
            .subscribe(n-> System.out.println("number:"+n));
}

If you want to see more examples take a look here https://github.com/politrons/reactive