Closed successful session response body stream

Consumed successful response body content.
Closed successful session response body stream.
Added new test com.cloudant.tests.HttpTest#successfulSessionStreamClose.

Author: ricellis

CHANGES.md

@@ -1,3 +1,7 @@
+# Unreleased
+- [FIXED] Connection leak regression introduced in 2.18.0 caused by not closing streams from
+  successful session response bodies.
+
 # 2.19.0 (2020-03-02)
 - [NEW] Add getter for total row count to `AllDocsResponse`
 - [NEW] Added `DbInfo#getSizes()` for access to improved sizes information.

cloudant-client/src/test/java/com/cloudant/tests/HttpTest.java

@@ -81,6 +81,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
+import java.net.HttpURLConnection;
 import java.net.URI;
 import java.net.URL;
 import java.net.URLClassLoader;
@@ -96,6 +97,7 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.regex.Pattern;
 import java.util.stream.Stream;
 
@@ -1254,5 +1256,48 @@ public static RecordedRequest[] takeN(MockWebServer server, int n) throws Except
         return recordedRequests;
     }
 
+    /**
+     * Test that a response stream from a successful _session request is consumed and closed.
+     *
+     * @throws Exception
+     */
+    @TestTemplate
+    public void successfulSessionStreamClose() throws Exception {
+        // Queue a mock response for the _session request
+        mockWebServer.enqueue(OK_COOKIE);
+        // Queue a mock response for an _all_dbs request
+        mockWebServer.enqueue(new MockResponse().setBody("[]"));
 
+        final AtomicReference<InputStream> responseStream = new AtomicReference<InputStream>();
+        CloudantClient c = CloudantClientHelper.newMockWebServerClientBuilder(mockWebServer)
+                .interceptors(new HttpConnectionResponseInterceptor() {
+                    @Override
+                    public HttpConnectionInterceptorContext interceptResponse(HttpConnectionInterceptorContext context) {
+                        try {
+                            // Store the response stream from the session request
+                            if (context.connection.url.toString().contains("_session")) {
+                                HttpURLConnection conn = context.connection.getConnection();
+                                responseStream.set(context.connection.getConnection().getInputStream());
+                            }
+                        } catch (IOException e) {
+                            fail("IOException in test interceptor.");
+                        }
+                        return context;
+                    }
+                })
+                .username("a")
+                .password("b")
+                .build();
+        // Make a request to init the session
+        List<String> dbs = c.getAllDbs();
+        // Get the response stream from the session request
+        InputStream stream = responseStream.get();
+        // Assert that the stream has been consumed
+        assertEquals(0, stream.available(), "There should be no bytes available from the stream.");
+        // Assert that the stream is closed
+        assertThrows(IOException.class,
+                () -> stream.read(),
+                "Reading the stream should throw an IOException because the stream should be " +
+                        "closed.");
+    }
 }

cloudant-http/src/main/java/com/cloudant/http/internal/interceptors/CookieInterceptorBase.java

@@ -87,7 +87,7 @@ protected CookieInterceptorBase(String baseUrl, String endpoint, String requestM
      */
     protected abstract byte[] getSessionRequestPayload(HttpConnectionInterceptorContext context);
 
-    private void requestCookie(HttpConnectionInterceptorContext context) {
+    private void requestCookie(HttpConnectionInterceptorContext context) throws IOException {
         // Check if the session was already updated on another thread before getting a cookie
         sessionLock.readLock().lock();
         try {
@@ -99,7 +99,14 @@ private void requestCookie(HttpConnectionInterceptorContext context) {
                     if (sessionId.equals(context.getState(this, sessionStateName, UUID.class))) {
                         HttpConnection sessionConn = makeSessionRequest(sessionRequestUrl,
                                 getSessionRequestPayload(context), sessionRequestMimeType, context);
-                        storeCookiesFromResponse(sessionConn.getConnection());
+                        HttpURLConnection sessionUrlConnection = sessionConn.getConnection();
+                        try {
+                            storeCookiesFromResponse(sessionUrlConnection);
+                        } finally {
+                            // We use collect rather than consume as we don't want to log
+                            // a warning, even though we don't actually need the body
+                            Utils.collectAndCloseStream(sessionUrlConnection.getInputStream());
+                        }
                         // We renewed a cookie, update the global sessionID and this request's context
                         sessionId = UUID.randomUUID();
                         context.setState(this, sessionStateName, sessionId);
@@ -163,21 +170,20 @@ public HttpConnectionInterceptorContext interceptRequest(HttpConnectionIntercept
         // Set the sessionId for this request
         context.setState(this, sessionStateName, sessionId);
         HttpURLConnection connection = context.connection.getConnection();
+        try {
+            // First time we will have no cookies
+            if (cookieManager.getCookieStore().getCookies().isEmpty()) {
+                requestCookie(context);
+            }
 
-        // First time we will have no cookies
-        if (cookieManager.getCookieStore().getCookies().isEmpty()) {
-            requestCookie(context);
-        }
-
-        // Debug logging
-        if (logger.isLoggable(Level.FINEST)) {
-            logger.finest("Attempt to add cookie to request.");
-            logger.finest("Cookies are stored for URIs: " + cookieManager.getCookieStore()
-                    .getURIs());
-        }
+            // Debug logging
+            if (logger.isLoggable(Level.FINEST)) {
+                logger.finest("Attempt to add cookie to request.");
+                logger.finest("Cookies are stored for URIs: " + cookieManager.getCookieStore()
+                        .getURIs());
+            }
 
-        // Apply any saved cookies to the request
-        try {
+            // Apply any saved cookies to the request
             Map<String, List<String>> requestCookieHeaders = cookieManager.get(connection
                     .getURL().toURI(), connection.getRequestProperties());
             for (Map.Entry<String, List<String>> requestCookieHeader :
@@ -217,7 +223,7 @@ public HttpConnectionInterceptorContext interceptResponse(HttpConnectionIntercep
                 context.replayRequest = true;
             }
         } catch (IOException e) {
-            throw wrapIOException("Failed to read HTTP reponse code or body from", connection, e);
+            throw wrapIOException("Failed to read HTTP response code or body from", connection, e);
         }
         return context;
     }