Let's add the RabbitMQ required dependencies. The following dependencies should be added:

<dependency>
  <groupId>io.projectreactor.rabbitmq</groupId>
  <artifactId>reactor-rabbitmq</artifactId>
  <version>1.0.0.M1</version>
</dependency>

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

The first one is about the reactive implementation for RabbitMQ and the second one is the starter AMQP, which will set up some configurations automatically.

We want to configure some RabbitMQ exchanges, queues, and bindings. It can be done using the RabbitMQ client library. We will configure our required infrastructure for the Mail microservice.

Our configuration class should look like this:

package springfive.airline.mailservice.infra.rabbitmq;

// imports are omitted

@Configuration
public class RabbitMQConfiguration {

  private final String pass;

  private final String user;

  private final String host;

  private final Integer port;

  private final String mailQueue;

  public RabbitMQConfiguration(@Value("${spring.rabbitmq.password}") String pass,
      @Value("${spring.rabbitmq.username}") String user,
      @Value("${spring.rabbitmq.host}") String host,
      @Value("${spring.rabbitmq.port}") Integer port,
      @Value("${mail.queue}") String mailQueue) {
    this.pass = pass;
    this.user = user;
    this.host = host;
    this.port = port;
    this.mailQueue = mailQueue;
  }

  @Bean("springConnectionFactory")
  public ConnectionFactory connectionFactory() {
    CachingConnectionFactory factory = new CachingConnectionFactory();
    factory.setUsername(this.user);
    factory.setPassword(this.pass);
    factory.setHost(this.host);
    factory.setPort(this.port);
    return factory;
  }

  @Bean
  public AmqpAdmin amqpAdmin(@Qualifier("springConnectionFactory") ConnectionFactory connectionFactory) {
    return new RabbitAdmin(connectionFactory);
  }

  @Bean
  public TopicExchange emailExchange() {
    return new TopicExchange("email", true, false);
  }

  @Bean
  public Queue mailQueue() {
    return new Queue(this.mailQueue, true, false, false);
  }

  @Bean
  public Binding mailExchangeBinding(Queue mailQueue) {
    return BindingBuilder.bind(mailQueue).to(emailExchange()).with("*");
  }

  @Bean
  public Receiver receiver() {
    val options = new ReceiverOptions();
    com.rabbitmq.client.ConnectionFactory connectionFactory = new com.rabbitmq.client.ConnectionFactory();
    connectionFactory.setUsername(this.user);
    connectionFactory.setPassword(this.pass);
    connectionFactory.setPort(this.port);
    connectionFactory.setHost(this.host);
    options.connectionFactory(connectionFactory);
    return ReactorRabbitMq.createReceiver(options);
  }

}

There is interesting stuff here, but all of it is about infrastructure in RabbitMQ. It is important because when our application is in bootstrapping time, it means our application is preparing to run. This code will be executed and create the necessary queues, exchanges, and bindings. Some configurations are provided by the application.yaml file, look at the constructor.

Our Mail service is abstract and can be used for different purposes, so we will create a simple class to represent a mail message in our system. Our Mail class should look like this:

package springfive.airline.mailservice.domain;

import lombok.Data;

@Data
public class Mail {

  String from;

  String to;

  String subject;

  String message;

}

Easy, this class represents an abstract message on our system.

As we can expect, we will integrate with the SendGrid services through the REST APIs. In our case, we will use the reactive WebClient provided by Spring WebFlux.

Now, we will use the SendGrid API Key created in the previous section. Our MailSender class should look like this:

package springfive.airline.mailservice.domain.service;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpStatus;
import org.springframework.http.ReactiveHttpOutputMessage;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.BodyInserter;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import springfive.airline.mailservice.domain.Mail;
import springfive.airline.mailservice.domain.service.data.SendgridMail;

@Service
public class MailSender {

  private final String apiKey;

  private final String url;

  private final WebClient webClient;

  public MailSender(@Value("${sendgrid.apikey}") String apiKey,
      @Value("${sendgrid.url}") String url,
      WebClient webClient) {
    this.apiKey = apiKey;
    this.webClient = webClient;
    this.url = url;
  }

  public Flux<Void> send(Mail mail){
    final BodyInserter<SendgridMail, ReactiveHttpOutputMessage> body = BodyInserters
      .fromObject(SendgridMail.builder().content(mail.getMessage()).from(mail.getFrom()).to(mail.getTo()).subject(mail.getSubject()).build());
    return this.webClient.mutate().baseUrl(this.url).build().post()
        .uri("/v3/mail/send")
        .body(body)
        .header("Authorization","Bearer " + this.apiKey)
        .header("Content-Type","application/json")
        .retrieve()
        .onStatus(HttpStatus::is4xxClientError, clientResponse ->
            Mono.error(new RuntimeException("Error on send email"))
        ).bodyToFlux(Void.class);
  }

}

We received the configurations in the constructor, that is, the sendgrid.apikey and sendgrid.url. They will be configured soon. In the send() method, there are some interesting constructions. Look at BodyInserters.fromObject(): it allows us to send a JSON object in the HTTP body. In our case, we will create a SendGrid mail object.

In the onStatus() function, we can pass a predicate to handle the HTTP errors family. In our case, we are interested in the 4xx error family.

This class will process sending the mail messages, but it is necessary to listen to the RabbbitMQ queue, which we will do in the next section.

Let's create our MailQueueConsumer class, which will listen to the RabbitMQ queue. The class should look like this:

package springfive.airline.mailservice.domain.service;

import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import javax.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import reactor.rabbitmq.Receiver;
import springfive.airline.mailservice.domain.Mail;

@Service
@Slf4j
public class MailQueueConsumer {

  private final MailSender mailSender;

  private final String mailQueue;

  private final Receiver receiver;

  private final ObjectMapper mapper;

  public MailQueueConsumer(MailSender mailSender, @Value("${mail.queue}") String mailQueue,
      Receiver receiver, ObjectMapper mapper) {
    this.mailSender = mailSender;
    this.mailQueue = mailQueue;
    this.receiver = receiver;
    this.mapper = mapper;
  }

  @PostConstruct
  public void startConsume() {
    this.receiver.consumeAutoAck(this.mailQueue).subscribe(message -> {
      try {
        val mail = this.mapper.readValue(new String(message.getBody()), Mail.class);
        this.mailSender.send(mail).subscribe(data ->{
          log.info("Mail sent successfully");
        });
      } catch (IOException e) {
        throw new RuntimeException("error on deserialize object");
      }
    });
  }

}

The method annotated with @PostConstruct will be invoked after MailQueueConsumer is ready, which will mean that the injections are processed. Then Receiver will start to process the messages.

Now, we will run our Mail microservice. Find the MailServiceApplication class, the main class of our project. The main class should look like this:

package springfive.airline.mailservice;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.netflix.eureka.EnableEurekaClient;
import org.springframework.cloud.netflix.hystrix.EnableHystrix;
import org.springframework.cloud.netflix.zuul.EnableZuulProxy;

@EnableHystrix
@EnableZuulProxy
@EnableEurekaClient
@SpringBootApplication
public class MailServiceApplication {

  public static void main(String[] args) {
    SpringApplication.run(MailServiceApplication.class, args);
  }

}

It is a standard Spring Boot Application.

We can run the application in IDE or via the Java command line.

Run it!