Project Reactor async send email with retry on error

Question

I need to send some data after user registered. I want to do first attempt in main thread, but if there are any errors, I want to retry 5 times with 10 minutes interval.

@Override
public void sendRegisterInfo(MailData data) {
    Mono.just(data)
        .doOnNext(this::send)
        .doOnError(ex -> logger.warn("Main queue {}", ex.getMessage()))
        .doOnSuccess(d -> logger.info("Send mail to {}", d.getRecipient()))
        .onErrorResume(ex -> retryQueue(data))
        .subscribe();
}

private Mono<MailData> retryQueue(MailData data) {
    return Mono.just(data)
               .delayElement(Duration.of(10, ChronoUnit.MINUTES))
               .doOnNext(this::send)
               .doOnError(ex -> logger.warn("Retry queue {}", ex.getMessage()))
               .doOnSuccess(d -> logger.info("Send mail to {}", d.getRecipient()))
               .retry(5)
               .subscribe();
}

It works. But I've got some questions:

1. Did I correct to make operation in doOnNext function?
2. Is it correct to use delayElement to make a delay between executions?
3. Did the thread blocked when waiting for delay?
4. And what the best practice to make a retries on error and make a delay between it?

Answer 1

1. doOnXXX for logging is fine. But for the actual element processing, you must prefer using flatMap rather than doOnNext (assuming your processing is asynchronous / can be converted to returning a Flux/Mono).

2. This is correct. Another way is to turn the code around and start from a Flux.interval, but here delayElement is better IMO.

3. The delay runs on a separate thread/scheduler (by default, Schedulers.parallel()), so not blocking the main thread.

4. There's actually a Retry builder dedicated to that kind of use case in the reactor-extra addon: https://github.com/reactor/reactor-addons/blob/master/reactor-extra/src/main/java/reactor/retry/Retry.java