JerseyCompletionStageRxInvoker should use AsyncInvoker

[olotenko]

Expected: configure asynchronous thread pool the size of CPU count, be able to saturate CPU with long-running requests to other services.

Actual: CPU mostly idle. RxInvoker uses SyncInvoker, and Jersey Client does the straightforward thing, which consumes the threads waiting for response.

If JAX-RS API issue 703 is addressed at the API level, so much the better. If that issue is not fixed, at least we probably can assume that AbstractRxInvoker.getSyncInvoker returns the same SyncInvoker that JerseyCompletionStageRxInvoker passed to it in the constructor.

A prototype fix with such type cast demonstrates the thread utilisation can become as Expected.

[jansupol]

If this is solved on the API level, it would possibly be done by a new method, and I am not sure what version of the API it should be in. So currently, it looks like this ought to be solved on the Jersey level.

[olotenko]

--- a/core-client/src/main/java/org/glassfish/jersey/client/JerseyCompletionStageRxInvoker.java
+++ b/core-client/src/main/java/org/glassfish/jersey/client/JerseyCompletionStageRxInvoker.java
@@ -39,19 +39,17 @@ public class JerseyCompletionStageRxInvoker extends AbstractRxInvoker<Completion
 
     @Override
     public <T> CompletionStage<T> method(final String name, final Entity<?> entity, final Class<T> responseType) {
-        final ExecutorService executorService = getExecutorService();
-
-        return executorService == null
-                ? CompletableFuture.supplyAsync(() -> getSyncInvoker().method(name, entity, responseType))
-                : CompletableFuture.supplyAsync(() -> getSyncInvoker().method(name, entity, responseType), executorService);
+        // the two type casts are necessary evil:
+        // - the casting of SyncInvoker is the only way to get to AsyncInvoker -
+        //   the API for RxInvoker is unnecessarily retricted to receivin a SyncInvoker, which makes no sense for RxInvoker;
+        // - the casting of the AsyncInvoker's return value is the only way to get the same behaviour as AsyncInvoker has
+        //   with respect to decoding the response and throwing Exceptions. Without this it would be necessary to essentially
+        //   duplicate the code of the private methods that AsyncInvoker appears to have access to.
+        return (CompletionStage<T>) ((Invocation.Builder) getSyncInvoker()).async().method(name, entity, responseType);
     }
 
     @Override
     public <T> CompletionStage<T> method(final String name, final Entity<?> entity, final GenericType<T> responseType) {
-        final ExecutorService executorService = getExecutorService();
-
-        return executorService == null
-                ? CompletableFuture.supplyAsync(() -> getSyncInvoker().method(name, entity, responseType))
-                : CompletableFuture.supplyAsync(() -> getSyncInvoker().method(name, entity, responseType), executorService);
+        return (CompletionStage<T>) ((Invocation.Builder) getSyncInvoker()).async().method(name, entity, responseType);
     }
 }

is possibly the simplest way to deal with it: basically AsyncInvoker completely fits the RxInvoker<CompletionStage> interface, except for the presence of generics (like, public <R> T get(Class<R>) in RxInvoker). So Jersey could reuse AsyncInvoker, if such class casts are not too ugly.

Alternatively,

--- a/core-client/src/main/java/org/glassfish/jersey/client/JerseyInvocation.java
+++ b/core-client/src/main/java/org/glassfish/jersey/client/JerseyInvocation.java
@@ -513,7 +514,7 @@ public class JerseyInvocation implements javax.ws.rs.client.Invocation {
         }
     }
 
-    private static class AsyncInvoker implements javax.ws.rs.client.AsyncInvoker {
+    private static class AsyncInvoker implements javax.ws.rs.client.AsyncInvoker, javax.ws.rs.client.RxInvoker<CompletionStage> {

(plus replace all Future<T> with CompletableFuture<T>), and now AsyncInvoker can be used either as AsyncInvoker, or as a RxInvoker - no need for the special class.

[olotenko]

rx_invoker.zip
rx_invoker.66f5cfa1.diff - class cast version
rx_invoker.998362ad.diff - AsyncInvoker implements CompletionStageRxInvoker. It's a private class, so the actual return types are explicitly visible only in the class where it is declared. It looks like JerseyCompletionStageRxInvoker can be deleted in this case, but maybe someone uses it directly.

[jansupol]

How about #4283 ?
You can use rx(JerseyCompletionStageAsyncRxInvoker.class) which would give you what you actually want.

[olotenko]

To me using SyncInvoker in Rx makes no sense ever. It defeats the purpose of being reactive. Rx is the Async with the better API.

So I am looking for changing what rx() uses out of the box.

[jansupol]

While the API is designed to work with SyncInvoker, I agree that AsyncInvoker makes more sense.

I have provided a PR #4283 with AsyncInvoker as the default for rx(). Note that RxInvoker does not support methods with InvocationCallback from AsyncInvoker. If those would be needed, it is possible to use rx(JerseyCompletionStageRxInvoker.class) that makes them accessible without a cast.