Refactored HttpUtils to AbstractProxyExchange

This allows the streaming media types and the buffer size to be configured with new properties GatewayMvcProperties.streamingBufferSize and GatewayMvcProperties.streamingMediaTypes.

See gh-3486

Author: spencergibb

spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/GatewayServerMvcAutoConfiguration.java

@@ -88,8 +88,9 @@ public RestClientCustomizer gatewayRestClientCustomizer(ClientHttpRequestFactory
 
 	@Bean
 	@ConditionalOnMissingBean(ProxyExchange.class)
-	public RestClientProxyExchange restClientProxyExchange(RestClient.Builder restClientBuilder) {
-		return new RestClientProxyExchange(restClientBuilder.build());
+	public RestClientProxyExchange restClientProxyExchange(RestClient.Builder restClientBuilder,
+			GatewayMvcProperties properties) {
+		return new RestClientProxyExchange(restClientBuilder.build(), properties);
 	}
 
 	@Bean

spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/common/HttpUtils.java renamed to spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/common/AbstractProxyExchange.java

@@ -20,29 +20,29 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 
+import org.springframework.cloud.gateway.server.mvc.config.GatewayMvcProperties;
+import org.springframework.cloud.gateway.server.mvc.handler.ProxyExchange;
 import org.springframework.http.client.ClientHttpResponse;
 import org.springframework.util.Assert;
 import org.springframework.util.StreamUtils;
 
-import static org.springframework.http.MediaType.TEXT_EVENT_STREAM;
+public abstract class AbstractProxyExchange implements ProxyExchange {
 
-public abstract class HttpUtils {
+	private final GatewayMvcProperties properties;
 
-	private static final int BUFFER_SIZE = 16384;
-
-	private HttpUtils() {
-		throw new AssertionError("Must not instantiate utility class.");
+	protected AbstractProxyExchange(GatewayMvcProperties properties) {
+		this.properties = properties;
 	}
 
-	public static int copyResponseBody(ClientHttpResponse clientResponse, InputStream inputStream,
+	protected int copyResponseBody(ClientHttpResponse clientResponse, InputStream inputStream,
 			OutputStream outputStream) throws IOException {
 		Assert.notNull(clientResponse, "No ClientResponse specified");
 		Assert.notNull(inputStream, "No InputStream specified");
 		Assert.notNull(outputStream, "No OutputStream specified");
 
 		int transferredBytes;
 
-		if (TEXT_EVENT_STREAM.equals(clientResponse.getHeaders().getContentType())) {
+		if (properties.getStreamingMediaTypes().contains(clientResponse.getHeaders().getContentType())) {
 			transferredBytes = copyResponseBodyWithFlushing(inputStream, outputStream);
 		}
 		else {
@@ -52,11 +52,10 @@ public static int copyResponseBody(ClientHttpResponse clientResponse, InputStrea
 		return transferredBytes;
 	}
 
-	private static int copyResponseBodyWithFlushing(InputStream inputStream, OutputStream outputStream)
-			throws IOException {
+	private int copyResponseBodyWithFlushing(InputStream inputStream, OutputStream outputStream) throws IOException {
 		int readBytes;
 		var totalReadBytes = 0;
-		var buffer = new byte[BUFFER_SIZE];
+		var buffer = new byte[properties.getStreamingBufferSize()];
 
 		while ((readBytes = inputStream.read(buffer)) != -1) {
 			outputStream.write(buffer, 0, readBytes);

spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/config/GatewayMvcProperties.java

@@ -18,6 +18,7 @@
 
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.LinkedHashMap;
 import java.util.List;
 
@@ -26,6 +27,7 @@
 
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.core.style.ToStringCreator;
+import org.springframework.http.MediaType;
 
 @ConfigurationProperties(GatewayMvcProperties.PREFIX)
 public class GatewayMvcProperties {
@@ -51,6 +53,18 @@ public class GatewayMvcProperties {
 
 	private HttpClient httpClient = new HttpClient();
 
+	/**
+	 * Mime-types that are streaming.
+	 */
+	private List<MediaType> streamingMediaTypes = Arrays.asList(MediaType.TEXT_EVENT_STREAM,
+			new MediaType("application", "stream+json"), new MediaType("application", "grpc"),
+			new MediaType("application", "grpc+protobuf"), new MediaType("application", "grpc+json"));
+
+	/**
+	 * Buffer size for streaming media mime-types.
+	 */
+	private int streamingBufferSize = 16384;
+
 	public List<RouteProperties> getRoutes() {
 		return routes;
 	}
@@ -71,11 +85,29 @@ public HttpClient getHttpClient() {
 		return httpClient;
 	}
 
+	public List<MediaType> getStreamingMediaTypes() {
+		return streamingMediaTypes;
+	}
+
+	public void setStreamingMediaTypes(List<MediaType> streamingMediaTypes) {
+		this.streamingMediaTypes = streamingMediaTypes;
+	}
+
+	public int getStreamingBufferSize() {
+		return streamingBufferSize;
+	}
+
+	public void setStreamingBufferSize(int streamingBufferSize) {
+		this.streamingBufferSize = streamingBufferSize;
+	}
+
 	@Override
 	public String toString() {
 		return new ToStringCreator(this).append("httpClient", httpClient)
 			.append("routes", routes)
 			.append("routesMap", routesMap)
+			.append("streamingMediaTypes", streamingMediaTypes)
+			.append("streamingBufferSize", streamingBufferSize)
 			.toString();
 	}
 

spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/ClientHttpRequestFactoryProxyExchange.java

@@ -20,19 +20,28 @@
 import java.io.InputStream;
 import java.io.UncheckedIOException;
 
-import org.springframework.cloud.gateway.server.mvc.common.HttpUtils;
+import org.springframework.cloud.gateway.server.mvc.common.AbstractProxyExchange;
 import org.springframework.cloud.gateway.server.mvc.common.MvcUtils;
+import org.springframework.cloud.gateway.server.mvc.config.GatewayMvcProperties;
 import org.springframework.http.client.ClientHttpRequest;
 import org.springframework.http.client.ClientHttpRequestFactory;
 import org.springframework.http.client.ClientHttpResponse;
 import org.springframework.util.StreamUtils;
 import org.springframework.web.servlet.function.ServerResponse;
 
-public class ClientHttpRequestFactoryProxyExchange implements ProxyExchange {
+public class ClientHttpRequestFactoryProxyExchange extends AbstractProxyExchange {
 
 	private final ClientHttpRequestFactory requestFactory;
 
+	@Deprecated
 	public ClientHttpRequestFactoryProxyExchange(ClientHttpRequestFactory requestFactory) {
+		super(new GatewayMvcProperties());
+		this.requestFactory = requestFactory;
+	}
+
+	public ClientHttpRequestFactoryProxyExchange(ClientHttpRequestFactory requestFactory,
+			GatewayMvcProperties properties) {
+		super(properties);
 		this.requestFactory = requestFactory;
 	}
 
@@ -55,7 +64,7 @@ public ServerResponse exchange(Request request) {
 						InputStream inputStream = MvcUtils.getAttribute(request.getServerRequest(),
 								MvcUtils.CLIENT_RESPONSE_INPUT_STREAM_ATTR);
 						// copy body from request to clientHttpRequest
-						HttpUtils.copyResponseBody(clientHttpResponse, inputStream,
+						ClientHttpRequestFactoryProxyExchange.this.copyResponseBody(clientHttpResponse, inputStream,
 								httpServletResponse.getOutputStream());
 					}
 					return null;

spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/RestClientProxyExchange.java

@@ -21,18 +21,26 @@
 import java.io.OutputStream;
 import java.io.UncheckedIOException;
 
-import org.springframework.cloud.gateway.server.mvc.common.HttpUtils;
+import org.springframework.cloud.gateway.server.mvc.common.AbstractProxyExchange;
 import org.springframework.cloud.gateway.server.mvc.common.MvcUtils;
+import org.springframework.cloud.gateway.server.mvc.config.GatewayMvcProperties;
 import org.springframework.http.client.ClientHttpResponse;
 import org.springframework.util.StreamUtils;
 import org.springframework.web.client.RestClient;
 import org.springframework.web.servlet.function.ServerResponse;
 
-public class RestClientProxyExchange implements ProxyExchange {
+public class RestClientProxyExchange extends AbstractProxyExchange {
 
 	private final RestClient restClient;
 
+	@Deprecated
 	public RestClientProxyExchange(RestClient restClient) {
+		super(new GatewayMvcProperties());
+		this.restClient = restClient;
+	}
+
+	public RestClientProxyExchange(RestClient restClient, GatewayMvcProperties properties) {
+		super(properties);
 		this.restClient = restClient;
 	}
 
@@ -60,7 +68,7 @@ private static int copyBody(Request request, OutputStream outputStream) throws I
 		return StreamUtils.copy(request.getServerRequest().servletRequest().getInputStream(), outputStream);
 	}
 
-	private static ServerResponse doExchange(Request request, ClientHttpResponse clientResponse) throws IOException {
+	private ServerResponse doExchange(Request request, ClientHttpResponse clientResponse) throws IOException {
 		InputStream body = clientResponse.getBody();
 		// put the body input stream in a request attribute so filters can read it.
 		MvcUtils.putAttribute(request.getServerRequest(), MvcUtils.CLIENT_RESPONSE_INPUT_STREAM_ATTR, body);
@@ -72,7 +80,8 @@ private static ServerResponse doExchange(Request request, ClientHttpResponse cli
 					InputStream inputStream = MvcUtils.getAttribute(request.getServerRequest(),
 							MvcUtils.CLIENT_RESPONSE_INPUT_STREAM_ATTR);
 					// copy body from request to clientHttpRequest
-					HttpUtils.copyResponseBody(clientResponse, inputStream, httpServletResponse.getOutputStream());
+					RestClientProxyExchange.this.copyResponseBody(clientResponse, inputStream,
+							httpServletResponse.getOutputStream());
 				}
 				return null;
 			});

spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/common/HttpUtilsTests.java renamed to spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/common/AbstractProxyExchangeTests.java

@@ -22,8 +22,10 @@
 
 import org.junit.jupiter.api.Test;
 
+import org.springframework.cloud.gateway.server.mvc.config.GatewayMvcProperties;
 import org.springframework.http.MediaType;
 import org.springframework.mock.http.client.MockClientHttpResponse;
+import org.springframework.web.servlet.function.ServerResponse;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
@@ -35,7 +37,7 @@
 /**
  * @author Jens Mallien
  */
-public class HttpUtilsTests {
+public class AbstractProxyExchangeTests {
 
 	@Test
 	public void copyResponseBodyForJson() throws IOException {
@@ -46,7 +48,7 @@ public void copyResponseBodyForJson() throws IOException {
 		when(inputStream.transferTo(any())).thenReturn(3L);
 		OutputStream outputStream = mock(OutputStream.class);
 
-		int result = HttpUtils.copyResponseBody(mockResponse, inputStream, outputStream);
+		int result = new TestProxyExchange().copyResponseBody(mockResponse, inputStream, outputStream);
 
 		assertThat(result).isEqualTo(3);
 		verify(outputStream, times(1)).flush();
@@ -58,14 +60,10 @@ public void copyResponseBodyForTextEventStream() throws IOException {
 		mockResponse.getHeaders().setContentType(MediaType.TEXT_EVENT_STREAM);
 
 		InputStream inputStream = mock(InputStream.class);
-		when(inputStream.read(any()))
-				.thenReturn(1)
-				.thenReturn(1)
-				.thenReturn(1)
-				.thenReturn(-1);
+		when(inputStream.read(any())).thenReturn(1).thenReturn(1).thenReturn(1).thenReturn(-1);
 		OutputStream outputStream = mock(OutputStream.class);
 
-		int result = HttpUtils.copyResponseBody(mockResponse, inputStream, outputStream);
+		int result = new TestProxyExchange().copyResponseBody(mockResponse, inputStream, outputStream);
 
 		assertThat(result).isEqualTo(3);
 		verify(outputStream, times(4)).flush();
@@ -79,10 +77,23 @@ public void copyResponseBodyWithoutContentType() throws IOException {
 		when(inputStream.transferTo(any())).thenReturn(3L);
 		OutputStream outputStream = mock(OutputStream.class);
 
-		int result = HttpUtils.copyResponseBody(mockResponse, inputStream, outputStream);
+		int result = new TestProxyExchange().copyResponseBody(mockResponse, inputStream, outputStream);
 
 		assertThat(result).isEqualTo(3);
 		verify(outputStream, times(1)).flush();
 	}
 
+	class TestProxyExchange extends AbstractProxyExchange {
+
+		protected TestProxyExchange() {
+			super(new GatewayMvcProperties());
+		}
+
+		@Override
+		public ServerResponse exchange(Request request) {
+			return ServerResponse.ok().build();
+		}
+
+	}
+
 }