2.21. 扩展 RESTEasy 支持异步请求处理和 Reactive Return 类型

public interface AsyncResponseProvider<T> {
   public CompletionStage toCompletionStage(T asyncResponse);
}

public interface AsyncStreamProvider<T> {
   public Publisher toAsyncStream(T asyncResponse);
}

public interface RxInvokerProvider<T extends RxInvoker> {
    public boolean isProviderFor(Class<T> clazz);
    public T getRxInvoker(SyncInvoker syncInvoker, ExecutorService executorService);
}

public <T extends RxInvoker> T rx(Class<T> clazz);

FlowableRxInvoker invoker = client.target(generateURL("/get/string")).request().rx(FlowableRxInvoker.class);
Flowable<String> flowable = (Flowable<String>) invoker.get();

public interface AsyncClientResponseProvider<T> {
   public T fromCompletionStage(CompletionStage<?> completionStage);
}

@Path("")
public interface RxCompletionStageResource {

   @GET
   @Path("get/string")
   @Produces(MediaType.TEXT_PLAIN)
   public CompletionStage<String> getString();
}

@Path("")
public class RxCompletionStageResourceImpl {

   @GET
   @Path("get/string")
   @Produces(MediaType.TEXT_PLAIN)
   public CompletionStage<String> getString() { .... }
}

public class RxCompletionStageProxyTest {

   private static ResteasyClient client;
   private static RxCompletionStageResource proxy;

   static {
      client = new ResteasyClientBuilder().build();
      proxy = client.target(generateURL("/")).proxy(RxCompletionStageResource.class);
   }

   @Test
   public void testGet() throws Exception {
      CompletionStage<String> completionStage = proxy.getString();
      Assert.assertEquals("x", completionStage.toCompletableFuture().get());
   }
}

public interface Rx2FlowableResource {

   @GET
   @Path("get/string")
   @Produces(MediaType.TEXT_PLAIN)
   @Stream
   public Flowable<String> getFlowable();
}

@Path("")
public class Rx2FlowableResourceImpl {

   @GET
   @Path("get/string")
   @Produces(MediaType.TEXT_PLAIN)
   @Stream
   public Flowable<String> getFlowable() { ... }
}

public class Rx2FlowableProxyTest {

   private static ResteasyClient client;
   private static Rx2FlowableResource proxy;

   static {
      client = new ResteasyClientBuilder().build();
      proxy = client.target(generateURL("/")).proxy(Rx2FlowableResource.class);
   }

   @Test
   public void testGet() throws Exception {
      Flowable<String> flowable = proxy.getFlowable();
      flowable.subscribe(
         (String o) -> stringList.add(o),
         (Throwable t) -> errors.incrementAndGet(),
         () -> latch.countDown());
      boolean waitResult = latch.await(30, TimeUnit.SECONDS);
      Assert.assertTrue("Waiting for event to be delivered has timed out.", waitResult);
      Assert.assertEquals(0, errors.get());
      Assert.assertEquals(xStringList, stringList);
   }
}