From 4456caffa16d3b7c8e9838937ec0c236591e5791 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 8 Sep 2021 17:14:29 -0400 Subject: [PATCH] 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 Co-authored-by: Gary Russell --- .../config/ConsumerEndpointFactoryBean.java | 4 + .../dsl/BaseIntegrationFlowDefinition.java | 12 + .../integration/dsl/IntegrationFlow.java | 13 +- .../dsl/IntegrationFlowAdapter.java | 7 +- .../integration/dsl/IntegrationFlows.java | 61 +++++ .../IntegrationFlowBeanPostProcessor.java | 3 +- .../IntegrationFlowLifecycleAdvice.java | 8 +- .../IntegrationFlowCompositionTests.java | 213 ++++++++++++++++++ src/reference/asciidoc/dsl.adoc | 55 +++++ src/reference/asciidoc/whats-new.adoc | 7 + 10 files changed, 379 insertions(+), 4 deletions(-) create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/dsl/composition/IntegrationFlowCompositionTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index aa2f49b045..b1922c1980 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java index 14fac1d5e2..9ebdb15eb0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java @@ -2917,6 +2917,18 @@ public abstract class BaseIntegrationFlowDefinition the expected {@code payload} type diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlow.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlow.java index d009ad7a2f..97a5b667a1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlow.java +++ b/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 getIntegrationComponents() { + return null; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowAdapter.java index 229fc18881..6e46d18ce4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowAdapter.java +++ b/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 getIntegrationComponents() { + return this.targetIntegrationFlow.getIntegrationComponents(); + } + @Override public void start() { assertTargetIntegrationFlow(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java index 1d87d6a4d2..ccf59eb92d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java +++ b/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> 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 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 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() { } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowBeanPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowBeanPostProcessor.java index 7d40802a55..8e2649e822 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowBeanPostProcessor.java +++ b/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", diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowLifecycleAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowLifecycleAdvice.java index fc27e7b74f..e98fe8dec0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowLifecycleAdvice.java +++ b/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(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/composition/IntegrationFlowCompositionTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/composition/IntegrationFlowCompositionTests.java new file mode 100644 index 0000000000..a120688993 --- /dev/null +++ b/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 + .transform(String::toUpperCase) + .to(otherFlow); + } + + @Bean + IntegrationFlow otherFlow() { + return f -> f + .transform(p -> p + " from other flow") + .channel(c -> c.queue("otherFlowResultChannel")); + } + + @Bean + IntegrationFlow requestReplyMainFlow(IntegrationFlow templateFlow) { + return f -> f + .transform(String::toLowerCase) + .to(templateFlow); + } + + @Bean + IntegrationFlow templateFlow() { + return f -> f + .transform("Reply for: "::concat); + } + + @Bean + IntegrationFlow templateSourceFlow() { + return IntegrationFlows.fromSupplier(() -> "test data") + .channel("sourceChannel") + .get(); + } + + @Bean + IntegrationFlow compositionMainFlow(IntegrationFlow templateSourceFlow) { + return IntegrationFlows.from(templateSourceFlow) + .transform(String::toUpperCase) + .channel(c -> c.queue("compositionMainFlowResult")) + .get(); + } + + @Bean + IntegrationFlow firstFlow() { + return f -> f + .transform(p -> p + ", and first flow"); + } + + @Bean + IntegrationFlow middleFlow(IntegrationFlow firstFlow, IntegrationFlow lastFlow) { + return IntegrationFlows.from(firstFlow) + .transform(p -> p + ", and middle flow") + .to(lastFlow); + } + + @Bean + IntegrationFlow lastFlow() { + return f -> f + .transform(p -> p + ", and last flow") + .channel(c -> c.queue("lastFlowResult")); + } + + } + +} diff --git a/src/reference/asciidoc/dsl.adoc b/src/reference/asciidoc/dsl.adoc index 8705c04f3a..878e128e5d 100644 --- a/src/reference/asciidoc/dsl.adoc +++ b/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 ``, 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) + .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 + .transform(String::toUpperCase) + .to(otherFlow); +} + +@Bean +IntegrationFlow otherFlow() { + return f -> f + .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. \ No newline at end of file diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index d0b0122d68..cea47a09e7 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/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