Add Java DSL .fluxTransform(Function) operator (#2541)

* Add Java DSL `.fluxTransform(Function)` operator

For better interoperability with Reactor `Flux` from the
end-user perspective introduce an operator which could
call a target `Function` with integration data wrapped to the `Flux`
and expect a `Publisher<?>` result to continue the flow.
This way end-user just needs to implement a `Function` (or method)
to accept the `Flux` as an input and return some `Publisher`
after the sequence of reactive operators.

Such a new operator also allows a smooth integration with the
Spring Cloud Function, where it is just enough to lookup the function
in the catalog and inject it into this operator

* * Move `fluxTransform()` body to `Transformers` for cleaner context
distribution

spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java

@@ -96,6 +96,7 @@ import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.StringUtils;
 
+import reactor.core.publisher.Flux;
 import reactor.util.function.Tuple2;
 
 /**
@@ -544,7 +545,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	 * @see LambdaMessageProcessor
 	 */
 	public <S, T> B transform(GenericTransformer<S, T> genericTransformer) {
-		return this.transform(null, genericTransformer);
+		return transform(null, genericTransformer);
 	}
 
 	/**
@@ -2866,6 +2867,30 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 		return handle(new ServiceActivatingHandler(triggerAction, "trigger"), endpointConfigurer);
 	}
 
+	/**
+	 * Populate a {@link FluxMessageChannel} to start a reactive processing for upstream data,
+	 * wrap it to a {@link Flux}, apply provided {@link Function} via {@link Flux#transform(Function)}
+	 * and emit the result to one more {@link FluxMessageChannel}, subscribed in the downstream flow.
+	 * @param fluxFunction the {@link Function} to process data reactive manner.
+	 * @return the current {@link IntegrationFlowDefinition}.
+	 */
+	@SuppressWarnings("unchecked")
+	public <I, O> B fluxTransform(Function<? super Flux<Message<I>>, ? extends Publisher<O>> fluxFunction) {
+		if (!(this.currentMessageChannel instanceof FluxMessageChannel)) {
+			channel(new FluxMessageChannel());
+		}
+
+		Publisher<Message<I>> upstream = (Publisher<Message<I>>) this.currentMessageChannel;
+
+		Flux<Message<O>> result = Transformers.transformWithFunction(upstream, fluxFunction);
+
+		FluxMessageChannel downstream = new FluxMessageChannel();
+		downstream.subscribeTo((Flux<Message<?>>) (Flux<?>) result);
+
+		this.currentMessageChannel = downstream;
+
+		return addComponent(this.currentMessageChannel);
+	}
 
 	/**
 	 * Represent an Integration Flow as a Reactive Streams {@link Publisher} bean.

spring-integration-core/src/main/java/org/springframework/integration/dsl/Transformers.java

@@ -16,8 +16,11 @@
 
 package org.springframework.integration.dsl;
 
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 
+import org.reactivestreams.Publisher;
+
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.core.serializer.Deserializer;
 import org.springframework.core.serializer.Serializer;
@@ -27,6 +30,7 @@ import org.springframework.integration.codec.Codec;
 import org.springframework.integration.expression.FunctionExpression;
 import org.springframework.integration.json.JsonToObjectTransformer;
 import org.springframework.integration.json.ObjectToJsonTransformer;
+import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.support.json.JsonObjectMapper;
 import org.springframework.integration.transformer.DecodingTransformer;
 import org.springframework.integration.transformer.EncodingPayloadTransformer;
@@ -41,6 +45,9 @@ import org.springframework.integration.transformer.SyslogToMapTransformer;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
 /**
  * An utility class to provide methods for out-of-the-box
  * {@link org.springframework.integration.transformer.Transformer}s.
@@ -262,4 +269,33 @@ public abstract class Transformers {
 		return new StreamTransformer(charset);
 	}
 
+
+	@SuppressWarnings("unchecked")
+	static <I, O> Flux<Message<O>> transformWithFunction(Publisher<Message<I>> publisher,
+			Function<? super Flux<Message<I>>, ? extends Publisher<O>> fluxFunction) {
+
+		return Flux.from(publisher)
+				.flatMap(message ->
+						Mono.subscriberContext()
+								.map(ctx -> {
+									ctx.get(RequestMessageHolder.class).set(message);
+									return message;
+								}))
+				.transform(fluxFunction)
+				.flatMap(data ->
+						data instanceof Message<?>
+								? Mono.just((Message<O>) data)
+								: Mono.subscriberContext()
+										.map(ctx -> ctx.get(RequestMessageHolder.class).get())
+										.map(requestMessage ->
+												MessageBuilder.withPayload(data)
+														.copyHeaders(requestMessage.getHeaders())
+														.build()))
+				.subscriberContext(ctx -> ctx.put(RequestMessageHolder.class, new RequestMessageHolder()));
+	}
+
+	private static class RequestMessageHolder extends AtomicReference<Message<?>> {
+
+	}
+
 }

spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java

@@ -33,6 +33,7 @@ import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Level;
+import java.util.stream.Collectors;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -101,7 +102,7 @@ public class ReactiveStreamsTests {
 		this.messageSource.start();
 		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		String[] strings = results.toArray(new String[results.size()]);
-		assertArrayEquals(new String[] {"A", "B", "C", "D", "E", "F"}, strings);
+		assertArrayEquals(new String[] { "A", "B", "C", "D", "E", "F" }, strings);
 		this.messageSource.stop();
 	}
 
@@ -172,6 +173,40 @@ public class ReactiveStreamsTests {
 		}
 	}
 
+	@Test
+	public void testFluxTransform() {
+		QueueChannel resultChannel = new QueueChannel();
+
+		IntegrationFlow integrationFlow = f -> f
+				.split()
+				.<String, String>fluxTransform(flux -> flux
+						.map(Message::getPayload)
+						.map(String::toUpperCase))
+				.aggregate(a -> a
+						.outputProcessor(group -> group
+								.getMessages()
+								.stream()
+								.map(Message::getPayload)
+								.map(String.class::cast)
+								.collect(Collectors.joining(","))))
+				.channel(resultChannel);
+
+		IntegrationFlowContext.IntegrationFlowRegistration integrationFlowRegistration =
+				this.integrationFlowContext
+						.registration(integrationFlow)
+						.register();
+
+		MessageChannel inputChannel = integrationFlowRegistration.getInputChannel();
+		inputChannel.send(new GenericMessage<>("a,b,c,d,e"));
+
+		Message<?> receive = resultChannel.receive(10_000);
+
+		assertNotNull(receive);
+		assertEquals("A,B,C,D,E", receive.getPayload());
+
+		integrationFlowRegistration.destroy();
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class ContextConfiguration {