How to send email reactive in spring web-flux

Question

I'd like to stay complete reactive within my new spring application. Therefor I use web-flux/ reactor and ReactiveRepository with MongoDB.

Do you know how to integrate java-mail reactively into the tech-stack? Any alternatives?

Answer 1

If you the bind a route to a send email function, the response will take several seconds, because it will be resolved after the email has been sent, even if you use ".subscribeOn(Schedulers.elastic())". So, I don't see the purpose of doing it reactively if this is the use case.

If you need a quick response and let the emails take as long as they need to be sent, you can wrap the send email function in a "CompletableFuture.runAsync":

Short code:

CompletableFuture.runAsync(() -> sendEmail());

Extended:

MimeMessage message = mailSender.createMimeMessage();
MimeMessageHelper helper = new MimeMessageHelper(message);
CompletableFuture.runAsync(() -> {
            try {
                helper.setFrom(EMAIL_FROM_ADDRESS, EMAIL_SENDER_NAME);
                helper.setTo(EMAIL_TO);
                helper.setSubject(EMAIL_SUBJECT);
                helper.setText(EMAIL_CONTENT, true);
                mailSender.send(message);
            } catch (MessagingException | UnsupportedEncodingException e) {
                // handle exceptions
            }
        });

Answer 2

The only useful non-blocking SMTP client I found and still use is https://vertx.io/docs/vertx-mail-client/java/
I even integrated it with spring-webflux and mongodb-driver-reactivestreams so that they share the same Netty EventLoopGroup.

Mono.create<MailResult> { sink ->
  mailClient.sendMail(email) { asyncResult ->
    if (asyncResult.succeeded()) {
      sink.success(asyncResult.result()
    } else {
      sink.error(asyncResult.cause()
    }
  }
}

Answer 3

How about using microsoft graph api, and use microsoft's exahnge servers to send emails https://learn.microsoft.com/en-us/graph/use-the-api. Its not the answer to original question, but I wonder same concept can be applied there or if anyone has some similar things using this API.

Answer 4

For sending email and still be non-blocking, you can run code about sending email in another thread. If you are using Spring WebFlux, this can be easily done, by wrapping your code for sending email in below factory methods of Mono (A Reactor library Publisher).

Mono.fromCallable()

or

Mono.fromRunnable()

Short Code

Mono.fromCallable(()-> sendEmail())
    .subscribe();

Where sendEmail() is your function to send email.

This is what is recommended in docs also - How Do I Wrap a Synchronous, Blocking Call?

Long Code

Below is complete sample code which i am using in my application -

    Mono.fromCallable(() -> {
            try {
                MimeMessageHelper helper = new MimeMessageHelper(sender.createMimeMessage());
                helper.setTo(to);
                helper.setSubject(subject);
                helper.setText(body);
                sender.send(helper.getMimeMessage());
                log.info("Email send successfully, subject {} , to {}", subject, to);
                return true;
            } catch (Exception e) {
                log.warn("Failed to send email with subject {}, due to {}",subject,e.getMessage(), e});
                return false;
            }

 )
.subscribe(result -> log.info("Mail sent {}", result));

And when it is reactive stack, never forget to subscribe :D

Answer 5

I also was looking for reactive SMTP client.

And I managed to find it ;)

Here it is: https://github.com/HubSpot/NioSmtpClient

from README:

High performance SMTP client in Java based on Netty. This client is well tested and heavily used at HubSpot.

I already verified it on local environment and it's realy reactive! However, it uses completableFuture instead of Mono or Flux so it will be necessary to wrap it manually.

In general this library looks good but I think that it would be better if the author provided some facade that would simplify SDK usage. (Anyway it's open-source so we can improve it) .

Here an example, how to use it(never mind on codeStyle, it's just for an example):

private static final String MESSAGE_DATA = "From: <sender@gmail.com\r\n" +
        "To: <recipient@gmail.com>\r\n" +
        "Subject: test mail\r\n\r\n" +
        "Hello stackOverFlow!";

public static void main(String[] args) {
    final SmtpSessionFactory smtpSessionFactory = createSmtpSessionFactory();

    final SmtpSessionConfig smtpSessionConfig = SmtpSessionConfig.builder().remoteAddress(InetSocketAddress.createUnresolved("smtp.gmail.com", 587)).build();
    Mono.fromFuture(smtpSessionFactory.connect(smtpSessionConfig))
            .flatMap(connection -> doInSession(connection, req(EHLO, "gmail.com")))
            .flatMap(connection -> Mono.fromFuture(connection.getSession().startTls()))
            .flatMap(connection -> Mono.fromFuture(connection.getSession().authLogin("sender@gmail.com", "SomeStrongPasswordLike123456")))
            .flatMap(connection -> doInSession(connection, req(MAIL, "FROM:<" + "sender@gmail.com" + ">")))
            .flatMap(connection -> doInSession(connection, req(RCPT, "TO:<" + "recipient@gmail.com" + ">")))
            .flatMap(connection -> doInSession(connection, req(DATA)))
            .map(connection -> connection.getSession()
                    .send(MessageContent.of(Unpooled.wrappedBuffer(MESSAGE_DATA.getBytes(StandardCharsets.UTF_8)))))
            .flatMap(Mono::fromFuture)
            .flatMap(connection -> doInSession(connection, req(QUIT)))
            .flatMap(connection -> Mono.fromFuture(connection.getSession().close()))
            .block();
}

private static SmtpSessionFactory createSmtpSessionFactory() {
    ThreadFactory threadFactory = new ThreadFactoryBuilder().setDaemon(true).setNameFormat("niosmtpclient-%d").build();
    final SmtpSessionFactoryConfig config = SmtpSessionFactoryConfig.builder()
            .eventLoopGroup(new NioEventLoopGroup(4, threadFactory))
            .executor(Executors.newCachedThreadPool(threadFactory))
            .build();
    return new SmtpSessionFactory(config);
}

private static Mono<SmtpClientResponse> doInSession(SmtpClientResponse connection, SmtpRequest request) {
    return Mono.fromFuture(connection.getSession().send(request));
}

private static SmtpRequest req(SmtpCommand command, CharSequence... arguments) {
    return new DefaultSmtpRequest(command, arguments);
}

How it works(in general):

1. we define session factory(see method createSmtpSessionFactory)
2. we open dialog(connection) with server
3. we start TLS
4. we do authentication with our credentials
5. we says email address of sender to the server
6. we says email address of recipient to the server
7. we start data sending phase
8. we send data (data must follow some pattern. From: ... To: ... Subject: .... see MESSAGE_DATA variable)
9. we notify server that we are finishing dialog.
10. we close session

Answer 6

I found a solution. It uses spring-boot-starter-data-mongodb-reactive and API of external services like Mailgun or SendGrid. Key point is to use reactive WebClient :

1. Build WebClient instance (for example, using Sendgrid API) :

   String endpoint = “https://api.sendgrid.com/v3/“;
   String sendUri = endpoint + “mail/send”;
   
   WebClient client = WebClient.builder().filter(…).clientConnector(new ReactorClientHttpConnector(HttpClient.create())).baseUrl(endpoint).build()
   

2. Implement response objects :

   @Data
   class Response implements Serializable {
       private boolean status;
       private String id;
       private String message;
   }
   
   @Data
   class NotificationStatusResponse implements Serializable {
   
       private LocalDateTime timestamp;
       private int status;
       private String message;
       private String traceId;
       private String responseId;
       private String providerResponseId;
       private String providerResponseMessage;
   }
   

3. Send your message :

   public Mono<NotificationStatusResponse> send(NotificationMessage<EmailId> email) throws NotificationSendFailedException {
   
       Mono<NotificationStatusResponse> response = null;
       try {
           MultiValueMap<String, Object> formMap = new LinkedMultiValueMap<>(); // email parameters here: from, to, subject, html etc.
           response = client.post().uri(sendUri)
           .header("Authorization", "Basic " + “your credentials here”)
           .contentType(MediaType.MULTIPART_FORM_DATA).syncBody(formMap).retrieve()
           .bodyToMono(Response.class).map(this::getNotificationStatusResponse)
           .doOnSuccess(message -> log.debug("sent email successfully"))
           .doOnError((error -> log.error("email failed ", error)));
       } catch (WebClientException webClientException) {
           throw new NotificationSendFailedException("webClientException received", webClientException);
       }
       return response;
   
       NotificationStatusResponse getNotificationStatusResponse(Response response) {
           NotificationStatusResponse notificationStatusResponse = new NotificationStatusResponse();
           notificationStatusResponse.setStatus(200);
           notificationStatusResponse.setTimestamp(LocalDateTime.now());
           notificationStatusResponse.setProviderResponseId(response.getId());
           notificationStatusResponse.setProviderResponseMessage(response.getMessage());
           return notificationStatusResponse;
       }
   }