Author – Erwin de Gier

On my current project, we opted not to use REST for the communication between our services. Instead we make use of AxonIQ’s AxonHub, which acts as a specialized message broker. Messages can be of three types:

• Command – You want to change something
• Event – You want to inform others of something that happened
• Query – You want to know something

The communication is asynchronous and we also have to deal with eventual consistency. If we would create an order by sending a CreateOrderCommand, this order would result in various events which update the state of the Order. We then need to send a query, of which we also receive the result asynchronous.

Our web and mobile frontend communicate with a microservices backend through a REST API. When the user creates an order (clicks the ‘buy’ button), they expect a result immediately. For both frontends, an API which models this user experience closely makes the most sense. This means that we want a more synchronous API, where we send a ‘create order’ request and immediately receive a response.

We implemented this using a small REST facade which translates our asynchronous communication in the backend to the synchronous communication for the frontend. This is a good use case for Spring 5’s reactive Webflux and Project Reactor. Using Project Reactor’s reactive API makes it possible to combine multiple asynchronous calls and operate on their result. Webflux handles the conversion of the reactive types (Mono, Flux) to REST responses. It optimizes the use of threads; by writing non-blocking code, we can reuse threads between asynchronous calls for handling other requests.

Diagram 1 gives us an overview of this approach.

Diagram 1

Implementation

Let’s have a more detailed look at the code for the create order example. Listing 1 shows the (slightly) simplified implementation of our REST controller method.

@PostMapping
public Mono<ResponseEntity<OrderResponse>> createOrder(CreateOrderRequest request) 
{
 CreateOrderCommand command = CreateOrderCommand.fromRequest(request); //1

 return this.commandGateway.send(command) //2
  .flatMap(id -> queryGateway.send(new FindOrderSummaryQuery(id))) //3
  .retryWhen(errors -> errors.delayElements(Duration.of(100, MILLIS)) //4
  .take(10)).concatWith(Mono.error(new RuntimeException())).next() //5
  .onErrorReturn(new OrderResponse(orderID, OrderStatus.CREATED)) //6
  .map(orderResponse -> ResponseEntity.ok().body(orderResponse)); //7
}

Listing 1

We first create a command out of the REST request (line 1). A command is a message with the specific intent to change something in our domain. In this specific case, we want to create a new order.

After creating the command, the two asynchronous calls we make are:

Mono<String> id = this.commandGateway.send(command);

and:

Mono<OrderResponse> orderResponse = queryGateway.send(new FindOrderSummaryQuery(id));

Both calls return a single value by using a Mono. A Mono is a reactive type, comparable to the Java’s CompletableFuture. It has zero or one element and can represent an error. As with all reactive types, the value (or error) is delivered over time.

The second call takes the result of the first call as its input. We need the returned id of the command to query for the order. We use the flatMap operator to achieve this (line 3). The flatMap takes the asynchronous result of call 1 and passes this as a parameter to the lambda of call 2. The callback version can be seen in Listing 2: notice the nested lambda, which makes the code complex and less readable.

this.commandGateway.send(command, id -> {
  queryGateway.send(new FindOrderSummaryQuery(id));
});

Listing 2

There is a delay between sending the command and being able to query the result. When the created order cannot be found (e.g. it isn’t create yet or something has gone wrong), an exception is thrown. In this chain, this is represented as a Mono.error(throwable). We use the retryWhen method to retry the query (line 4). We do this 10 times with a delay of 100 ms. When we still don’t get a result, we throw an error (line 5). We don’t expose the error to the client, but pass an OrderResponse with the id and status CREATED (line 6). The client can then query the status of the order later by using this id. This is a form of graceful degradation.

Finally we map the order response from the query to a response entity which can be returned by Spring. Spring actually subscribes to this whole chain and sends out the REST response for us.

Conclusion

Spring 5 and Project Reactor allow us to handle asynchronous communication with concise and readable code. We can do retries, error handling and the combination of multiple asynchronous calls in just a few lines.
The integration of Webflux with Project Reactor allows the use of reactive paradigms in a REST controller. Webflux uses an asynchronous approach. While we wait for a backend query to return, we don’t block the main thread. this allows it to be used for other requests.
Our specific use case is a good example of one of the applications of Spring 5 Webflux and Project Reactor.