GH-2300: Add Flux support in WebFluxRequestExecMH

Fixes: spring-projects/spring-integration#2300

To allow to consume a streaming HTTP response downstream expose
`replyToFlux` option on the `WebFluxRequestExecutingMessageHandler`.
This way the body of the HTTP response can be converted now to the
`Flux` for subsequent output message.
The option is `false` by default; can be changed to `true` in the `5.1`

spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests-context.xml

@@ -20,22 +20,23 @@
 	</si:channel>
 
 	<outbound-gateway id="reactiveMinimalConfig" url="http://localhost/test1" request-channel="requests"
-							   web-client="webClient"/>
+					  web-client="webClient"/>
 
 	<outbound-gateway id="reactiveFullConfig"
-							   url="http://localhost/test2"
-							   http-method="PUT"
-							   request-channel="requests"
-							   reply-timeout="1234"
-							   extract-request-payload="false"
-							   expected-response-type="java.lang.String"
-							   mapped-request-headers="requestHeader1, requestHeader2"
-							   mapped-response-headers="responseHeader"
-							   reply-channel="replies"
-							   charset="UTF-8"
-							   order="77"
-							   auto-startup="false"
-							   transfer-cookies="true">
+					  url="http://localhost/test2"
+					  http-method="PUT"
+					  request-channel="requests"
+					  reply-timeout="1234"
+					  extract-request-payload="false"
+					  expected-response-type="java.lang.String"
+					  mapped-request-headers="requestHeader1, requestHeader2"
+					  mapped-response-headers="responseHeader"
+					  reply-channel="replies"
+					  charset="UTF-8"
+					  order="77"
+					  auto-startup="false"
+					  transfer-cookies="true"
+					  reply-to-flux="true">
 		<uri-variable name="foo" expression="headers.bar"/>
 	</outbound-gateway>
 

spring-integration-webflux/src/test/java/org/springframework/integration/webflux/config/WebFluxOutboundGatewayParserTests.java

@@ -82,6 +82,7 @@ public class WebFluxOutboundGatewayParserTests {
 		assertEquals(Charset.forName("UTF-8"), handlerAccessor.getPropertyValue("charset"));
 		assertEquals(true, handlerAccessor.getPropertyValue("extractPayload"));
 		assertEquals(false, handlerAccessor.getPropertyValue("transferCookies"));
+		assertEquals(false, handlerAccessor.getPropertyValue("replyToFlux"));
 	}
 
 	@Test
@@ -123,6 +124,7 @@ public class WebFluxOutboundGatewayParserTests {
 		assertTrue(ObjectUtils.containsElement(mappedRequestHeaders, "requestHeader2"));
 		assertEquals("responseHeader", mappedResponseHeaders[0]);
 		assertEquals(true, handlerAccessor.getPropertyValue("transferCookies"));
+		assertEquals(true, handlerAccessor.getPropertyValue("replyToFlux"));
 	}
 
 }

spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java

@@ -28,7 +28,10 @@ import java.util.List;
 import org.junit.Test;
 import org.reactivestreams.Subscriber;
 
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
 import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
 import org.springframework.http.client.reactive.ClientHttpConnector;
 import org.springframework.integration.channel.FluxMessageChannel;
 import org.springframework.integration.channel.QueueChannel;
@@ -42,6 +45,7 @@ import org.springframework.test.web.reactive.server.HttpHandlerConnector;
 import org.springframework.web.reactive.function.client.WebClient;
 import org.springframework.web.reactive.function.client.WebClientResponseException;
 
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
@@ -54,7 +58,7 @@ import reactor.test.StepVerifier;
 public class WebFluxRequestExecutingMessageHandlerTests {
 
 	@Test
-	public void testReactiveReturn() throws Throwable {
+	public void testReactiveReturn() {
 		ClientHttpConnector httpConnector =
 				new HttpHandlerConnector((request, response) -> {
 					response.setStatusCode(HttpStatus.OK);
@@ -84,7 +88,7 @@ public class WebFluxRequestExecutingMessageHandlerTests {
 	}
 
 	@Test
-	public void testReactiveErrorOneWay() throws Throwable {
+	public void testReactiveErrorOneWay() {
 		ClientHttpConnector httpConnector =
 				new HttpHandlerConnector((request, response) -> {
 					response.setStatusCode(HttpStatus.UNAUTHORIZED);
@@ -114,7 +118,7 @@ public class WebFluxRequestExecutingMessageHandlerTests {
 	}
 
 	@Test
-	public void testReactiveConnectErrorOneWay() throws Throwable {
+	public void testReactiveConnectErrorOneWay() {
 		ClientHttpConnector httpConnector =
 				new HttpHandlerConnector((request, response) -> {
 					throw new RuntimeException("Intentional connection error");
@@ -182,4 +186,50 @@ public class WebFluxRequestExecutingMessageHandlerTests {
 		assertNull(replyMessage);
 	}
 
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testFluxReply() {
+		ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> {
+			response.setStatusCode(HttpStatus.OK);
+			response.getHeaders().setContentType(MediaType.TEXT_PLAIN);
+
+			DataBufferFactory bufferFactory = response.bufferFactory();
+
+			Flux<DataBuffer> data =
+					Flux.just(bufferFactory.wrap("foo".getBytes()),
+							bufferFactory.wrap("bar".getBytes()),
+							bufferFactory.wrap("baz".getBytes()));
+
+			return response.writeWith(data)
+					.then(Mono.defer(response::setComplete));
+		});
+
+		WebClient webClient = WebClient.builder()
+				.clientConnector(httpConnector)
+				.build();
+
+		String destinationUri = "http://www.springsource.org/spring-integration";
+		WebFluxRequestExecutingMessageHandler reactiveHandler =
+				new WebFluxRequestExecutingMessageHandler(destinationUri, webClient);
+
+		QueueChannel replyChannel = new QueueChannel();
+		reactiveHandler.setOutputChannel(replyChannel);
+		reactiveHandler.setExpectedResponseType(String.class);
+		reactiveHandler.setReplyToFlux(true);
+
+		reactiveHandler.handleMessage(MessageBuilder.withPayload("hello, world").build());
+
+		Message<?> receive = replyChannel.receive(10_000);
+
+		assertNotNull(receive);
+
+		assertThat(receive.getPayload(), instanceOf(Flux.class));
+
+		Flux<String> flux = (Flux<String>) receive.getPayload();
+
+		StepVerifier.create(flux)
+				.expectNext("foo", "bar", "baz")
+				.verifyComplete();
+	}
+
 }