Introduce high-level API for flows composition (#3624)

* Introduce high-level API for flows composition

For better end-user experience and more smooth integration logic
decomposition and distribution introduce an `IntegrationFlows.from(IntegrationFlow)`
to let to start the current flow from existing one.
On the other hand introduce an `BaseIntegrationFlowDefinition.to(IntegrationFlow)`
to let to continue the flow logic in the other existing one.
This way we can extract some templating logic into separate `IntegrationFlow` definitions
allowing at the same time to decompose a complex flow definition into logical reusable parts

* * Add more tests

* * Fix Checkstyle violation
* Add `@SuppressWarnings("overloads")` to new `from(IntegrationFlow)` and existing `from(Publisher)`.
Technically it does not make sense since `PublisherIntegrationFlow` is not a `public` class

* * Add docs
* Fix language in JavaDocs according review

* Fix language in the docs after review

Co-authored-by: Gary Russell <grussell@vmware.com>

Co-authored-by: Gary Russell <grussell@vmware.com>

spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java

@@ -137,6 +137,10 @@ public class ConsumerEndpointFactoryBean
 		}
 	}
 
+	public MessageHandler getHandler() {
+		return this.handler;
+	}
+
 	public void setInputChannel(MessageChannel inputChannel) {
 		this.inputChannel = inputChannel;
 	}

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

@@ -2917,6 +2917,18 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
 				.get();
 	}
 
+	/**
+	 * Finish this flow with delegation to other {@link IntegrationFlow} instance.
+	 * @param other the {@link IntegrationFlow} to compose with.
+	 * @return The {@link IntegrationFlow} instance based on this definition.
+	 * @since 5.5.4
+	 */
+	public IntegrationFlow to(IntegrationFlow other) {
+		MessageChannel otherFlowInputChannel = obtainInputChannelFromFlow(other);
+		return channel(otherFlowInputChannel)
+				.get();
+	}
+
 	/**
 	 * Represent an Integration Flow as a Reactive Streams {@link Publisher} bean.
 	 * @param <T> the expected {@code payload} type

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -16,6 +16,8 @@
 
 package org.springframework.integration.dsl;
 
+import java.util.Map;
+
 import org.springframework.messaging.MessageChannel;
 
 /**
@@ -92,4 +94,13 @@ public interface IntegrationFlow {
 		return null;
 	}
 
+	/**
+	 * Return a map of integration components managed by this flow (if any).
+	 * @return the map of integration components managed by this flow.
+	 * @since 5.5.4
+	 */
+	default Map<Object, String> getIntegrationComponents() {
+		return null;
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2020 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -16,6 +16,7 @@
 
 package org.springframework.integration.dsl;
 
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
@@ -79,6 +80,10 @@ public abstract class IntegrationFlowAdapter implements IntegrationFlow, Managea
 		return this.targetIntegrationFlow.getInputChannel();
 	}
 
+	@Override public Map<Object, String> getIntegrationComponents() {
+		return this.targetIntegrationFlow.getIntegrationComponents();
+	}
+
 	@Override
 	public void start() {
 		assertTargetIntegrationFlow();

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

@@ -16,22 +16,29 @@
 
 package org.springframework.integration.dsl;
 
+import java.util.Map;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
 import org.reactivestreams.Publisher;
 
+import org.springframework.aop.framework.Advised;
+import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.FluxMessageChannel;
+import org.springframework.integration.channel.PublishSubscribeChannel;
+import org.springframework.integration.config.ConsumerEndpointFactoryBean;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype;
 import org.springframework.integration.dsl.support.MessageChannelReference;
 import org.springframework.integration.endpoint.AbstractMessageSource;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.gateway.MessagingGatewaySupport;
+import org.springframework.integration.handler.AbstractMessageProducingHandler;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
 import org.springframework.util.Assert;
 
 /**
@@ -328,12 +335,52 @@ public final class IntegrationFlows {
 	 * @param publisher the {@link Publisher} to subscribe to.
 	 * @return new {@link IntegrationFlowBuilder}.
 	 */
+	@SuppressWarnings("overloads")
 	public static IntegrationFlowBuilder from(Publisher<? extends Message<?>> publisher) {
 		FluxMessageChannel reactiveChannel = new FluxMessageChannel();
 		reactiveChannel.subscribeTo(publisher);
 		return from((MessageChannel) reactiveChannel);
 	}
 
+	/**
+	 * Start the flow with a composition from the {@link IntegrationFlow}.
+	 * @param other the {@link IntegrationFlow} from which to compose.
+	 * @return new {@link IntegrationFlowBuilder}.
+	 * @since 5.5.4
+	 */
+	@SuppressWarnings("overloads")
+	public static IntegrationFlowBuilder from(IntegrationFlow other) {
+		Map<Object, String> integrationComponents = other.getIntegrationComponents();
+		Assert.notNull(integrationComponents, () ->
+				"The provided integration flow to compose from '" + other +
+						"' must be declared as a bean in the application context");
+		Object lastIntegrationComponentFromOther =
+				integrationComponents.keySet().stream().reduce((prev, next) -> next).orElse(null);
+		if (lastIntegrationComponentFromOther instanceof MessageChannel) {
+			return from((MessageChannel) lastIntegrationComponentFromOther);
+		}
+		else if (lastIntegrationComponentFromOther instanceof ConsumerEndpointFactoryBean) {
+			MessageHandler handler = ((ConsumerEndpointFactoryBean) lastIntegrationComponentFromOther).getHandler();
+			handler = extractProxyTarget(handler);
+			if (handler instanceof AbstractMessageProducingHandler) {
+				return buildFlowFromOutputChannel((AbstractMessageProducingHandler) handler);
+			}
+			lastIntegrationComponentFromOther = handler; // for the exception message below
+		}
+		throw new BeanCreationException("The 'IntegrationFlow' to start from must end with " +
+				"a 'MessageChannel' or reply-producing endpoint to let the result from that flow to be " +
+				"processed in this instance. The provided flow ends with: " + lastIntegrationComponentFromOther);
+	}
+
+	private static IntegrationFlowBuilder buildFlowFromOutputChannel(AbstractMessageProducingHandler handler) {
+		MessageChannel outputChannel = handler.getOutputChannel();
+		if (outputChannel == null) {
+			outputChannel = new PublishSubscribeChannel();
+			handler.setOutputChannel(outputChannel);
+		}
+		return from(outputChannel);
+	}
+
 	private static IntegrationFlowBuilder from(MessagingGatewaySupport inboundGateway,
 			@Nullable IntegrationFlowBuilder integrationFlowBuilderArg) {
 
@@ -360,6 +407,20 @@ public final class IntegrationFlows {
 		return null;
 	}
 
+	@SuppressWarnings("unchecked")
+	private static <T> T extractProxyTarget(T target) {
+		if (!(target instanceof Advised)) {
+			return target;
+		}
+		Advised advised = (Advised) target;
+		try {
+			return (T) extractProxyTarget(advised.getTargetSource().getTarget());
+		}
+		catch (Exception e) {
+			throw new BeanCreationException("Could not extract target", e);
+		}
+	}
+
 	private IntegrationFlows() {
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2020 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -321,6 +321,7 @@ public class IntegrationFlowBeanPostProcessor
 					new NameMatchMethodPointcutAdvisor(new IntegrationFlowLifecycleAdvice(target));
 			integrationFlowAdvice.setMappedNames(
 					"getInputChannel",
+					"getIntegrationComponents",
 					"start",
 					"stop",
 					"isRunning",

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2021 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.
@@ -75,6 +75,12 @@ class IntegrationFlowLifecycleAdvice implements MethodInterceptor {
 				result = this.delegate.getInputChannel();
 			}
 		}
+		else if ("getIntegrationComponents".equals(method)) {
+			result = invocation.proceed();
+			if (result == null) {
+				result = this.delegate.getIntegrationComponents();
+			}
+		}
 		else {
 			if (target instanceof SmartLifecycle) {
 				result = invocation.proceed();

spring-integration-core/src/test/java/org/springframework/integration/dsl/composition/IntegrationFlowCompositionTests.java

@@ -0,0 +1,213 @@
+/*
+ * Copyright 2021 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.integration.dsl.composition;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.Pollers;
+import org.springframework.integration.dsl.context.IntegrationFlowContext;
+import org.springframework.integration.scheduling.PollerMetadata;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 5.5.4
+ */
+@SpringJUnitConfig
+@DirtiesContext
+public class IntegrationFlowCompositionTests {
+
+	@Autowired
+	IntegrationFlowContext integrationFlowContext;
+
+	@Autowired
+	@Qualifier("mainFlow.input")
+	DirectChannel mainFlowInput;
+
+	@Autowired
+	QueueChannel otherFlowResultChannel;
+
+	@Test
+	void testToOperator() {
+		this.mainFlowInput.send(new GenericMessage<>("hello"));
+		Message<?> receive = this.otherFlowResultChannel.receive(10_000);
+		assertThat(receive).isNotNull()
+				.extracting(Message::getPayload)
+				.isEqualTo("HELLO from other flow");
+	}
+
+	@Autowired
+	@Qualifier("requestReplyMainFlow.input")
+	DirectChannel requestReplyMainFlowInput;
+
+	@Test
+	void testToWithRequestReply() {
+		QueueChannel replyChannel = new QueueChannel();
+		this.requestReplyMainFlowInput.send(
+				MessageBuilder.withPayload("TEST")
+						.setReplyChannel(replyChannel)
+						.build());
+		Message<?> receive = replyChannel.receive(10_000);
+		assertThat(receive).isNotNull()
+				.extracting(Message::getPayload)
+				.isEqualTo("Reply for: test");
+	}
+
+	@Autowired
+	QueueChannel compositionMainFlowResult;
+
+	@Test
+	void testFromComposition() {
+		Message<?> receive = this.compositionMainFlowResult.receive(10_000);
+		assertThat(receive).isNotNull()
+				.extracting(Message::getPayload)
+				.isEqualTo("TEST DATA");
+
+		receive = this.compositionMainFlowResult.receive(10_000);
+		assertThat(receive).isNotNull()
+				.extracting(Message::getPayload)
+				.isEqualTo("TEST DATA");
+	}
+
+	@Autowired
+	@Qualifier("firstFlow.input")
+	DirectChannel firstFlowInput;
+
+	@Autowired
+	QueueChannel lastFlowResult;
+
+	@Test
+	void testFromToComposition() {
+		this.firstFlowInput.send(new GenericMessage<>("start"));
+
+		Message<?> receive = this.lastFlowResult.receive(10_000);
+		assertThat(receive).isNotNull()
+				.extracting(Message::getPayload)
+				.isEqualTo("start, and first flow, and middle flow, and last flow");
+	}
+
+	@Test
+	void testInvalidStartFlowForComposition() {
+		IntegrationFlow startFlow = f -> f.handle(m -> { });
+
+		assertThatIllegalArgumentException()
+				.isThrownBy(() -> IntegrationFlows.from(startFlow))
+				.withMessageContaining("must be declared as a bean in the application context");
+
+		IntegrationFlowContext.IntegrationFlowRegistration startRegistration =
+				this.integrationFlowContext.registration(startFlow).register();
+
+		assertThatExceptionOfType(BeanCreationException.class)
+				.isThrownBy(() -> IntegrationFlows.from(startRegistration.getIntegrationFlow()))
+				.withMessageContaining("The 'IntegrationFlow' to start from must end with " +
+						"a 'MessageChannel' or reply-producing endpoint");
+
+	}
+
+	@Configuration
+	@EnableIntegration
+	public static class ContextConfiguration {
+
+		@Bean(PollerMetadata.DEFAULT_POLLER)
+		PollerMetadata defaultPoller() {
+			return Pollers.fixedDelay(100).get();
+		}
+
+		@Bean
+		IntegrationFlow mainFlow(IntegrationFlow otherFlow) {
+			return f -> f
+					.<String, String>transform(String::toUpperCase)
+					.to(otherFlow);
+		}
+
+		@Bean
+		IntegrationFlow otherFlow() {
+			return f -> f
+					.<String, String>transform(p -> p + " from other flow")
+					.channel(c -> c.queue("otherFlowResultChannel"));
+		}
+
+		@Bean
+		IntegrationFlow requestReplyMainFlow(IntegrationFlow templateFlow) {
+			return f -> f
+					.<String, String>transform(String::toLowerCase)
+					.to(templateFlow);
+		}
+
+		@Bean
+		IntegrationFlow templateFlow() {
+			return f -> f
+					.<String, String>transform("Reply for: "::concat);
+		}
+
+		@Bean
+		IntegrationFlow templateSourceFlow() {
+			return IntegrationFlows.fromSupplier(() -> "test data")
+					.channel("sourceChannel")
+					.get();
+		}
+
+		@Bean
+		IntegrationFlow compositionMainFlow(IntegrationFlow templateSourceFlow) {
+			return IntegrationFlows.from(templateSourceFlow)
+					.<String, String>transform(String::toUpperCase)
+					.channel(c -> c.queue("compositionMainFlowResult"))
+					.get();
+		}
+
+		@Bean
+		IntegrationFlow firstFlow() {
+			return f -> f
+					.<String, String>transform(p -> p + ", and first flow");
+		}
+
+		@Bean
+		IntegrationFlow middleFlow(IntegrationFlow firstFlow, IntegrationFlow lastFlow) {
+			return IntegrationFlows.from(firstFlow)
+					.<String, String>transform(p -> p + ", and middle flow")
+					.to(lastFlow);
+		}
+
+		@Bean
+		IntegrationFlow lastFlow() {
+			return f -> f
+					.<String, String>transform(p -> p + ", and last flow")
+					.channel(c -> c.queue("lastFlowResult"));
+		}
+
+	}
+
+}

src/reference/asciidoc/dsl.adoc

@@ -1378,3 +1378,58 @@ public IntegrationFlow customFlowDefinition() {
 }
 ----
 ====
+
+[[integration-flows-composition]]
+=== Integration Flows Composition
+
+With the `MessageChannel` abstraction as a first class citizen in Spring Integration, the composition of integration flows was always assumed.
+The input channel of any endpoint in the flow can be used to send messages from any other endpoint and not only from the one which has this channel as an output.
+Furthermore, with a `@MessagingGateway` contract, Content Enricher components, composite endpoints like a `<chain>`, and now with `IntegrationFlow` beans (e.g. `IntegrationFlowAdapter`), it is straightforward enough to distribute the business logic between shorter, reusable parts.
+All that is needed for the final composition is knowledge about a `MessageChannel` to send to or receive from.
+
+Starting with version `5.5.4`, to abstract more from `MessageChannel` and hide implementation details from the end-user, the `IntegrationFlows` introduces the `from(IntegrationFlow)` factory method to allow starting the current `IntegrationFlow` from the output of an existing flow:
+
+====
+[source,java]
+----
+@Bean
+IntegrationFlow templateSourceFlow() {
+    return IntegrationFlows.fromSupplier(() -> "test data")
+            .channel("sourceChannel")
+            .get();
+}
+
+@Bean
+IntegrationFlow compositionMainFlow(IntegrationFlow templateSourceFlow) {
+    return IntegrationFlows.from(templateSourceFlow)
+            .<String, String>transform(String::toUpperCase)
+            .channel(c -> c.queue("compositionMainFlowResult"))
+            .get();
+}
+----
+====
+
+On the other hand,  the `IntegrationFlowDefinition` has added a `to(IntegrationFlow)` terminal operator to continue the current flow at the input channel of some other flow:
+
+====
+[source,java]
+----
+@Bean
+IntegrationFlow mainFlow(IntegrationFlow otherFlow) {
+    return f -> f
+            .<String, String>transform(String::toUpperCase)
+            .to(otherFlow);
+}
+
+@Bean
+IntegrationFlow otherFlow() {
+    return f -> f
+            .<String, String>transform(p -> p + " from other flow")
+            .channel(c -> c.queue("otherFlowResultChannel"));
+}
+----
+====
+
+The composition in the middle of the flow is simply achievable with an existing `gateway(IntegrationFlow)` EIP-method.
+This way we can build flows with any complexity by composing them from simpler, reusable logical blocks.
+For example, you may add a library of `IntegrationFlow` beans as a dependency and it is just enough to have their configuration classes imported to the final project and autowired for your `IntegrationFlow` definitions.

src/reference/asciidoc/whats-new.adoc

@@ -47,6 +47,13 @@ See <<./aggregator.adoc#aggregator,Aggregator>> for more information.
 The `MessageGroup` abstraction can be supplied with a `condition` to evaluate later on to make a decision for the group.
 See <<./message-store.adoc#message-group-condition,Message Group Condition>> for more information.
 
+[[x5.5-integration-flows-composition]]
+==== Integration Flows Composition
+
+The new `IntegrationFlows.from(IntegrationFlow)` factory method has been added to allow starting the current `IntegrationFlow` from the output of an existing flow.
+In addition, the `IntegrationFlowDefinition` has added a `to(IntegrationFlow)` terminal operator to continue the current flow at the input channel of some other flow.
+See <<./dsl.adoc#integration-flows-composition,Integration Flows Composition>> for more information.
+
 [[x5.5-amqp]]
 ==== AMQP Changes