Merge pull request cloudant#508 from cloudant/507-cookie-response-stream

Closed successful session response body stream

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;
  }