CompletionStage returned from the resource methods might not support #toCompletableFuture.

Change-Id: I30b4e57c44979df35a643a214b14e8572a84b517

Author: pavelbucek

core-client/src/main/java/org/glassfish/jersey/client/JerseyCompletionStageRxInvoker.java

@@ -57,7 +57,7 @@
  */
 public class JerseyCompletionStageRxInvoker extends AbstractRxInvoker<CompletionStage> implements CompletionStageRxInvoker {
 
-    public JerseyCompletionStageRxInvoker(Invocation.Builder builder, ExecutorService executor) {
+    JerseyCompletionStageRxInvoker(Invocation.Builder builder, ExecutorService executor) {
         super(builder, executor);
     }
 

core-server/src/main/java/org/glassfish/jersey/server/model/ResourceMethodInvoker.java

@@ -53,7 +53,6 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CancellationException;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.function.BiConsumer;
 import java.util.function.Supplier;
@@ -418,15 +417,15 @@ public ContainerResponse apply(final RequestProcessingContext processingContext)
             if (response.hasEntity()) {
                 Object entityFuture = response.getEntity();
                 if (entityFuture instanceof CompletionStage) {
-                    CompletableFuture completableFuture = ((CompletionStage) entityFuture).toCompletableFuture();
+                    CompletionStage completionStage = ((CompletionStage) entityFuture);
 
                     // suspend - we know that this feature is not done, see AbstractJavaResourceMethodDispatcher#invoke
                     if (!processingContext.asyncContext().suspend()) {
                         throw new ProcessingException(LocalizationMessages.ERROR_SUSPENDING_ASYNC_REQUEST());
                     }
 
                     // wait for a response
-                    completableFuture.whenComplete(whenComplete(processingContext));
+                    completionStage.whenComplete(whenComplete(processingContext));
 
                     return null; // return null on the current thread
                 }

core-server/src/main/java/org/glassfish/jersey/server/model/internal/AbstractJavaResourceMethodDispatcher.java

@@ -148,9 +148,16 @@ public Object run() {
 
                         // if a response is a CompletionStage and is done, we don't need to suspend and resume
                         if (result instanceof CompletionStage) {
-                            CompletableFuture resultFuture = ((CompletionStage) result).toCompletableFuture();
+                            CompletableFuture resultFuture;
+                            try {
+                                resultFuture = ((CompletionStage) result).toCompletableFuture();
+                            } catch (UnsupportedOperationException e) {
+                                // CompletionStage is not required to implement "toCompletableFuture". If it doesn't
+                                // we treat it as "uncompleted" future.
+                                return result;
+                            }
 
-                            if (resultFuture.isDone()) {
+                            if (resultFuture != null && resultFuture.isDone()) {
                                 if (resultFuture.isCancelled()) {
                                     return Response.status(Response.Status.SERVICE_UNAVAILABLE).build();
                                 } else {

tests/e2e-server/src/test/java/org/glassfish/jersey/tests/e2e/server/CompletionStageTest.java

@@ -42,8 +42,13 @@
 
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+import java.util.function.Function;
 
 import javax.ws.rs.GET;
 import javax.ws.rs.Path;
@@ -130,6 +135,22 @@ public void testGetCancelledAsync() {
         assertThat(response.getStatus(), is(503));
     }
 
+    @Test
+    public void testGetCustomCompleted() {
+        Response response = target("cs/custom").request().get();
+
+        assertThat(response.getStatus(), is(200));
+        assertThat(response.readEntity(String.class), is(ENTITY));
+    }
+
+    @Test
+    public void testGetCustomAsync() {
+        Response response = target("cs/customAsync").request().get();
+
+        assertThat(response.getStatus(), is(200));
+        assertThat(response.readEntity(String.class), is(ENTITY));
+    }
+
     @Path("/cs")
     public static class CompletionStageResource {
 
@@ -203,6 +224,32 @@ public CompletionStage<String> getCancelledAsync() {
             return cs;
         }
 
+        /**
+         * Return completed CompletionStage which doesn't support #toCompletableFuture().
+         *
+         * @return CompletionStage which doesn't support #toCompletableFuture().
+         */
+        @GET
+        @Path("/custom")
+        public CompletionStage<String> getCustomCompletionStage() {
+            return new CustomCompletionStage<>(CompletableFuture.completedFuture(ENTITY));
+        }
+
+        /**
+         * Return uncompleted CompletionStage which doesn't support #toCompletableFuture().
+         *
+         * @return CompletionStage which doesn't support #toCompletableFuture().
+         */
+        @GET
+        @Path("/customAsync")
+        public CompletionStage<String> getCustomCompletionStageAsync() {
+            CompletableFuture<String> cf = new CompletableFuture<>();
+            CustomCompletionStage<String> cs = new CustomCompletionStage<>(cf);
+            delaySubmit(() -> cf.complete(ENTITY));
+
+            return cs;
+        }
+
         private void delaySubmit(Runnable runnable) {
             EXECUTOR_SERVICE.submit(() -> {
                 try {
@@ -215,4 +262,214 @@ private void delaySubmit(Runnable runnable) {
             });
         }
     }
+
+    private static class CustomCompletionStage<T> implements CompletionStage<T> {
+
+        private final CompletionStage<T> completedFuture;
+
+        CustomCompletionStage(CompletionStage<T> completedFuture) {
+            this.completedFuture = completedFuture;
+        }
+
+        @Override
+        public <U> CompletionStage<U> thenApply(Function<? super T, ? extends U> fn) {
+            return completedFuture.thenApply(fn);
+        }
+
+        @Override
+        public <U> CompletionStage<U> thenApplyAsync(Function<? super T, ? extends U> fn) {
+            return completedFuture.thenApplyAsync(fn);
+        }
+
+        @Override
+        public <U> CompletionStage<U> thenApplyAsync(Function<? super T, ? extends U> fn, Executor executor) {
+            return completedFuture.thenApplyAsync(fn, executor);
+        }
+
+        @Override
+        public CompletionStage<Void> thenAccept(Consumer<? super T> action) {
+            return completedFuture.thenAccept(action);
+        }
+
+        @Override
+        public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action) {
+            return completedFuture.thenAcceptAsync(action);
+        }
+
+        @Override
+        public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor) {
+            return completedFuture.thenAcceptAsync(action, executor);
+        }
+
+        @Override
+        public CompletionStage<Void> thenRun(Runnable action) {
+            return completedFuture.thenRun(action);
+        }
+
+        @Override
+        public CompletionStage<Void> thenRunAsync(Runnable action) {
+            return completedFuture.thenRunAsync(action);
+        }
+
+        @Override
+        public CompletionStage<Void> thenRunAsync(Runnable action, Executor executor) {
+            return completedFuture.thenRunAsync(action, executor);
+        }
+
+        @Override
+        public <U, V> CompletionStage<V> thenCombine(CompletionStage<? extends U> other,
+                                                     BiFunction<? super T, ? super U, ? extends V> fn) {
+            return completedFuture.thenCombine(other, fn);
+        }
+
+        @Override
+        public <U, V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,
+                                                          BiFunction<? super T, ? super U, ? extends V> fn) {
+            return completedFuture.thenCombineAsync(other, fn);
+        }
+
+        @Override
+        public <U, V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,
+                                                          BiFunction<? super T, ? super U, ? extends V> fn, Executor executor) {
+            return completedFuture.thenCombineAsync(other, fn, executor);
+        }
+
+        @Override
+        public <U> CompletionStage<Void> thenAcceptBoth(CompletionStage<? extends U> other,
+                                                        BiConsumer<? super T, ? super U> action) {
+            return completedFuture.thenAcceptBoth(other, action);
+        }
+
+        @Override
+        public <U> CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,
+                                                             BiConsumer<? super T, ? super U> action) {
+            return completedFuture.thenAcceptBothAsync(other, action);
+        }
+
+        @Override
+        public <U> CompletionStage<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,
+                                                             BiConsumer<? super T, ? super U> action, Executor executor) {
+            return completedFuture.thenAcceptBothAsync(other, action, executor);
+        }
+
+        @Override
+        public CompletionStage<Void> runAfterBoth(CompletionStage<?> other, Runnable action) {
+            return completedFuture.runAfterBoth(other, action);
+        }
+
+        @Override
+        public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action) {
+            return completedFuture.runAfterBothAsync(other, action);
+        }
+
+        @Override
+        public CompletionStage<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action, Executor executor) {
+            return completedFuture.runAfterBothAsync(other, action, executor);
+        }
+
+        @Override
+        public <U> CompletionStage<U> applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn) {
+            return completedFuture.applyToEither(other, fn);
+        }
+
+        @Override
+        public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn) {
+            return completedFuture.applyToEitherAsync(other, fn);
+        }
+
+        @Override
+        public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other,
+                                                         Function<? super T, U> fn,
+                                                         Executor executor) {
+            return completedFuture.applyToEitherAsync(other, fn, executor);
+        }
+
+        @Override
+        public CompletionStage<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action) {
+            return completedFuture.acceptEither(other, action);
+        }
+
+        @Override
+        public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action) {
+            return completedFuture.acceptEitherAsync(other, action);
+        }
+
+        @Override
+        public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other,
+                                                       Consumer<? super T> action,
+                                                       Executor executor) {
+            return completedFuture.acceptEitherAsync(other, action, executor);
+        }
+
+        @Override
+        public CompletionStage<Void> runAfterEither(CompletionStage<?> other, Runnable action) {
+            return completedFuture.runAfterEither(other, action);
+        }
+
+        @Override
+        public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action) {
+            return completedFuture.runAfterEitherAsync(other, action);
+        }
+
+        @Override
+        public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action, Executor executor) {
+            return completedFuture.runAfterEitherAsync(other, action, executor);
+        }
+
+        @Override
+        public <U> CompletionStage<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) {
+            return completedFuture.thenCompose(fn);
+        }
+
+        @Override
+        public <U> CompletionStage<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn) {
+            return completedFuture.thenComposeAsync(fn);
+        }
+
+        @Override
+        public <U> CompletionStage<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn,
+                                                       Executor executor) {
+            return completedFuture.thenComposeAsync(fn, executor);
+        }
+
+        @Override
+        public CompletionStage<T> exceptionally(Function<Throwable, ? extends T> fn) {
+            return completedFuture.exceptionally(fn);
+        }
+
+        @Override
+        public CompletionStage<T> whenComplete(BiConsumer<? super T, ? super Throwable> action) {
+            return completedFuture.whenComplete(action);
+        }
+
+        @Override
+        public CompletionStage<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action) {
+            return completedFuture.whenCompleteAsync(action);
+        }
+
+        @Override
+        public CompletionStage<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor) {
+            return completedFuture.whenCompleteAsync(action, executor);
+        }
+
+        @Override
+        public <U> CompletionStage<U> handle(BiFunction<? super T, Throwable, ? extends U> fn) {
+            return completedFuture.handle(fn);
+        }
+
+        @Override
+        public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn) {
+            return completedFuture.handleAsync(fn);
+        }
+
+        @Override
+        public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) {
+            return completedFuture.handleAsync(fn, executor);
+        }
+
+        @Override
+        public CompletableFuture<T> toCompletableFuture() {
+            throw new UnsupportedOperationException();
+        }
+    }
 }