JERSEY-2616: AsyncResponse and ChunkedOutput handlers called repeatedly

 - fixed resume of suspended continuation with complete call (which doesn't redispatch repeatedly)
 - reproducible test-case

Change-Id: Ica1988f79be6745092e7082c80b410e7d620ed94
Signed-off-by: Michal Gajdos <[email protected]>

Author: Michal Gajdos

containers/jetty-http/pom.xml

@@ -40,13 +40,16 @@
     holder.
 
 -->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+        xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
     <parent>
         <artifactId>project</artifactId>
         <groupId>org.glassfish.jersey.containers</groupId>
         <version>2.13-SNAPSHOT</version>
     </parent>
-    <modelVersion>4.0.0</modelVersion>
 
     <artifactId>jersey-container-jetty-http</artifactId>
     <packaging>jar</packaging>
@@ -64,12 +67,10 @@
             <groupId>org.eclipse.jetty</groupId>
             <artifactId>jetty-server</artifactId>
         </dependency>
-
         <dependency>
             <groupId>org.eclipse.jetty</groupId>
             <artifactId>jetty-continuation</artifactId>
         </dependency>
-
     </dependencies>
 
     <build>

containers/jetty-http/src/main/java/org/glassfish/jersey/jetty/JettyHttpContainer.java

@@ -101,11 +101,13 @@
  */
 public final class JettyHttpContainer extends AbstractHandler implements Container {
 
-    private static final ExtendedLogger logger =
+    private static final ExtendedLogger LOGGER =
             new ExtendedLogger(Logger.getLogger(JettyHttpContainer.class.getName()), Level.FINEST);
 
-    private static final Type RequestTYPE = (new TypeLiteral<Ref<Request>>() {}).getType();
-    private static final Type ResponseTYPE = (new TypeLiteral<Ref<Response>>() {}).getType();
+    private static final Type REQUEST_TYPE = (new TypeLiteral<Ref<Request>>() {}).getType();
+    private static final Type RESPONSE_TYPE = (new TypeLiteral<Ref<Response>>() {}).getType();
+
+    private static final int INTERNAL_SERVER_ERROR = javax.ws.rs.core.Response.Status.INTERNAL_SERVER_ERROR.getStatusCode();
 
     /**
      * Cached value of configuration property
@@ -161,15 +163,17 @@ protected void configure() {
     private volatile ContainerLifecycleListener containerListener;
 
     @Override
-    public void handle(final String target, final Request request, final HttpServletRequest httpServletRequest, final HttpServletResponse httpServletResponse) throws IOException, ServletException {
+    public void handle(final String target, final Request request, final HttpServletRequest httpServletRequest,
+                       final HttpServletResponse httpServletResponse) throws IOException, ServletException {
+
         final Response response = Response.getResponse(httpServletResponse);
         final ResponseWriter responseWriter = new ResponseWriter(request, response, configSetStatusOverSendError);
         final URI baseUri = getBaseUri(request);
 
         final String originalQuery = request.getUri().getQuery();
         final String encodedQuery = ContainerUtils.encodeUnsafeCharacters(originalQuery);
-        final String uriString = (originalQuery == null || originalQuery.isEmpty() || originalQuery.equals(encodedQuery)) ?
-                request.getUri().toString() : request.getUri().toString().replace(originalQuery, encodedQuery);
+        final String uriString = (originalQuery == null || originalQuery.isEmpty() || originalQuery.equals(encodedQuery))
+                ? request.getUri().toString() : request.getUri().toString().replace(originalQuery, encodedQuery);
         final URI requestUri = baseUri.resolve(uriString);
         try {
             final ContainerRequest requestContext = new ContainerRequest(
@@ -188,8 +192,8 @@ public void handle(final String target, final Request request, final HttpServlet
             requestContext.setRequestScopedInitializer(new RequestScopedInitializer() {
                 @Override
                 public void initialize(final ServiceLocator locator) {
-                    locator.<Ref<Request>>getService(RequestTYPE).set(request);
-                    locator.<Ref<Response>>getService(ResponseTYPE).set(response);
+                    locator.<Ref<Request>>getService(REQUEST_TYPE).set(request);
+                    locator.<Ref<Response>>getService(RESPONSE_TYPE).set(response);
                 }
             });
 
@@ -250,6 +254,7 @@ private String getBasePath(final Request request) {
     }
 
     private static final class ResponseWriter implements ContainerResponseWriter {
+
         private final Response response;
         private final Continuation continuation;
         private final boolean configSetStatusOverSendError;
@@ -261,11 +266,14 @@ private static final class ResponseWriter implements ContainerResponseWriter {
         }
 
         @Override
-        public OutputStream writeResponseStatusAndHeaders(final long contentLength, final ContainerResponse context) throws ContainerException {
+        public OutputStream writeResponseStatusAndHeaders(final long contentLength, final ContainerResponse context)
+                throws ContainerException {
+
             final javax.ws.rs.core.Response.StatusType statusInfo = context.getStatusInfo();
 
             final int code = statusInfo.getStatusCode();
-            final String reason = statusInfo.getReasonPhrase() == null ? HttpStatus.getMessage(code) : statusInfo.getReasonPhrase();
+            final String reason = statusInfo.getReasonPhrase() == null
+                    ? HttpStatus.getMessage(code) : statusInfo.getReasonPhrase();
 
             response.setStatusWithReason(code, reason);
 
@@ -304,7 +312,7 @@ public void onTimeout(final Continuation continuation) {
                         }
                     }
                 });
-                continuation.suspend();
+                continuation.suspend(response);
                 return true;
             } catch (final Exception ex) {
                 return false;
@@ -322,14 +330,14 @@ public void setSuspendTimeout(final long timeOut, final TimeUnit timeUnit) throw
         @Override
         public void commit() {
             try {
-                if (continuation.isSuspended()) {
-                    continuation.resume();
-                }
                 response.closeOutput();
             } catch (final IOException e) {
-                logger.log(Level.WARNING, LocalizationMessages.UNABLE_TO_CLOSE_RESPONSE(), e);
+                LOGGER.log(Level.WARNING, LocalizationMessages.UNABLE_TO_CLOSE_RESPONSE(), e);
             } finally {
-                logger.log(Level.FINEST, "commit() called");
+                if (continuation.isSuspended()) {
+                    continuation.complete();
+                }
+                LOGGER.log(Level.FINEST, "commit() called");
             }
         }
 
@@ -340,21 +348,21 @@ public void failure(final Throwable error) {
                     try {
                         if (configSetStatusOverSendError) {
                             response.reset();
-                            response.setStatus(500, "Request failed.");
+                            //noinspection deprecation
+                            response.setStatus(INTERNAL_SERVER_ERROR, "Request failed.");
                         } else {
-                            response.sendError(500, "Request failed.");
+                            response.sendError(INTERNAL_SERVER_ERROR, "Request failed.");
                         }
                     } catch (final IllegalStateException ex) {
                         // a race condition externally committing the response can still occur...
-                        logger.log(Level.FINER, "Unable to reset failed response.", ex);
+                        LOGGER.log(Level.FINER, "Unable to reset failed response.", ex);
                     } catch (final IOException ex) {
-                        throw new ContainerException(
-                                LocalizationMessages.EXCEPTION_SENDING_ERROR_RESPONSE(500, "Request failed."),
-                                ex);
+                        throw new ContainerException(LocalizationMessages.EXCEPTION_SENDING_ERROR_RESPONSE(INTERNAL_SERVER_ERROR,
+                                "Request failed."), ex);
                     }
                 }
             } finally {
-                logger.log(Level.FINEST, "failure(...) called");
+                LOGGER.log(Level.FINEST, "failure(...) called");
                 commit();
                 rethrow(error);
             }
@@ -365,9 +373,8 @@ public boolean enableResponseBuffering() {
             return false;
         }
 
-
         /**
-         * Rethrow the original exception as required by JAX-RS, 3.3.4
+         * Rethrow the original exception as required by JAX-RS, 3.3.4.
          *
          * @param error throwable to be re-thrown
          */
@@ -381,7 +388,6 @@ private void rethrow(final Throwable error) {
 
     }
 
-
     @Override
     public ResourceConfig getConfiguration() {
         return appHandler.getConfiguration();
@@ -409,8 +415,9 @@ public ApplicationHandler getApplicationHandler() {
 
     /**
      * Inform this container that the server has been started.
-     *
      * This method must be implicitly called after the server containing this container is started.
+     *
+     * @throws java.lang.Exception if a problem occurred during server startup.
      */
     @Override
     protected void doStart() throws Exception {
@@ -420,8 +427,9 @@ protected void doStart() throws Exception {
 
     /**
      * Inform this container that the server is being stopped.
-     *
      * This method must be implicitly called before the server containing this container is stopped.
+     *
+     * @throws java.lang.Exception if a problem occurred during server shutdown.
      */
     @Override
     public void doStop() throws Exception {

containers/jetty-http/src/test/java/org/glassfish/jersey/jetty/AsyncTest.java

@@ -1,7 +1,7 @@
 /*
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
  *
- * Copyright (c) 2013 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2013-2014 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
@@ -39,7 +39,11 @@
  */
 package org.glassfish.jersey.jetty;
 
-import org.junit.Test;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import javax.ws.rs.GET;
 import javax.ws.rs.Path;
@@ -49,32 +53,41 @@
 import javax.ws.rs.container.Suspended;
 import javax.ws.rs.container.TimeoutHandler;
 import javax.ws.rs.core.Response;
-import java.util.concurrent.*;
 
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
 
 /**
  * @author Arul Dhesiaseelan (aruld at acm.org)
+ * @author Michal Gajdos (michal.gajdos at oracle.com)
  */
 public class AsyncTest extends AbstractJettyServerTester {
 
     @Path("/async")
+    @SuppressWarnings("VoidMethodAnnotatedWithGET")
     public static class AsyncResource {
+
+        public static AtomicInteger INVOCATION_COUNT = new AtomicInteger(0);
+
         @GET
         public void asyncGet(@Suspended final AsyncResponse asyncResponse) {
             new Thread(new Runnable() {
 
                 @Override
                 public void run() {
-                    String result = veryExpensiveOperation();
+                    final String result = veryExpensiveOperation();
                     asyncResponse.resume(result);
                 }
 
                 private String veryExpensiveOperation() {
                     // ... very expensive operation that typically finishes within 5 seconds, simulated using sleep()
                     try {
                         Thread.sleep(5000);
-                    } catch (InterruptedException e) {
+                    } catch (final InterruptedException e) {
                         // ignore
                     }
                     return "DONE";
@@ -88,9 +101,9 @@ public void asyncGetWithTimeout(@Suspended final AsyncResponse asyncResponse) {
             asyncResponse.setTimeoutHandler(new TimeoutHandler() {
 
                 @Override
-                public void handleTimeout(AsyncResponse asyncResponse) {
-                    asyncResponse.resume(Response.status(Response.Status.SERVICE_UNAVAILABLE)
-                            .entity("Operation time out.").build());
+                public void handleTimeout(final AsyncResponse asyncResponse) {
+                    asyncResponse.resume(Response.status(Response.Status.SERVICE_UNAVAILABLE).entity("Operation time out.")
+                            .build());
                 }
             });
             asyncResponse.setTimeout(3, TimeUnit.SECONDS);
@@ -99,28 +112,53 @@ public void handleTimeout(AsyncResponse asyncResponse) {
 
                 @Override
                 public void run() {
-                    String result = veryExpensiveOperation();
+                    final String result = veryExpensiveOperation();
                     asyncResponse.resume(result);
                 }
 
                 private String veryExpensiveOperation() {
                     // ... very expensive operation that typically finishes within 10 seconds, simulated using sleep()
                     try {
                         Thread.sleep(7000);
-                    } catch (InterruptedException e) {
+                    } catch (final InterruptedException e) {
                         // ignore
                     }
                     return "DONE";
                 }
             }).start();
         }
 
+        @GET
+        @Path("multiple-invocations")
+        public void asyncMultipleInvocations(@Suspended final AsyncResponse asyncResponse) {
+            INVOCATION_COUNT.incrementAndGet();
+
+            new Thread(new Runnable() {
+                @Override
+                public void run() {
+                    asyncResponse.resume("OK");
+                }
+            }).start();
+        }
+    }
+
+    private Client client;
+
+    @Before
+    public void setUp() throws Exception {
+        startServer(AsyncResource.class);
+        client = ClientBuilder.newClient();
+    }
+
+    @Override
+    @After
+    public void tearDown() {
+        super.tearDown();
+        client = null;
     }
 
     @Test
     public void testAsyncGet() throws ExecutionException, InterruptedException {
-        startServer(AsyncResource.class);
-        Client client = ClientBuilder.newClient();
         final Future<Response> responseFuture = client.target(getUri().path("/async")).request().async().get();
         // Request is being processed asynchronously.
         final Response response = responseFuture.get();
@@ -130,8 +168,6 @@ public void testAsyncGet() throws ExecutionException, InterruptedException {
 
     @Test
     public void testAsyncGetWithTimeout() throws ExecutionException, InterruptedException, TimeoutException {
-        startServer(AsyncResource.class);
-        Client client = ClientBuilder.newClient();
         final Future<Response> responseFuture = client.target(getUri().path("/async/timeout")).request().async().get();
         // Request is being processed asynchronously.
         final Response response = responseFuture.get();
@@ -141,4 +177,16 @@ public void testAsyncGetWithTimeout() throws ExecutionException, InterruptedExce
         assertEquals("Operation time out.", response.readEntity(String.class));
     }
 
+    /**
+     * JERSEY-2616 reproducer. Make sure resource method is only invoked once per one request.
+     */
+    @Test
+    public void testAsyncMultipleInvocations() throws Exception {
+        final Response response = client.target(getUri().path("/async/multiple-invocations")).request().get();
+
+        assertThat(AsyncResource.INVOCATION_COUNT.get(), is(1));
+
+        assertThat(response.getStatus(), is(200));
+        assertThat(response.readEntity(String.class), is("OK"));
+    }
 }