Support for returning CompletionStage from a resource method.

pavelbucek · pavelbucek · commit 65aa8f16bf54 · 2017-05-18T13:39:50.000+02:00
Change-Id: I6c8ed84cce803559bd4cfe40da411d8784671380
diff --git a/core-server/src/main/java/org/glassfish/jersey/server/ServerRuntime.java b/core-server/src/main/java/org/glassfish/jersey/server/ServerRuntime.java
@@ -854,6 +854,11 @@ public void run() {
             });
         }
 
+        @Override
+        public void invokeManaged(Runnable runnable) {
+            responder.runtime.managedAsyncExecutor.get().submit(runnable);
+        }
+
         @Override
         public boolean suspend() {
             synchronized (stateLock) {
diff --git a/core-server/src/main/java/org/glassfish/jersey/server/internal/process/AsyncContext.java b/core-server/src/main/java/org/glassfish/jersey/server/internal/process/AsyncContext.java
@@ -1,7 +1,7 @@
 /*
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
  *
- * Copyright (c) 2012-2014 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2012-2017 Oracle and/or its affiliates. All rights reserved.
  *
  * The contents of this file are subject to the terms of either the GNU
  * General Public License Version 2 only ("GPL") or the Common Development
@@ -97,4 +97,11 @@ public static enum State {
      * @param producer response producer.
      */
     public void invokeManaged(Producer<Response> producer);
+
+    /**
+     * Invoke the provided runnable in a Jersey-managed asynchronous thread.
+     *
+     * @param runnable to be invoked.
+     */
+    public void invokeManaged(Runnable runnable);
 }
diff --git a/core-server/src/main/java/org/glassfish/jersey/server/model/ResourceMethodInvoker.java b/core-server/src/main/java/org/glassfish/jersey/server/model/ResourceMethodInvoker.java
@@ -52,6 +52,10 @@
 import java.util.List;
 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;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
@@ -114,7 +118,7 @@ public class ResourceMethodInvoker implements Endpoint, ResourceInfo {
 
     /**
      * Resource method invoker helper.
-     *
+     * <p>
      * The builder API provides means for constructing a properly initialized
      * {@link ResourceMethodInvoker resource method invoker} instances.
      */
@@ -409,10 +413,46 @@ public ContainerResponse apply(final RequestProcessingContext processingContext)
             return null; // return null on current thread
         } else {
             // TODO replace with processing context factory method.
-            return new ContainerResponse(request, invoke(processingContext, resource));
+            Response response = invoke(processingContext, resource);
+
+            if (response.hasEntity()) {
+                Object entityFuture = response.getEntity();
+                if (entityFuture instanceof CompletionStage) {
+                    CompletableFuture completableFuture = ((CompletionStage) entityFuture).toCompletableFuture();
+
+                    // 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.whenCompleteAsync(
+                            whenComplete(processingContext),
+                            command -> processingContext.asyncContext().invokeManaged(command));
+
+                    return null; // return null on the current thread
+                }
+            }
+
+            return new ContainerResponse(request, response);
         }
     }
 
+    private BiConsumer whenComplete(RequestProcessingContext processingContext) {
+        return (entity, exception) -> {
+
+            if (exception != null) {
+                if (exception instanceof CancellationException) {
+                    processingContext.asyncContext().resume(Response.status(Response.Status.SERVICE_UNAVAILABLE).build());
+                } else {
+                    processingContext.asyncContext().resume(((Throwable) exception));
+                }
+            } else {
+                processingContext.asyncContext().resume(entity);
+            }
+        };
+    }
+
     private Response invoke(final RequestProcessingContext context, final Object resource) {
 
         Response jaxrsResponse;
diff --git a/core-server/src/main/java/org/glassfish/jersey/server/model/internal/AbstractJavaResourceMethodDispatcher.java b/core-server/src/main/java/org/glassfish/jersey/server/model/internal/AbstractJavaResourceMethodDispatcher.java
@@ -1,7 +1,7 @@
 /*
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
  *
- * Copyright (c) 2011-2015 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2011-2017 Oracle and/or its affiliates. All rights reserved.
  *
  * The contents of this file are subject to the terms of either the GNU
  * General Public License Version 2 only ("GPL") or the Common Development
@@ -44,6 +44,9 @@
 import java.lang.reflect.Method;
 import java.lang.reflect.UndeclaredThrowableException;
 import java.security.PrivilegedAction;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ExecutionException;
 
 import javax.ws.rs.ProcessingException;
 import javax.ws.rs.WebApplicationException;
@@ -141,7 +144,26 @@ public Object run() {
                     final long timestamp = tracingLogger.timestamp(ServerTraceEvent.METHOD_INVOKE);
                     try {
 
-                        return methodHandler.invoke(resource, method, args);
+                        Object result = methodHandler.invoke(resource, method, args);
+
+                        // 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();
+
+                            if (resultFuture.isDone()) {
+                                if (resultFuture.isCancelled()) {
+                                    return Response.status(Response.Status.SERVICE_UNAVAILABLE).build();
+                                } else {
+                                    try {
+                                        return resultFuture.get();
+                                    } catch (ExecutionException e) {
+                                        throw new InvocationTargetException(e.getCause());
+                                    }
+                                }
+                            }
+                        }
+
+                        return result;
 
                     } catch (IllegalAccessException | IllegalArgumentException | UndeclaredThrowableException ex) {
                         throw new ProcessingException(LocalizationMessages.ERROR_RESOURCE_JAVA_METHOD_INVOCATION(), ex);
diff --git a/core-server/src/main/java/org/glassfish/jersey/server/model/internal/JavaResourceMethodDispatcherProvider.java b/core-server/src/main/java/org/glassfish/jersey/server/model/internal/JavaResourceMethodDispatcherProvider.java
@@ -209,8 +209,6 @@ protected Response doDispatch(final Object resource, final ContainerRequest cont
 
             if (o instanceof Response) {
                 return Response.class.cast(o);
-//            } else if (o instanceof JResponse) {
-//                context.getResponseContext().setResponse(((JResponse)o).toResponse());
             } else if (o != null) {
                 return Response.ok().entity(o).build();
             } else {
@@ -236,12 +234,10 @@ private static final class TypeOutInvoker extends AbstractMethodParamInvoker {
         protected Response doDispatch(final Object resource, final ContainerRequest containerRequest) throws ProcessingException {
             final Object o = invoke(containerRequest, resource, getParamValues(containerRequest));
             if (o != null) {
-
-                Response response = Response.ok().entity(o).build();
-                // TODO set the method return Java type to the proper context.
-//                Response r = new ResponseBuilderImpl().
-//                        entityWithType(o, t).status(200).build();
-                return response;
+                if (o instanceof Response) {
+                    return Response.class.cast(o);
+                }
+                return Response.ok().entity(o).build();
             } else {
                 return Response.noContent().build();
             }
diff --git a/tests/e2e-server/src/test/java/org/glassfish/jersey/tests/e2e/server/CompletionStageTest.java b/tests/e2e-server/src/test/java/org/glassfish/jersey/tests/e2e/server/CompletionStageTest.java
@@ -0,0 +1,218 @@
+/*
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
+ *
+ * Copyright (c) 2017 Oracle and/or its affiliates. All rights reserved.
+ *
+ * The contents of this file are subject to the terms of either the GNU
+ * General Public License Version 2 only ("GPL") or the Common Development
+ * and Distribution License("CDDL") (collectively, the "License").  You
+ * may not use this file except in compliance with the License.  You can
+ * obtain a copy of the License at
+ * http://glassfish.java.net/public/CDDL+GPL_1_1.html
+ * or packager/legal/LICENSE.txt.  See the License for the specific
+ * language governing permissions and limitations under the License.
+ *
+ * When distributing the software, include this License Header Notice in each
+ * file and include the License file at packager/legal/LICENSE.txt.
+ *
+ * GPL Classpath Exception:
+ * Oracle designates this particular file as subject to the "Classpath"
+ * exception as provided by Oracle in the GPL Version 2 section of the License
+ * file that accompanied this code.
+ *
+ * Modifications:
+ * If applicable, add the following below the License Header, with the fields
+ * enclosed by brackets [] replaced by your own identifying information:
+ * "Portions Copyright [year] [name of copyright owner]"
+ *
+ * Contributor(s):
+ * If you wish your version of this file to be governed by only the CDDL or
+ * only the GPL Version 2, indicate your decision by adding "[Contributor]
+ * elects to include this software in this distribution under the [CDDL or GPL
+ * Version 2] license."  If you don't indicate a single choice of license, a
+ * recipient has the option to distribute your version of this file under
+ * either the CDDL, the GPL Version 2 or to extend the choice of license to
+ * its licensees as provided above.  However, if you add GPL Version 2 code
+ * and therefore, elected the GPL Version 2 license, then the option applies
+ * only if the new code is made subject to such option by the copyright
+ * holder.
+ */
+
+package org.glassfish.jersey.tests.e2e.server;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.Application;
+import javax.ws.rs.core.Response;
+
+import org.glassfish.jersey.server.ResourceConfig;
+import org.glassfish.jersey.test.JerseyTest;
+
+import org.junit.Test;
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+/**
+ * @author Pavel Bucek (pavel.bucek at oracle.com)
+ */
+public class CompletionStageTest extends JerseyTest {
+
+    static final String ENTITY = "entity";
+    // delay of async operations in seconds.
+    static final int DELAY = 1;
+
+    @Override
+    protected Application configure() {
+        return new ResourceConfig(CompletionStageResource.class);
+    }
+
+    @Test
+    public void testGetCompleted() {
+        Response response = target("cs/completed").request().get();
+
+        assertThat(response.getStatus(), is(200));
+        assertThat(response.readEntity(String.class), is(ENTITY));
+    }
+
+    @Test
+    public void testGetException400() {
+        Response response = target("cs/exception400").request().get();
+
+        assertThat(response.getStatus(), is(400));
+    }
+
+    @Test
+    public void testGetException405() {
+        Response response = target("cs/exception405").request().get();
+
+        assertThat(response.getStatus(), is(405));
+    }
+
+    @Test
+    public void testGetCancelled() {
+        Response response = target("cs/cancelled").request().get();
+
+        assertThat(response.getStatus(), is(503));
+    }
+
+    @Test
+    public void testGetCompletedAsync() {
+        Response response = target("cs/completedAsync").request().get();
+
+        assertThat(response.getStatus(), is(200));
+        assertThat(response.readEntity(String.class), is(ENTITY));
+    }
+
+    @Test
+    public void testGetException400Async() {
+        Response response = target("cs/exception400Async").request().get();
+
+        assertThat(response.getStatus(), is(400));
+    }
+
+    @Test
+    public void testGetException405Async() {
+        Response response = target("cs/exception405Async").request().get();
+
+        assertThat(response.getStatus(), is(405));
+    }
+
+    @Test
+    public void testGetCancelledAsync() {
+        Response response = target("cs/cancelledAsync").request().get();
+
+        assertThat(response.getStatus(), is(503));
+    }
+
+    @Path("/cs")
+    public static class CompletionStageResource {
+
+        private static final ExecutorService EXECUTOR_SERVICE = Executors.newCachedThreadPool();
+
+        @GET
+        @Path("/completed")
+        public CompletionStage<String> getCompleted() {
+            return CompletableFuture.completedFuture(ENTITY);
+        }
+
+        @GET
+        @Path("/exception400")
+        public CompletionStage<String> getException400() {
+            CompletableFuture<String> cs = new CompletableFuture<>();
+            cs.completeExceptionally(new WebApplicationException(400));
+
+            return cs;
+        }
+
+        @GET
+        @Path("/exception405")
+        public CompletionStage<String> getException405() {
+            CompletableFuture<String> cs = new CompletableFuture<>();
+            cs.completeExceptionally(new WebApplicationException(405));
+
+            return cs;
+        }
+
+        @GET
+        @Path("/cancelled")
+        public CompletionStage<String> getCancelled() {
+            CompletableFuture<String> cs = new CompletableFuture<>();
+            cs.cancel(true);
+
+            return cs;
+        }
+
+        @GET
+        @Path("/completedAsync")
+        public CompletionStage<String> getCompletedAsync() {
+            CompletableFuture<String> cs = new CompletableFuture<>();
+            delaySubmit(() -> cs.complete(ENTITY));
+            return cs;
+        }
+
+        @GET
+        @Path("/exception400Async")
+        public CompletionStage<String> getException400Async() {
+            CompletableFuture<String> cs = new CompletableFuture<>();
+            delaySubmit(() -> cs.completeExceptionally(new WebApplicationException(400)));
+
+            return cs;
+        }
+
+        @GET
+        @Path("/exception405Async")
+        public CompletionStage<String> getException405Async() {
+            CompletableFuture<String> cs = new CompletableFuture<>();
+            delaySubmit(() -> cs.completeExceptionally(new WebApplicationException(405)));
+
+            return cs;
+        }
+
+        @GET
+        @Path("/cancelledAsync")
+        public CompletionStage<String> getCancelledAsync() {
+            CompletableFuture<String> cs = new CompletableFuture<>();
+            delaySubmit(() -> cs.cancel(true));
+
+            return cs;
+        }
+
+        private void delaySubmit(Runnable runnable) {
+            EXECUTOR_SERVICE.submit(() -> {
+                try {
+                    Thread.sleep(DELAY * 1000);
+                } catch (InterruptedException e) {
+                    // ignore
+                }
+
+                runnable.run();
+            });
+        }
+    }
+}
