Adds flushing after each buffer write in RestClientProxyExchange in case of text/event-stream

Fixes gh-3410
Fixes gh-3486

Author: jensmatw

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

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

@@ -20,6 +20,7 @@
 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 ServerResponse exchange(Request request) {
 						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;
 				});

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

@@ -21,6 +21,7 @@
 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 @@ 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
-					StreamUtils.copy(inputStream, httpServletResponse.getOutputStream());
+					HttpUtils.copyResponseBody(clientResponse, inputStream, httpServletResponse.getOutputStream());
 				}
 				return null;
 			});

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