From 6a1b21f58bbaff0ae8d9d31d69062c7228536e92 Mon Sep 17 00:00:00 2001 From: Jens Mallien Date: Tue, 30 Jul 2024 13:16:53 +0200 Subject: [PATCH] Adds flushing after each buffer write in RestClientProxyExchange in case of text/event-stream Fixes gh-3410 Fixes gh-3486 --- .../gateway/server/mvc/common/HttpUtils.java | 79 +++++++++++++++++ ...ClientHttpRequestFactoryProxyExchange.java | 4 +- .../mvc/handler/RestClientProxyExchange.java | 3 +- .../server/mvc/common/HttpUtilsTests.java | 88 +++++++++++++++++++ 4 files changed, 172 insertions(+), 2 deletions(-) create mode 100644 spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/common/HttpUtils.java create mode 100644 spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/common/HttpUtilsTests.java diff --git a/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/common/HttpUtils.java b/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/common/HttpUtils.java new file mode 100644 index 00000000..9cd20c9f --- /dev/null +++ b/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/common/HttpUtils.java @@ -0,0 +1,79 @@ +/* + * Copyright 2013-2023 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 + * + * https://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.cloud.gateway.server.mvc.common; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + +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 HttpUtils { + + private static final int BUFFER_SIZE = 16384; + + private HttpUtils() { + throw new AssertionError("Must not instantiate utility class."); + } + + public static 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())) { + transferredBytes = copyResponseBodyWithFlushing(inputStream, outputStream); + } + else { + transferredBytes = StreamUtils.copy(inputStream, outputStream); + } + + return transferredBytes; + } + + private static int copyResponseBodyWithFlushing(InputStream inputStream, OutputStream outputStream) + throws IOException { + int readBytes; + var totalReadBytes = 0; + var buffer = new byte[BUFFER_SIZE]; + + while ((readBytes = inputStream.read(buffer)) != -1) { + outputStream.write(buffer, 0, readBytes); + outputStream.flush(); + if (totalReadBytes < Integer.MAX_VALUE) { + try { + totalReadBytes = Math.addExact(totalReadBytes, readBytes); + } + catch (ArithmeticException e) { + totalReadBytes = Integer.MAX_VALUE; + } + } + } + + outputStream.flush(); + + return totalReadBytes; + } + +} diff --git a/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/ClientHttpRequestFactoryProxyExchange.java b/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/ClientHttpRequestFactoryProxyExchange.java index ccda73d4..056653c1 100644 --- a/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/ClientHttpRequestFactoryProxyExchange.java +++ b/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/ClientHttpRequestFactoryProxyExchange.java @@ -20,6 +20,7 @@ import java.io.IOException; 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.MvcUtils; import org.springframework.http.client.ClientHttpRequest; import org.springframework.http.client.ClientHttpRequestFactory; @@ -54,7 +55,8 @@ public class ClientHttpRequestFactoryProxyExchange implements ProxyExchange { InputStream inputStream = MvcUtils.getAttribute(request.getServerRequest(), MvcUtils.CLIENT_RESPONSE_INPUT_STREAM_ATTR); // copy body from request to clientHttpRequest - StreamUtils.copy(inputStream, httpServletResponse.getOutputStream()); + HttpUtils.copyResponseBody(clientHttpResponse, inputStream, + httpServletResponse.getOutputStream()); } return null; }); diff --git a/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/RestClientProxyExchange.java b/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/RestClientProxyExchange.java index 27a02569..91fce3f7 100644 --- a/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/RestClientProxyExchange.java +++ b/spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/RestClientProxyExchange.java @@ -21,6 +21,7 @@ import java.io.InputStream; 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.MvcUtils; import org.springframework.http.client.ClientHttpResponse; import org.springframework.util.StreamUtils; @@ -71,7 +72,7 @@ public class RestClientProxyExchange implements ProxyExchange { InputStream inputStream = MvcUtils.getAttribute(request.getServerRequest(), MvcUtils.CLIENT_RESPONSE_INPUT_STREAM_ATTR); // copy body from request to clientHttpRequest - StreamUtils.copy(inputStream, httpServletResponse.getOutputStream()); + HttpUtils.copyResponseBody(clientResponse, inputStream, httpServletResponse.getOutputStream()); } return null; }); diff --git a/spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/common/HttpUtilsTests.java b/spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/common/HttpUtilsTests.java new file mode 100644 index 00000000..839232ca --- /dev/null +++ b/spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/common/HttpUtilsTests.java @@ -0,0 +1,88 @@ +/* + * Copyright 2013-2024 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 + * + * https://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.cloud.gateway.server.mvc.common; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + +import org.junit.jupiter.api.Test; + +import org.springframework.http.MediaType; +import org.springframework.mock.http.client.MockClientHttpResponse; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * @author Jens Mallien + */ +public class HttpUtilsTests { + + @Test + public void copyResponseBodyForJson() throws IOException { + MockClientHttpResponse mockResponse = new MockClientHttpResponse(new byte[0], 200); + mockResponse.getHeaders().setContentType(MediaType.APPLICATION_JSON); + + InputStream inputStream = mock(InputStream.class); + when(inputStream.transferTo(any())).thenReturn(3L); + OutputStream outputStream = mock(OutputStream.class); + + int result = HttpUtils.copyResponseBody(mockResponse, inputStream, outputStream); + + assertThat(result).isEqualTo(3); + verify(outputStream, times(1)).flush(); + } + + @Test + public void copyResponseBodyForTextEventStream() throws IOException { + MockClientHttpResponse mockResponse = new MockClientHttpResponse(new byte[0], 200); + mockResponse.getHeaders().setContentType(MediaType.TEXT_EVENT_STREAM); + + InputStream inputStream = mock(InputStream.class); + when(inputStream.read(any())) + .thenReturn(1) + .thenReturn(1) + .thenReturn(1) + .thenReturn(-1); + OutputStream outputStream = mock(OutputStream.class); + + int result = HttpUtils.copyResponseBody(mockResponse, inputStream, outputStream); + + assertThat(result).isEqualTo(3); + verify(outputStream, times(4)).flush(); + } + + @Test + public void copyResponseBodyWithoutContentType() throws IOException { + MockClientHttpResponse mockResponse = new MockClientHttpResponse(new byte[0], 200); + + InputStream inputStream = mock(InputStream.class); + when(inputStream.transferTo(any())).thenReturn(3L); + OutputStream outputStream = mock(OutputStream.class); + + int result = HttpUtils.copyResponseBody(mockResponse, inputStream, outputStream); + + assertThat(result).isEqualTo(3); + verify(outputStream, times(1)).flush(); + } + +}