Properly close the Apache response so that connections can be reused

Author: alessandro.gherardi

connectors/apache-connector/src/main/java/org/glassfish/jersey/apache/connector/ApacheConnector.java

@@ -485,8 +485,9 @@ public ClientResponse apply(final ClientRequest clientRequest) throws Processing
                 }
             }
 
+            final boolean hasSocketTimeout = clientRequest.resolveProperty(ClientProperties.READ_TIMEOUT, -1) >= 0;
             try {
-                responseContext.setEntityStream(new HttpClientResponseInputStream(getInputStream(response)));
+                responseContext.setEntityStream(getInputStream(response, hasSocketTimeout));
             } catch (final IOException e) {
                 LOGGER.log(Level.SEVERE, null, e);
             }
@@ -625,19 +626,8 @@ private static Map<String, String> writeOutBoundHeaders(final MultivaluedMap<Str
         return stringHeaders;
     }
 
-    private static final class HttpClientResponseInputStream extends FilterInputStream {
-
-        HttpClientResponseInputStream(final InputStream inputStream) throws IOException {
-            super(inputStream);
-        }
-
-        @Override
-        public void close() throws IOException {
-            super.close();
-        }
-    }
-
-    private static InputStream getInputStream(final CloseableHttpResponse response) throws IOException {
+    private static InputStream getInputStream(final CloseableHttpResponse response,
+                                              final boolean hasSocketTimeout) throws IOException {
 
         final InputStream inputStream;
 
@@ -655,8 +645,16 @@ private static InputStream getInputStream(final CloseableHttpResponse response)
         return new FilterInputStream(inputStream) {
             @Override
             public void close() throws IOException {
-                response.close();
-                super.close();
+                try {
+                    // If no read timeout was set on the socket, super.close() can block indefinitely
+                    if (hasSocketTimeout) {
+                        super.close();
+                    }
+                } catch (IOException ex) {
+                    // Ignore
+                } finally {
+                    response.close();
+                }
             }
         };
     }

connectors/apache-connector/src/test/java/org/glassfish/jersey/apache/connector/StreamingTest.java

@@ -49,10 +49,13 @@
 import javax.ws.rs.client.WebTarget;
 import javax.ws.rs.core.Application;
 import javax.ws.rs.core.MediaType;
-
+import javax.ws.rs.core.Response;
 import javax.inject.Singleton;
 
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+
 import org.glassfish.jersey.client.ClientConfig;
+import org.glassfish.jersey.client.ClientProperties;
 import org.glassfish.jersey.server.ChunkedOutput;
 import org.glassfish.jersey.server.ResourceConfig;
 import org.glassfish.jersey.test.JerseyTest;
@@ -64,6 +67,7 @@
  * @author Petr Janouch (petr.janouch at oracle.com)
  */
 public class StreamingTest extends JerseyTest {
+    private PoolingHttpClientConnectionManager connectionManager;
 
     /**
      * Test that a data stream can be terminated from the client side.
@@ -85,8 +89,27 @@ public void clientCloseTest() throws IOException {
         assertEquals("NOK", sendTarget.request().get().readEntity(String.class));
     }
 
+    /**
+     * Tests that closing a response after completely reading the entity reuses the connection
+     */
+    @Test
+    public void reuseConnectionTest() throws IOException {
+        Response response = target().path("/streamingEndpoint/get").request().
+                property(ClientProperties.READ_TIMEOUT, 5_000).get();
+        InputStream is = response.readEntity(InputStream.class);
+        byte[] buf = new byte[8192];
+        is.read(buf);
+        is.close();
+        response.close();
+
+        assertEquals(1, connectionManager.getTotalStats().getAvailable());
+        assertEquals(0, connectionManager.getTotalStats().getLeased());
+    }
+
     @Override
     protected void configureClient(ClientConfig config) {
+        connectionManager = new PoolingHttpClientConnectionManager();
+        config.property(ApacheClientProperties.CONNECTION_MANAGER, connectionManager);
         config.connectorProvider(new ApacheConnectorProvider());
     }
 
@@ -118,5 +141,12 @@ public String sendEvent() {
         public ChunkedOutput<String> get() {
             return output;
         }
+
+        @GET
+        @Path("get")
+        @Produces(MediaType.TEXT_PLAIN)
+        public String getString() {
+            return "OK";
+        }
     }
 }