Support streaming with HTTP interfaces + RestClient

Closes gh-32358

Author: rstoyanchev

framework-docs/modules/ROOT/pages/integration/rest-clients.adoc

@@ -999,6 +999,10 @@ Method parameters cannot be `null` unless the `required` attribute (where availa
 parameter annotation) is set to `false`, or the parameter is marked optional as determined by
 {spring-framework-api}/core/MethodParameter.html#isOptional()[`MethodParameter#isOptional`].
 
+`RestClientAdapter` provides additional support for a method parameter of type
+`StreamingHttpOutputMessage.Body` that allows sending the request body by writing to an
+`OutputStream`.
+
 
 
 [[rest-http-interface.custom-resolver]]
@@ -1094,6 +1098,11 @@ depends on how the underlying HTTP client is configured. You can set a `blockTim
 value on the adapter level as well, but we recommend relying on timeout settings of the
 underlying HTTP client, which operates at a lower level and provides more control.
 
+`RestClientAdapter` provides supports additional support for a return value of type
+`InputStream` or `ResponseEntity<InputStream>` that provides access to the raw response
+body content.
+
+
 
 [[rest-http-interface-exceptions]]
 === Error Handling

spring-web/src/main/java/org/springframework/web/client/support/RestClientAdapter.java

@@ -16,6 +16,7 @@
 
 package org.springframework.web.client.support;
 
+import java.io.InputStream;
 import java.net.URI;
 import java.util.ArrayList;
 import java.util.List;
@@ -27,6 +28,7 @@
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.ResponseEntity;
+import org.springframework.http.StreamingHttpOutputMessage;
 import org.springframework.util.Assert;
 import org.springframework.web.client.RestClient;
 import org.springframework.web.service.invoker.HttpExchangeAdapter;
@@ -70,8 +72,12 @@ public HttpHeaders exchangeForHeaders(HttpRequestValues values) {
 		return newRequest(values).retrieve().toBodilessEntity().getHeaders();
 	}
 
+	@SuppressWarnings("unchecked")
 	@Override
 	public <T> @Nullable T exchangeForBody(HttpRequestValues values, ParameterizedTypeReference<T> bodyType) {
+		if (bodyType.getType().equals(InputStream.class)) {
+			return (T) newRequest(values).exchange((request, response) -> response.getBody(), false);
+		}
 		return newRequest(values).retrieve().body(bodyType);
 	}
 
@@ -80,8 +86,15 @@ public ResponseEntity<Void> exchangeForBodilessEntity(HttpRequestValues values)
 		return newRequest(values).retrieve().toBodilessEntity();
 	}
 
+	@SuppressWarnings("unchecked")
 	@Override
 	public <T> ResponseEntity<T> exchangeForEntity(HttpRequestValues values, ParameterizedTypeReference<T> bodyType) {
+		if (bodyType.getType().equals(InputStream.class)) {
+			return (ResponseEntity<T>) newRequest(values).exchangeForRequiredValue((request, response) ->
+					ResponseEntity.status(response.getStatusCode())
+							.headers(response.getHeaders())
+							.body(response.getBody()), false);
+		}
 		return newRequest(values).retrieve().toEntity(bodyType);
 	}
 
@@ -130,7 +143,10 @@ else if (values.getUriTemplate() != null) {
 
 		B body = (B) values.getBodyValue();
 		if (body != null) {
-			if (values.getBodyValueType() != null) {
+			if (body instanceof StreamingHttpOutputMessage.Body streamingBody) {
+				bodySpec.body(streamingBody);
+			}
+			else if (values.getBodyValueType() != null) {
 				bodySpec.body(body, (ParameterizedTypeReference<? super B>) values.getBodyValueType());
 			}
 			else {

spring-web/src/main/java/org/springframework/web/service/invoker/RequestBodyArgumentResolver.java

@@ -24,6 +24,7 @@
 import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.core.ReactiveAdapter;
 import org.springframework.core.ReactiveAdapterRegistry;
+import org.springframework.http.StreamingHttpOutputMessage;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 import org.springframework.web.bind.annotation.RequestBody;
@@ -66,6 +67,11 @@ public RequestBodyArgumentResolver(HttpExchangeAdapter exchangeAdapter) {
 	public boolean resolve(
 			@Nullable Object argument, MethodParameter parameter, HttpRequestValues.Builder requestValues) {
 
+		if (parameter.getParameterType().equals(StreamingHttpOutputMessage.Body.class)) {
+			requestValues.setBodyValue(argument);
+			return true;
+		}
+
 		RequestBody annot = parameter.getParameterAnnotation(RequestBody.class);
 		if (annot == null) {
 			return false;

spring-web/src/test/java/org/springframework/web/client/support/RestClientAdapterTests.java

@@ -17,11 +17,13 @@
 package org.springframework.web.client.support;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.lang.annotation.ElementType;
 import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 import java.net.URI;
+import java.nio.charset.StandardCharsets;
 import java.util.LinkedHashSet;
 import java.util.Optional;
 import java.util.Set;
@@ -41,8 +43,10 @@
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
+import org.springframework.http.StreamingHttpOutputMessage;
 import org.springframework.util.LinkedMultiValueMap;
 import org.springframework.util.MultiValueMap;
+import org.springframework.util.StreamUtils;
 import org.springframework.web.bind.annotation.CookieValue;
 import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.RequestBody;
@@ -300,6 +304,25 @@ void putWithSameNameCookies(MockWebServer server, Service service) throws Except
 		assertThat(request.getHeader("Cookie")).isEqualTo("testCookie=test1; testCookie=test2");
 	}
 
+	@Test
+	void getInputStream() throws Exception {
+		InputStream inputStream = initService().getInputStream();
+
+		RecordedRequest request = this.anotherServer.takeRequest();
+		assertThat(request.getPath()).isEqualTo("/input-stream");
+		assertThat(StreamUtils.copyToString(inputStream, StandardCharsets.UTF_8)).isEqualTo("Hello Spring 2!");
+	}
+
+	@Test
+	void postOutputStream() throws Exception {
+		String body = "test stream";
+		initService().postOutputStream(outputStream -> outputStream.write(body.getBytes()));
+
+		RecordedRequest request = this.anotherServer.takeRequest();
+		assertThat(request.getPath()).isEqualTo("/output-stream");
+		assertThat(request.getBody().readUtf8()).isEqualTo(body);
+	}
+
 
 	private static MockWebServer anotherServer() {
 		MockWebServer server = new MockWebServer();
@@ -309,6 +332,13 @@ private static MockWebServer anotherServer() {
 		return server;
 	}
 
+	private Service initService() {
+		String url = this.anotherServer.url("/").toString();
+		RestClient restClient = RestClient.builder().baseUrl(url).build();
+		RestClientAdapter adapter = RestClientAdapter.create(restClient);
+		return HttpServiceProxyFactory.builderFor(adapter).build().createClient(Service.class);
+	}
+
 
 	private interface Service {
 
@@ -353,6 +383,12 @@ ResponseEntity<String> getWithUriBuilderFactory(
 		void putWithSameNameCookies(
 				@CookieValue("testCookie") String firstCookie, @CookieValue("testCookie") String secondCookie);
 
+		@GetExchange(url = "/input-stream")
+		InputStream getInputStream();
+
+		@PostExchange(url = "/output-stream")
+		void postOutputStream(StreamingHttpOutputMessage.Body body);
+
 	}