Reactive programming with JAX-RS

Hello!

The latest Java Enterprise Developer course this year has been successfully launched and we have the latest material on this topic that we want to share with you about using asynchronous approach and styling to develop responsive reactive applications.

Go.
')
Reactive programming first sounds like the name of the emerging paradigm, but in fact, refers to a programming method that uses an event-oriented approach to work with asynchronous data streams. Based on constantly running data, reactive systems respond to them by performing a series of events.
Reactive programming follows the “Observer” design pattern, which can be defined as follows: if a state change occurs in one object, all other objects are notified and updated accordingly. Therefore, instead of polling events for changes, events are pushed asynchronously so that observers can process them. In this example, observers are functions that are executed when an event is sent. And the mentioned data stream is the actual observable.

Almost all languages ​​and frameworks use this approach in their ecosystem, and the latest versions of Java are no exception. In this article, I will explain how you can apply reactive programming using the latest version of JAX-RS in Java EE 8 and Java 8 functionality.



Jet Manifesto

The Reactive Manifesto lists four fundamental aspects that an application needs to be more flexible, loosely coupled and easy to scale, and therefore capable of being reactive. It says that the application must be responsive, flexible (and therefore scalable), robust and message-driven.

The underlying goal is a truly responsive application. Suppose there is an application in which one large stream deals with the processing of user requests, and after performing the work, this stream sends the answers back to the original requesters. When an application receives more requests than it can handle, this thread becomes a bottleneck, and the application loses its former responsiveness. To maintain responsiveness, the application must be scalable and sustainable. Sustainable can be considered an application in which there is functionality for auto-recovery. According to the experience of most developers, only the message-driven architecture allows the application to be scalable, stable and responsive.

Reactive programming was introduced in Java 8 and Java EE 8. Java introduced concepts such as CompletionStage and its implementation CompletableFuture , and Java began to use these functions in specifications such as the Reactive Client API in JAX-RS.

JAX-RS 2.1 Reactive Client API

Let's see how reactive programming can be used in Java EE 8 applications. To understand the process, you need a basic knowledge of the Java EE API.

JAX-RS 2.1 introduced a new way to create a REST client with support for reactive programming. The default invoker implementation offered by JAX-RS is synchronous, which means that the client being created will send a blocking call to the server’s endpoint. An example implementation is presented in Listing 1.

Listing 1

 Response response = ClientBuilder.newClient() .target("http://localhost:8080/service-url") .request() .get(); 

Starting with version 2.0, JAX-RS provides support for creating an asynchronous invoker on the client API using a simple call to the async() method, as shown in Listing 2.

Listing 2

 Future<Response> response = ClientBuilder.newClient() .target("http://localhost:8080/service-url") .request() .async() .get(); 

Using an asynchronous invoker on the client returns a Future instance with type javax.ws.rs.core.Response . This can lead to polling the response, calling future.get() , or registering a callback, which will be called when there is an available HTTP response. Both implementations are suitable for asynchronous programming, but are usually complicated if you want to group callbacks or add conditional cases to these asynchronous execution minima.

JAX-RS 2.1 provides a reactive way to overcome these problems with the new JAX-RS Reactive Client API for building a client. This is as simple as calling the rx() method during client building. In Listing 3 rx() method returns a reactive invoker that exists during client execution, and the client returns a response with the type CompletionStage.rx() , which allows the transition from synchronous invoker to asynchronous using a simple call.

Listing 3

 CompletionStage<Response> response = ClientBuilder.newClient() .target("http://localhost:8080/service-url") .request() .rx() .get(); 

CompletionStage<> is a new interface introduced in Java 8. It represents a calculation, which can be a step in a larger calculation, as the name suggests. This is the only representative of Java 8 reactivity that got into JAX-RS.
After receiving the response instance, I can call AcceptAsync() , where I can provide a snippet of code that will be executed asynchronously when the answer becomes available, as shown in Listing 4.

Listing 4

 response.thenAcceptAsync(res -> { Temperature t = res.readEntity(Temperature.class); //do stuff with t }); 

Adding reactivity to the endpoint REST

The reactive approach is not limited to the client side of JAX-RS; it can also be used on the server side. For example, first I’ll create a simple script where I can query the list of locations for one destination. For each position, I will make a separate call with the location data to another point to get the temperature values. The interaction of the destination points will be as shown in Figure 1.


Figure 1. Interaction between destination points

At first, I simply define the model of the domain, and then the services for each model. Listing 5 shows how the Forecast class is defined, which wraps the Location and Temperature classes.

Listing 5

 public class Temperature { private Double temperature; private String scale; // getters & setters } public class Location { String name; public Location() {} public Location(String name) { this.name = name; } // getters & setters } public class Forecast { private Location location; private Temperature temperature; public Forecast(Location location) { this.location = location; } public Forecast setTemperature( final Temperature temperature) { this.temperature = temperature; return this; } // getters } 

For wrapping a list of predictions, the ServiceResponse class ServiceResponse implemented in Listing 6.

Listing 6

 public class ServiceResponse { private long processingTime; private List<Forecast> forecasts = new ArrayList<>(); public void setProcessingTime(long processingTime) { this.processingTime = processingTime; } public ServiceResponse forecasts(List<Forecast> forecasts) { this.forecasts = forecasts; return this; } // getters } 

LocationResource shown in Listing 7 defines three sample locations returned with the /location path.

Listing 7

 @Path("/location") public class LocationResource { @GET @Produces(MediaType.APPLICATION_JSON) public Response getLocations() { List<Location> locations = new ArrayList<>(); locations.add(new Location("London")); locations.add(new Location("Istanbul")); locations.add(new Location("Prague")); return Response.ok(new GenericEntity<List<Location>>(locations){}).build(); } } 

TemperatureResource , shown in Listing 8, returns a randomly generated temperature value between 30 and 50 for a given location. A 500 ms delay is added to the implementation to simulate the sensor readout.

Listing 8

 @Path("/temperature") public class TemperatureResource { @GET @Path("/{city}") @Produces(MediaType.APPLICATION_JSON) public Response getAverageTemperature(@PathParam("city") String cityName) { Temperature temperature = new Temperature(); temperature.setTemperature((double) (new Random().nextInt(20) + 30)); temperature.setScale("Celsius"); try { Thread.sleep(500); } catch (InterruptedException ignored) { ignored.printStackTrace(); } return Response.ok(temperature).build(); } } 

First, I'll show the synchronous ForecastResource implementation (see Listing 9), which gives all the locations. Then, for each position, it calls temperature service to get the values ​​in degrees Celsius.

Listing 9

 @Path("/forecast") public class ForecastResource { @Uri("location") private WebTarget locationTarget; @Uri("temperature/{city}") private WebTarget temperatureTarget; @GET @Produces(MediaType.APPLICATION_JSON) public Response getLocationsWithTemperature() { long startTime = System.currentTimeMillis(); ServiceResponse response = new ServiceResponse(); List<Location> locations = locationTarget .request() .get(new GenericType<List<Location>>(){}); locations.forEach(location -> { Temperature temperature = temperatureTarget .resolveTemplate("city", location.getName()) .request() .get(Temperature.class); response.getForecasts().add( new Forecast(location).setTemperature(temperature)); }); long endTime = System.currentTimeMillis(); response.setProcessingTime(endTime - startTime); return Response.ok(response).build(); } } 

When the forecast destination is requested as /forecast , you will get a conclusion similar to that shown in Listing 10. Note that the request processing took 1.533 ms, which is logical, since a synchronous request for temperature values ​​from three different locations adds up to 1.5 ms

Listing 10

 { "forecasts": [ { "location": { "name": "London" }, "temperature": { "scale": "Celsius", "temperature": 33 } }, { "location": { "name": "Istanbul" }, "temperature": { "scale": "Celsius", "temperature": 38 } }, { "location": { "name": "Prague" }, "temperature": { "scale": "Celsius", "temperature": 46 } } ], "processingTime": 1533 } 

As long as everything goes according to plan. It is time to introduce reactive programming on the server side, where calls to each location can be executed in parallel after receiving all the locations. This clearly can improve the synchronous flow shown earlier. This is done in Listing 11, where the definition of the reactive version of the forecast service is shown.

Listing 11

 @Path("/reactiveForecast") public class ForecastReactiveResource { @Uri("location") private WebTarget locationTarget; @Uri("temperature/{city}") private WebTarget temperatureTarget; @GET @Produces(MediaType.APPLICATION_JSON) public void getLocationsWithTemperature(@Suspended final AsyncResponse async) { long startTime = System.currentTimeMillis(); //   (stage)    CompletionStage<List<Location>> locationCS = locationTarget.request() .rx() .get(new GenericType<List<Location>>() {}); //      , //  ,   , //    CompletionStage final CompletionStage<List<Forecast>> forecastCS = locationCS.thenCompose(locations -> { //      //   ompletionStage List<CompletionStage<Forecast>> forecastList = //      //     locations.stream().map(location -> { //     //      //    final CompletionStage<Temperature> tempCS = temperatureTarget .resolveTemplate("city", location.getName()) .request() .rx() .get(Temperature.class); //   CompletableFuture,   //    //      return CompletableFuture.completedFuture( new Forecast(location)) .thenCombine(tempCS, Forecast::setTemperature); }).collect(Collectors.toList()); //    CompletableFuture, //     completable future //  return CompletableFuture.allOf( forecastList.toArray( new CompletableFuture[forecastList.size()])) .thenApply(v -> forecastList.stream() .map(CompletionStage::toCompletableFuture) .map(CompletableFuture::join) .collect(Collectors.toList())); }); //   ServiceResponse, //       //    . //   future    // forecastCS,    //      CompletableFuture.completedFuture( new ServiceResponse()) .thenCombine(forecastCS, ServiceResponse::forecasts) .whenCompleteAsync((response, throwable) -> { response.setProcessingTime( System.currentTimeMillis() - startTime); async.resume(response); }); } } 

A reactive implementation may seem complicated at first glance, but after a closer look, you will notice that it is quite simple. In the implementation of ForecastReactiveResource I first create a client call to location services using the JAX-RS Reactive Client API. As I mentioned above, this is an add-on for Java EE 8, and it helps to create a reactive call simply by using the rx() method.

Now I create a new stage based on location to collect a list of forecasts. They will be stored as a list of forecasts in one large completion stage, called forecastCS . Ultimately, I will create a service call response using only forecastCS .

And now, let's collect the forecasts in the form of a list of completion stages defined in the variable forecastList . To create a completion stage for each forecast, I transmit the data by location, and then create the tempCS variable, again using the JAX-RS Reactive Client API, which calls the temperature service with the name of the city. Here, I use the resolveTemplate() method to build the client, and this allows me to pass the city name to the collector as a parameter.

As the last step of streaming, I make a call to CompletableFuture.completedFuture() , passing the new Forecast instance as a parameter. I combine this future with the tempCS stage so that I have a temperature value for the iterated locations.

The CompletableFuture.allOf() method in Listing 11 converts the completion stage list to forecastCS . Performing this step returns a large instance of a completable future when all the objects provided are completable future.

The service response is an instance of the ServiceResponse class, so I create a completed future, and then combine the forecastCS completion stage with the list of forecasts and calculate the response time of the service.

Of course, reactive programming forces only the server side to run asynchronously; the client side will be blocked until the server sends the response back to the requester. To overcome this problem, Server Sent Events (SSEs) can be used to partially send a response as soon as it is available, so that the temperature values ​​for each location are transmitted to the client one by one. The output of the ForecastReactiveResource will be similar to that presented in Listing 12. As shown in the output, the processing time is 515 ms, which is the ideal execution time for obtaining temperature values ​​from one location.

Listing 12

 { "forecasts": [ { "location": { "name": "London" }, "temperature": { "scale": "Celsius", "temperature": 49 } }, { "location": { "name": "Istanbul" }, "temperature": { "scale": "Celsius", "temperature": 32 } }, { "location": { "name": "Prague" }, "temperature": { "scale": "Celsius", "temperature": 45 } } ], "processingTime": 515 } 

Conclusion

In the examples of this article, I first showed the synchronous method of obtaining forecasts using location and temperature services. Then, I switched to the reactive approach so that asynchronous processing was performed between service calls. When you use the JAX-RS Reactive Client API in Java EE 8 along with the CompletionStage and CompletableFuture classes available in Java 8, the power of asynchronous processing is pulled free by reactive programming.

Reactive programming is more than just an implementation of an asynchronous model from a synchronous one; it also simplifies working with concepts such as the nesting stage. The more it is used, the easier it will be to manage complex scenarios in parallel programming.

THE END

Thanks for attention. As always, we are waiting for your comments and questions.