1. Introduction
In this tutorial, we look at JAX-RS support for reactive (Rx) programming using the Jersey API. This article assumes the reader has knowledge of the Jersey REST client API.
Some familiarity with reactive programming concepts will be helpful but isn’t necessary.
2. Dependencies
First, we need the standard Jersey client library dependencies:
<dependency>
<groupId>org.glassfish.jersey.core</groupId>
<artifactId>jersey-client</artifactId>
<version>2.27</version>
</dependency>
<dependency>
<groupId>org.glassfish.jersey.inject</groupId>
<artifactId>jersey-hk2</artifactId>
<version>2.27</version>
</dependency>
These dependencies give us stock JAX-RS reactive programming support. The current versions of jersey-client and jersey-hk2 are available on Maven Central.
For third-party reactive framework support, we’ll use these extensions:
<dependency>
<groupId>org.glassfish.jersey.ext.rx</groupId>
<artifactId>jersey-rx-client-rxjava</artifactId>
<version>2.27</version>
</dependency>
The dependency above provides support for RxJava’s Observable; for the newer RxJava2’s Flowable, we use the following extension:
<dependency>
<groupId>org.glassfish.jersey.ext.rx</groupId>
<artifactId>jersey-rx-client-rxjava2</artifactId>
<version>2.27</version>
</dependency>
The dependencies to rxjava and rxjava2 are also available on Maven Central.
3. Why We Need Reactive JAX-RS Clients
Let’s say we have three REST APIs to consume:
- the id-service provides a list of Long user IDs
- the name-service provides a username for a given user ID
- the hash-service will return a hash of both the user ID and the username
We create a client for each of the services:
Client client = ClientBuilder.newClient();
WebTarget userIdService = client.target("http://localhost:8080/id-service/ids");
WebTarget nameService
= client.target("http://localhost:8080/name-service/users/{userId}/name");
WebTarget hashService = client.target("http://localhost:8080/hash-service/{rawValue}");
This is a contrived example but it works for the purpose of our illustration. The JAX-RS specification supports at least three approaches for consuming these services together:
- Synchronous (blocking)
- Asynchronous (non-blocking)
- Reactive (functional, non-blocking)
3.1. The Problem With Synchronous Jersey Client Invocation
The vanilla approach to consuming these services will see us consuming the id-service to get the user IDs, and then calling the name-service and hash-service APIs sequentially for each ID returned.
With this approach, each call blocks the running thread until the request is fulfilled, spending a lot of time in total to fulfill the combined request. This is clearly less than satisfactory in any non-trivial use case.
3.2. The Problem With Asynchronous Jersey Client Invocation
A more sophisticated approach is to use the InvocationCallback mechanism supported by JAX-RS. At its most basic form, we pass a callback to the get method to define what happens when the given API call completes.
While we now get true asynchronous execution (with some limitations on thread efficiency), it’s easy to see how this style of code can get unreadable and unwieldy in anything but trivial scenarios. The JAX-RS specification specifically highlights this scenario as the Pyramid of Doom:
// used to keep track of the progress of the subsequent calls
CountDownLatch completionTracker = new CountDownLatch(expectedHashValues.size());
userIdService.request()
.accept(MediaType.APPLICATION_JSON)
.async()
.get(new InvocationCallback<List<Long>>() {
@Override
public void completed(List<Long> employeeIds) {
employeeIds.forEach((id) -> {
// for each employee ID, get the name
nameService.resolveTemplate("userId", id).request()
.async()
.get(new InvocationCallback<String>() {
@Override
public void completed(String response) {
hashService.resolveTemplate("rawValue", response + id).request()
.async()
.get(new InvocationCallback<String>() {
@Override
public void completed(String response) {
//complete the business logic
}
// ommitted implementation of the failed() method
});
}
// omitted implementation of the failed() method
});
});
}
// omitted implementation of the failed() method
});
// wait for inner requests to complete in 10 seconds
if (!completionTracker.await(10, TimeUnit.SECONDS)) {
logger.warn("Some requests didn't complete within the timeout");
}
So we’ve achieved asynchronous, time-efficient code, but:
- it’s difficult to read
- each call spawns a new thread
Note that we’re using a CountDownLatch in all code examples in order to wait for all expected values to be delivered by the hash-service. We do this so that we can assert that the code works in a unit test by checking that all expected values actually were delivered.
A usual client would not wait, but rather do whatever should be done with the result within the callback in order not to block the thread.
3.3. The Functional, Reactive Solution
A functional and reactive approach will give us:
- Great code readability
- Fluent coding style
- Effective thread management
JAX-RS supports these objectives in the following components:
- CompletionStageRxInvoker supports the CompletionStage interface as the default reactive component
- RxObservableInvokerProvider supports RxJava’s Observable
- RxFlowableInvokerProvider support RxJava’s Flowable
There is also an API for adding support for other Reactive libraries.
4. JAX-RS Reactive Component Support
4.1. CompletionStage in JAX-RS
Using the CompletionStage and its concrete implementation – CompletableFuture – we can write an elegant, non-blocking and fluent service call orchestration.
Let’s start by retrieving the user IDs:
CompletionStage<List<Long>> userIdStage = userIdService.request()
.accept(MediaType.APPLICATION_JSON)
.rx()
.get(new GenericType<List<Long>>() {
}).exceptionally((throwable) -> {
logger.warn("An error has occurred");
return null;
});
The rx() method call is the point from which the reactive handling kicks in. We use the exceptionally function to fluently define our exception handling scenario.
From here, we can cleanly orchestrate the calls to retrieve the username from the Name service and then hash the combination of both the name and user ID:
List<String> expectedHashValues = ...;
List<String> receivedHashValues = new ArrayList<>();
// used to keep track of the progress of the subsequent calls
CountDownLatch completionTracker = new CountDownLatch(expectedHashValues.size());
userIdStage.thenAcceptAsync(employeeIds -> {
logger.info("id-service result: {}", employeeIds);
employeeIds.forEach((Long id) -> {
CompletableFuture completable = nameService.resolveTemplate("userId", id).request()
.rx()
.get(String.class)
.toCompletableFuture();
completable.thenAccept((String userName) -> {
logger.info("name-service result: {}", userName);
hashService.resolveTemplate("rawValue", userName + id).request()
.rx()
.get(String.class)
.toCompletableFuture()
.thenAcceptAsync(hashValue -> {
logger.info("hash-service result: {}", hashValue);
receivedHashValues.add(hashValue);
completionTracker.countDown();
}).exceptionally((throwable) -> {
logger.warn("Hash computation failed for {}", id);
return null;
});
});
});
});
if (!completionTracker.await(10, TimeUnit.SECONDS)) {
logger.warn("Some requests didn't complete within the timeout");
}
assertThat(receivedHashValues).containsAll(expectedHashValues);
In the sample above, we compose our execution of the 3 services with fluent and readable code.
The method thenAcceptAsync will execute the supplied function after the given CompletionStage has completed execution (or thrown an exception).
Each successive call is non-blocking, making judicious use of system resources.
The CompletionStage interface provides a wide variety of staging and orchestration methods that allow us to compose, order and asynchronously execute any number of steps in a multi-step orchestration (or a single service call).
4.2. Observable in JAX-RS
To use the Observable RxJava component, we must first register the RxObservableInvokerProvider provider (and not the “ObservableRxInvokerProvider” as is stated in the Jersey specification document) on the client:
Client client = client.register(RxObservableInvokerProvider.class);
Then we override the default invoker:
Observable<List<Long>> userIdObservable = userIdService
.request()
.rx(RxObservableInvoker.class)
.get(new GenericType<List<Long>>(){});
From this point, we can use standard Observable semantics to orchestrate the processing flow:
userIdObservable.subscribe((List<Long> listOfIds)-> {
/** define processing flow for each ID */
});
4.3. Flowable in JAX-RS
The semantics for using RxJava Flowable is similar to that of Observable. We register the appropriate provider:
client.register(RxFlowableInvokerProvider.class);
Then we supply the RxFlowableInvoker:
Flowable<List<Long>> userIdFlowable = userIdService
.request()
.rx(RxFlowableInvoker.class)
.get(new GenericType<List<Long>>(){});
Following that, we can use the normal Flowable API.
5. Conclusion
The JAX-RS specification supplies a good number of options that yield clean, non-blocking execution of REST calls.
The CompletionStage interface, in particular, provides a robust set of methods that cover a variety of service orchestration scenarios, as well as opportunities to supply custom Executors for more fine-grained control of the thread management.
You can check out the code for this article over on Github.