Now that we've seen RxJava support in RESTEasy, we're ready to build on more reactive applications to illustrate common reactive use-cases.

Let's create an application with several types of requests, that we collect statistics on, such as the number of times each request has been called. And let's store those statistics in a Redis instance. For that we will use the Vert.x Redis client because it supports RxJava out of the box.

A sample application

So we will be importing the following Maven modules:

<!-- For RESTEasy -->
<dependency>
    <groupId>org.jboss.resteasy</groupId>
    <artifactId>resteasy-jaxrs</artifactId>
    <version>4.0.0-SNAPSHOT</version>
</dependency>
<!-- For RESTEasy's support of RxJava 2 -->
<dependency>
    <groupId>org.jboss.resteasy</groupId>
    <artifactId>resteasy-rxjava2</artifactId>
    <version>4.0.0-SNAPSHOT</version>
</dependency>
<!-- For the Vert.x Redis client -->
<dependency>
    <groupId>io.vertx</groupId>
    <artifactId>vertx-redis-client</artifactId>
    <version>3.5.3</version>
</dependency>
<!-- For the Vert.x RxJava 2 support -->
<dependency>
    <groupId>io.vertx</groupId>
    <artifactId>vertx-rx-java2</artifactId>
    <version>3.5.3</version>
</dependency>

Now, in order to make sure I get a single Redis client for my application, I will have to make it injectable by RESTEasy with the @Context annotation, and there's no support for pluggable injection in JAX-RS so it's a little convoluted, but we can achieve that with the help of this custom Feature:

@Provider
public class RedisFeature implements Feature {

  private RedisClient redis;

  public RedisFeature(){
    // connect to the local Redis
    redis = RedisClient.create(Vertx.vertx());
  }

  public boolean configure(FeatureContext context) {
    // this is tied to the deployment, which is what we want for the redis client
    if(context.getConfiguration().getRuntimeType() == RuntimeType.CLIENT)
      return false;
    Dispatcher dispatcher = ResteasyProviderFactory.getContextData(Dispatcher.class);
    if(dispatcher == null) {
      // this can happen, but it means we're not able to find a deployment
      return false;
    }
    dispatcher.getDefaultContextObjects().put(RedisClient.class, redis);
    return true;
  }
}

We can now write our three requests that collect usage statistics (they inject the RedisClient):

@Path("/")
public class Resource {
  @Context
  private RedisClient redis;

  @Path("req1")
  @GET
  public Single<String> req1() {
    return redis.rxIncr("req1.count").map(count -> "Req1 count: "+count);
  }

  @Path("req2")
  @GET
  public Single<String> req2() {
    return redis.rxIncr("req2.count").map(count -> "Req2 count: "+count);
  }

  @Path("req3")
  @GET
  public Single<String> req3() {
    return redis.rxIncr("req3.count").map(count -> "Req3 count: "+count);
  }
}

As you can see we count usage in the Redis keys req1.count, req2.count and req3.count.

Now, if we want to display them, we have to get all three values, which either means a lot of nesting with RxJava, or (better) using the Single.zip operator:

@GET
public Single<String> info(){
  return Single.zip(redis.rxGet("req1.count"), redis.rxGet("req2.count"), redis.rxGet("req3.count"),
    (req1, req2, req3) -> "Request 1: "+req1+"\nRequest 2: "+req2+"\nRequest 2: "+req3);  
}

As you can see, with RxJava, getting several values is a little more verbose than if we were doing it in blocking style. In fact, in real applications it is very common to start most requests with actions that depend on resolving a few asynchronous values. They can be waiting for database results, querying caches, or even obtaining permission lists, but eventually, lots of your requests will start with a Single.zip call to get the values you need in your request. That's annoying, and when they are often the same values, that's just plain boilerplate.

The solution

What if RESTEasy could take all those async values that you need, and resolve them before it called your resource method? This is called asynchronous injection, and the latest RESTEasy does just that.

The async values we want to be resolved are originally of type Single<String>, so their resolved value is of type String. In order to get async injection, we annotate our injected resolved value with @Context, and RESTEasy will look up a ContextInjector that is declared to resolve values to String. In our case, we declare our ContextInjector to resolve values from type Single<String> to String, but any async type is supported, thanks to the existing support for pluggable async types.

Once we've declared our ContextInjector, RESTEasy will call it to get the Single<String> that provides the String we want asynchronously injected, and will wait for the async value to be resolved, and only then proceed to inject it in your resource method. This way, when you start your resource method, you already have all your async values resolved!

For our example, we're going to describe our redis queries with the @RedisQuery annotation:

@Target(ElementType.PARAMETER)
@Retention(RetentionPolicy.RUNTIME)
public @interface RedisQuery {
  String value();
}

Now we can write our new resource method that wants these redis queries injected:

@Path("inject")
@GET
public String infoInjection(@Context @RedisQuery("req1.count") String req1,
    @Context @RedisQuery("req2.count") String req2,
    @Context @RedisQuery("req3.count") String req3){
  return "Request 1: "+req1+"\nRequest 2: "+req2+"\nRequest 2: "+req3;
}

And all we have to do for that async injection to work is to declare our ContextInjector:

@Provider
public class RedisInjector implements ContextInjector<Single<String>, String> {

  @Override
  public Single<String> resolve(Class<?> rawType, Type genericType, Annotation[] annotations) {
    RedisClient redisClient = ResteasyProviderFactory.getContextData(RedisClient.class);
    for (Annotation annotation : annotations) {
      if(annotation.annotationType() == RedisQuery.class) {
        String query = ((RedisQuery) annotation).value();
         // let's inject !
        return redisClient.rxGet(query);
      }
    }
    // not for us: try other injectors
    return null;
  }
}

As you can see, we just have to declare that our ContextInjector can provide values of type String via an async type Single<String>, and check the annotations on the injection point to figure out what query to run.

As I mentioned previously, this is not limited to async values of type Single, because any async value type is supported via plugins, and in fact only CompletionStage is supported by default (Single being provided by the resteasy-rxjava2 module we're using).

Conclusion

We've removed yet another common cause of boilerplate: rejoice!

Async injection was added in the latest 4.0.0.Beta4 release (RESTEASY-1905). Go ahead and try it out while it's fresh!

JAX-RS 1.0: The blocking origins

Originally, JAX-RS code use to be blocking and straightforward, like this resource for example:

@Path("/hello")
public class HelloResource {

    @Path("classic")
    @GET
    public String classic() {
        return "Hello World";
    }

}

Here, the resource method would return a value immediately. If it would take a long time producing that value, the server would make one of its thread block for the entire duration of the request until it was returned, but more often than not, producing values took negligible time.

I say negligible, but with the appearance of micro-services, this became much less true. Let's take for example a micro-service that runs parallel to our first service, and delegates to it:

@Path("/service")
public class ServiceResource {


    @Context 
    private UriInfo uriInfo;
    
    private URI getUri(Class klass, String method) {
        return uriInfo.getBaseUriBuilder().path(klass).path(klass, method).build();
    }
    
    @Path("classic")
    @GET
    public String classic() {
        Client client = ClientBuilder.newClient();
        try {
            URI uri = getUri(HelloResource.class, "classic");
            String entity = client.target(uri).request().get(String.class);
            return "Service got: "+entity;
        } finally {
            client.close();
        }
    }
}

Here we're using the JAX-RS 2.0 REST client to make a blocking HTTP call to our classic hello resource, and then we return it. This is already a heavier class of resource method: one where the thread delegated to it will have to wait for IO for the web-service call to complete, before the response is returned to the client. So it will be more thread-intensive, where those threads will mostly sit around waiting for IO, which is less than optimal.

JAX-RS 2.0: First taste of asynchrony

JAX-RS 2.0 saw the writing on the wall and added support for asynchronous processing: you could now tell the container to suspend the request until you were ready to resume it. You could then run your heavy computation in another thread until you were ready to resume it without blocking the resource method threads:

@Path("/hello")
public class HelloResource {


    @Path("suspended")
    @GET
    public void suspended(@Suspended AsyncResponse response) {
        new Thread(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            response.resume("Hello World (resumed)");
        }).start();
    }
}

Naturally this is a bit contrived, with sleep as an example of a heavy task, and creating a new unmanaged thread, but in reality this allowed you to launch your IO requests in an IO pool that would be more efficient and would not block the resource method threads, and once you were done with your IO you could then resume it:

@Path("/service")
public class ServiceResource {

//...

    @Path("suspended")
    @GET
    public void suspended(@Suspended AsyncResponse response) {
        Client client = ClientBuilder.newClient();
        URI uri = getUri(HelloResource.class, "suspended");
        client.target(uri).request().async().get(new InvocationCallback() {


            @Override
            public void completed(String entity) {
                response.resume("Service got: "+entity);
            }


            @Override
            public void failed(Throwable throwable) {
                response.resume(throwable);
            }
        });
        response.register((CompletionCallback)t -> client.close());
    }
}

In this last example, we used the asynchronous support of the JAX-RS client to offload the IO processing to an IO worker, and notify us when we have a result to send back to our client by resuming our suspended request. Notice how everything got more complex than the original blocking example? We had to suspend the request, register listeners for success and failure, and a custom finally block.

JAX-RS 2.1: Getting Reactive with CompletionStage

JAX-RS 2.1 shipped with awesome features, such as support for the Promise-like JDK CompletionStage for dealing with asynchronous code without the associated Callback Hell.

It became simpler to write asynchronous code because you didn't have to manually suspend the request: your return type of CompletionStage indicated that the request should be suspended until that CompletionStage was completed (successfully or not):

@Path("/hello")
public class HelloResource {

    @Path("completion-stage")
    @GET
    public CompletionStage completionStage() {
        CompletableFuture future = new CompletableFuture<>();
        new Thread(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                future.completeExceptionally(e);
                return;
            }
            future.complete("Hello World (resumed from CompletionStage)");
        }).start();
        return future;
    }
}

In this (yet again) contrived example, you can see we no longer had to use the @Suspended annotation or deal with the request in any way: we simply returned a CompletableFuture (which implements CompletionStage), and the container is responsible for registering a listener on it. Meanwhile, when we're ready to send our return value, we complete the CompletableFuture and that will resume the request.

Naturally, again in this case the benefits shine more when it comes to composition, because JAX-RS 2.1 also provides a Reactive REST Client interface via the rx() method, which allows us to get a CompletionStage for the client request, which becomes much easier to compose and return now:

@Path("/service")
public class ServiceResource {

  // ...

    @Path("completion-stage")
    @GET
    public CompletionStage completionStage() {
        Client client = ClientBuilder.newClient();
        URI uri = getUri(HelloResource.class, "completionStage");
        CompletionStage future = client.target(uri).request().rx().get(String.class);
        return future
                .thenApply(entity -> "Service got: "+entity)
                .whenComplete((entity, throwable) -> client.close());
    }
}

Here we really see the benefits of using reactive: we can provide a pipeline of work, on which we can register operations (thenApply) and even our finally block (whenComplete) that we just pass along to the container without doing any manual plumbing.

New in Resteasy: pluggable support for reactive libraries

It may have sounded like we were at the end of the adventure and had solved every issue, but in reality CompletionStage is great, but not that rich: libraries such as Reactive Streams and RxJava 1 and 2 provide types that are much richer than CompletionStage, which allow you to compose many more operations in an easier manner.

JAX-RS 2.1 added support for pluggable reactive libraries in the Reactive REST Client, via the rx(RxInvoker) method, but did not ship with any default implementation of it short of the default CompletionStage. But annoyingly it is not symmetrical because there is no support for pluggable resource method return types! You can make JAX-RS 2.1 support RxJava in the Reactive REST Client, but you cannot return RxJava values from your resource methods.

The latest Resteasy snapshots fix both issues:

• You can now plug in support for any reactive library via the AsyncResponseProvider interface ([RESTEASY-1701] Extend server-side async support / RxJava integration - JBoss Issue Tracker), and
• We now have the (optional) modules resteasy-rxjava1 and resteasy-rxjava2 which provide RxInvoker and AsyncResponseProvider implementations for RxJava types ([RESTEASY-1798] Client-side (proxy) support for RxClient and CompletionStage - JBoss Issue Tracker)

Those mean that you can now implement your hello resource using RxJava 2:

@Path("/hello")
public class HelloResource {
    
    @Path("rx")
    @GET
    public Single rx() {
        return Single.just("Hello World (resumed from rx)")
                .delay(1, TimeUnit.SECONDS);
    }
}

And our service similarly:

@Path("/service")
public class ServiceResource {

  // ...
    
    @Path("rx")
    @GET
    public Single rx() {
        Client client = ClientBuilder.newClient();
        URI uri = getUri(HelloResource.class, "rx");
        Single ret = client.target(uri).request().rx(SingleRxInvoker.class).get(String.class);
        return ret.map(entity -> "Service got: "+entity)
                .doFinally(() -> client.close());
    }   
}

As you can see, it's now much easier to build Reactive pipelines in Resteasy.

Note that although the pluggable support for reactive types landed in Resteasy 3.5.0.Final and 4.0.0.Beta2 releases, the RxInvoker implementations are only available in the latest 4.0.0-SNAPSHOT snapshots (they're likely being added to 3.6 branch soon, though).