Improve HTTP support for JDK client and HttpInvoker server

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/main/java/org/springframework/remoting/httpinvoker/HttpInvokerServiceExporter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 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.
@@ -16,6 +16,7 @@
 
 package org.springframework.remoting.httpinvoker;
 
+import java.io.FilterOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.ObjectInputStream;
@@ -169,7 +170,8 @@ protected void writeRemoteInvocationResult(
 			HttpServletRequest request, HttpServletResponse response, RemoteInvocationResult result, OutputStream os)
 			throws IOException {
 
-		ObjectOutputStream oos = createObjectOutputStream(decorateOutputStream(request, response, os));
+		ObjectOutputStream oos =
+				createObjectOutputStream(new FlushGuardedOutputStream(decorateOutputStream(request, response, os)));
 		try {
 			doWriteRemoteInvocationResult(result, oos);
 		}
@@ -195,4 +197,25 @@ protected OutputStream decorateOutputStream(
 		return os;
 	}
 
+	/**
+	 * Decorate an OutputStream to guard against {@code flush()} calls, which
+	 * are turned into no-ops.
+	 * <p>Because {@link ObjectOutputStream#close()} will in fact flush/drain
+	 * the underlying stream twice, this {@link FilterOutputStream} will
+	 * guard against individual flush calls. Multiple flush calls can lead
+	 * to performance issues, since writes aren't gathered as they should be.
+	 *
+	 * @see <a href="https://jira.spring.io/browse/SPR-14040">SPR-14040</a>
+	 */
+	class FlushGuardedOutputStream extends FilterOutputStream {
+		public FlushGuardedOutputStream(OutputStream out) {
+			super(out);
+		}
+
+		@Override
+		public void flush() throws IOException {
+			// Do nothing
+		}
+	}
+
 }

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