Support of HTTP persistent connections for JDK client

Prior to this commit, HTTP clients relying on the JDK HTTP client would
not properly reuse existing TCP connections (i.e. HTTP 1.1 persisten
connection). The SimpleClientHttpResponse would close the actual connection once the
response is handled.

As explained in the JDK documentation
(http://docs.oracle.com/javase/8/docs/technotes/guides/net/http-keepalive.html)
HTTP clients should do the following to allow resource reuse:

* consume the whole HTTP response content
* close the response inputstream once done

This commit makes sure that the response content is
totally drained and then the stream closed (and not the connection).

Issue: SPR-14040

Author: bclozel

spring-core/src/main/java/org/springframework/util/StreamUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -131,6 +131,25 @@ public static int copy(InputStream in, OutputStream out) throws IOException {
 		return byteCount;
 	}
 
+	/**
+	 * Drain the remaining content of the given InputStream.
+	 * Leaves the InputStream open when done.
+	 * @param in the InputStream to drain
+	 * @return the number of bytes read
+	 * @throws IOException in case of I/O errors
+	 * @since 4.3.0
+	 */
+	public static int drain(InputStream in) throws IOException {
+		Assert.notNull(in, "No InputStream specified");
+		byte[] buffer = new byte[BUFFER_SIZE];
+		int bytesRead = -1;
+		int byteCount = 0;
+		while ((bytesRead = in.read(buffer)) != -1) {
+			byteCount += bytesRead;
+		}
+		return byteCount;
+	}
+
 	/**
 	 * Return an efficient empty {@link InputStream}.
 	 * @return a {@link ByteArrayInputStream} based on an empty byte array

spring-web/src/main/java/org/springframework/http/client/SimpleClientHttpResponse.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2016 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,6 +21,7 @@
 import java.net.HttpURLConnection;
 
 import org.springframework.http.HttpHeaders;
+import org.springframework.util.StreamUtils;
 import org.springframework.util.StringUtils;
 
 /**
@@ -29,6 +30,7 @@
  * {@link SimpleStreamingClientHttpRequest#execute()}.
  *
  * @author Arjen Poutsma
+ * @author Brian Clozel
  * @since 3.0
  */
 final class SimpleClientHttpResponse extends AbstractClientHttpResponse {
@@ -37,6 +39,8 @@ final class SimpleClientHttpResponse extends AbstractClientHttpResponse {
 
 	private HttpHeaders headers;
 
+	private InputStream responseStream;
+
 
 	SimpleClientHttpResponse(HttpURLConnection connection) {
 		this.connection = connection;
@@ -78,12 +82,19 @@ public HttpHeaders getHeaders() {
 	@Override
 	public InputStream getBody() throws IOException {
 		InputStream errorStream = this.connection.getErrorStream();
-		return (errorStream != null ? errorStream : this.connection.getInputStream());
+		this.responseStream = (errorStream != null ? errorStream : this.connection.getInputStream());
+		return this.responseStream;
 	}
 
 	@Override
 	public void close() {
-		this.connection.disconnect();
+		if (this.responseStream != null) {
+			try {
+				StreamUtils.drain(this.responseStream);
+				this.responseStream.close();
+			}
+			catch (IOException e) { }
+		}
 	}
 
 }

spring-web/src/test/java/org/springframework/http/client/SimpleClientHttpResponseTests.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright 2002-2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.http.client;
+
+import static org.hamcrest.MatcherAssert.*;
+import static org.hamcrest.Matchers.*;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.BDDMockito.*;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.HttpURLConnection;
+import java.nio.charset.Charset;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.util.StreamUtils;
+
+/**
+ * @author Brian Clozel
+ */
+public class SimpleClientHttpResponseTests {
+
+	private final Charset UTF8 = Charset.forName("UTF-8");
+
+	private SimpleClientHttpResponse response;
+
+	private HttpURLConnection connection;
+
+	@Before
+	public void setup() throws Exception {
+		this.connection = mock(HttpURLConnection.class);
+		this.response = new SimpleClientHttpResponse(this.connection);
+	}
+
+	// SPR-14040
+	@Test
+	public void shouldNotCloseConnectionWhenResponseClosed() throws Exception {
+		TestByteArrayInputStream is = new TestByteArrayInputStream("Spring".getBytes(UTF8));
+		given(this.connection.getErrorStream()).willReturn(null);
+		given(this.connection.getInputStream()).willReturn(is);
+
+		InputStream responseStream = this.response.getBody();
+		assertThat(StreamUtils.copyToString(responseStream, UTF8), is("Spring"));
+
+		this.response.close();
+		assertTrue(is.isClosed());
+		verify(this.connection, never()).disconnect();
+	}
+
+	// SPR-14040
+	@Test
+	public void shouldDrainStreamWhenResponseClosed() throws Exception {
+		byte[] buf = new byte[6];
+		TestByteArrayInputStream is = new TestByteArrayInputStream("SpringSpring".getBytes(UTF8));
+		given(this.connection.getErrorStream()).willReturn(null);
+		given(this.connection.getInputStream()).willReturn(is);
+
+		InputStream responseStream = this.response.getBody();
+		responseStream.read(buf);
+		assertThat(new String(buf, UTF8), is("Spring"));
+		assertThat(is.available(), is(6));
+
+		this.response.close();
+		assertThat(is.available(), is(0));
+		assertTrue(is.isClosed());
+		verify(this.connection, never()).disconnect();
+	}
+
+	// SPR-14040
+	@Test
+	public void shouldDrainErrorStreamWhenResponseClosed() throws Exception {
+		byte[] buf = new byte[6];
+		TestByteArrayInputStream is = new TestByteArrayInputStream("SpringSpring".getBytes(UTF8));
+		given(this.connection.getErrorStream()).willReturn(is);
+
+		InputStream responseStream = this.response.getBody();
+		responseStream.read(buf);
+		assertThat(new String(buf, UTF8), is("Spring"));
+		assertThat(is.available(), is(6));
+
+		this.response.close();
+		assertThat(is.available(), is(0));
+		assertTrue(is.isClosed());
+		verify(this.connection, never()).disconnect();
+	}
+
+
+	class TestByteArrayInputStream extends ByteArrayInputStream {
+
+		private boolean closed;
+
+		public TestByteArrayInputStream(byte[] buf) {
+			super(buf);
+			this.closed = false;
+		}
+
+		public boolean isClosed() {
+			return closed;
+		}
+
+		@Override
+		public void close() throws IOException {
+			super.close();
+			this.closed = true;
+		}
+	}
+
+}