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 9b80669956..74a6abad6d 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 @@ -18,11 +18,15 @@ package org.springframework.integration.dsl; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; +import java.util.function.Supplier; + +import org.reactivestreams.Publisher; import org.springframework.context.SmartLifecycle; import org.springframework.integration.core.MessageSource; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.gateway.MessagingGatewaySupport; +import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; @@ -178,6 +182,27 @@ public abstract class IntegrationFlowAdapter implements IntegrationFlow, SmartLi return IntegrationFlows.from(service, methodName, endpointConfigurer); } + protected IntegrationFlowBuilder from(Supplier messageSource) { + return IntegrationFlows.from(messageSource); + } + + protected IntegrationFlowBuilder from(Supplier messageSource, + Consumer endpointConfigurer) { + return IntegrationFlows.from(messageSource, endpointConfigurer); + } + + protected IntegrationFlowBuilder from(Class serviceInterface) { + return IntegrationFlows.from(serviceInterface); + } + + protected IntegrationFlowBuilder from(Class serviceInterface, String beanName) { + return IntegrationFlows.from(serviceInterface, beanName); + } + + protected IntegrationFlowBuilder from(Publisher> publisher) { + return IntegrationFlows.from(publisher); + } + protected abstract IntegrationFlowDefinition buildFlow(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java index c7d3bb6253..c57cf6d284 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java @@ -20,6 +20,8 @@ import java.util.Collection; import java.util.LinkedHashMap; import java.util.Map; +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.aop.support.NameMatchMethodPointcutAdvisor; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanCreationNotAllowedException; import org.springframework.beans.factory.BeanFactory; @@ -33,6 +35,7 @@ import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.context.SmartLifecycle; import org.springframework.core.io.DescriptiveResource; import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.channel.DirectChannel; @@ -250,11 +253,44 @@ public class IntegrationFlowBeanPostProcessor return flow; } + /** + * Only invoked for {@link IntegrationFlow} instances that are not + * {@link StandardIntegrationFlow}s; typically lambdas. Creates a new + * {@link StandardIntegrationFlow} with an input channel named {@code beanName.input} + * and the flow defined by the flow parameter. If the flow is not an + * {@link IntegrationFlowAdapter} the original, user-provided {@link IntegrationFlow} + * is wrapped in a proxy and advised with a {@link IntegrationFlowLifecycleAdvice}; + * see its javadocs for more information. + */ private Object processIntegrationFlowImpl(IntegrationFlow flow, String beanName) { IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(beanName + ".input"); + flow.configure(flowBuilder); - Object standardIntegrationFlow = processStandardIntegrationFlow(flowBuilder.get(), beanName); - return isLambda(flow) ? standardIntegrationFlow : flow; + + StandardIntegrationFlow target = flowBuilder.get(); + processStandardIntegrationFlow(target, beanName); + + if (!(flow instanceof IntegrationFlowAdapter)) { + NameMatchMethodPointcutAdvisor integrationFlowAdvice = + new NameMatchMethodPointcutAdvisor(new IntegrationFlowLifecycleAdvice(target)); + integrationFlowAdvice.setMappedNames( + "getInputChannel", + "start", + "stop", + "isRunning", + "isAutoStartup", + "getPhase"); + + ProxyFactory proxyFactory = new ProxyFactory(flow); + proxyFactory.addAdvisor(integrationFlowAdvice); + if (!(flow instanceof SmartLifecycle)) { + proxyFactory.addInterface(SmartLifecycle.class); + } + return proxyFactory.getProxy(this.beanFactory.getBeanClassLoader()); + } + else { + return flow; + } } private void processIntegrationComponentSpec(IntegrationComponentSpec bean) { @@ -321,9 +357,4 @@ public class IntegrationFlowBeanPostProcessor return id; } - private static boolean isLambda(Object o) { - Class aClass = o.getClass(); - return aClass.isSynthetic() && !aClass.isAnonymousClass() && !aClass.isLocalClass(); - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowLifecycleAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowLifecycleAdvice.java new file mode 100644 index 0000000000..4a075e919f --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowLifecycleAdvice.java @@ -0,0 +1,115 @@ +/* + * Copyright 2018 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 + * + * http://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; + +import org.aopalliance.intercept.MethodInterceptor; +import org.aopalliance.intercept.MethodInvocation; + +import org.springframework.context.SmartLifecycle; +import org.springframework.util.ObjectUtils; + +/** + * An AOP {@link MethodInterceptor} for the {@link IntegrationFlow} proxies + * with a delegation to an associated {@link StandardIntegrationFlow} instance, which + * is not exposed as a bean during a target {@link IntegrationFlow} bean processing. + * + *

In most cases an associated internal {@link StandardIntegrationFlow} + * exposes an {@code inputChannel} bean for the target {@link IntegrationFlow}, + * which doesn't start from the channel, e.g. instantiated from lambda. + * This way the advice first tries to obtain an {@code inputChannel} from the + * target {@link IntegrationFlow} and then falls back to the {@link #delegate}. + * + *

Another aspect of this advice is to control and delegate {@link SmartLifecycle} + * of the target {@link IntegrationFlow} and associated {@link #delegate}. + * The {@link SmartLifecycle#start()} and {@link SmartLifecycle#stop()} operations + * are delegated to the {@link StandardIntegrationFlow} as is because that instance + * isn't controlled by the standard application context lifecycle. + * The {@link SmartLifecycle#isAutoStartup()}, {@link SmartLifecycle#getPhase()} + * and {@link SmartLifecycle#isRunning()} are called on the {@link #delegate} + * only in case when {@link MethodInvocation#proceed()} returns {@code null} + * or isn't called at all, e.g. when target {@link IntegrationFlow} doesn't + * implement {@link SmartLifecycle}. + * + * @author Artem Bilan + * + * @since 5.1 + */ +class IntegrationFlowLifecycleAdvice implements MethodInterceptor { + + private final StandardIntegrationFlow delegate; + + IntegrationFlowLifecycleAdvice(StandardIntegrationFlow delegate) { + this.delegate = delegate; + } + + @Override + public Object invoke(MethodInvocation invocation) throws Throwable { + Object target = invocation.getThis(); + String method = invocation.getMethod().getName(); + Object result = null; + + if ("getInputChannel".equals(method)) { + result = invocation.proceed(); + if (result == null) { + result = this.delegate.getInputChannel(); + } + } + else { + if (target instanceof SmartLifecycle) { + result = invocation.proceed(); + } + + switch (method) { + + case "start": + this.delegate.start(); + break; + + case "stop": + Object[] arguments = invocation.getArguments(); + if (!ObjectUtils.isEmpty(arguments)) { + this.delegate.stop((Runnable) arguments[0]); + } + else { + this.delegate.stop(); + } + break; + + case "isRunning": + if (result == null) { + result = this.delegate.isRunning(); + } + break; + + case "isAutoStartup": + if (result == null) { + result = this.delegate.isAutoStartup(); + } + break; + + case "getPhase": + if (result == null) { + result = this.delegate.getPhase(); + } + break; + } + } + + return result; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/RouterSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/RouterSpec.java index 4a0913f9b7..496c27d46d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/RouterSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/RouterSpec.java @@ -143,6 +143,14 @@ public final class RouterSpec * Add a subflow as an alternative to a {@link #channelMapping(Object, String)}. * {@link #prefix(String)} and {@link #suffix(String)} cannot be used when subflow * mappings are used. + *

If subflow should refer to the external {@link IntegrationFlow} bean and + * there is a requirement to expect reply from there, such a reference should be + * wrapped with a {@code .gateway()}: + * 

+	 * {@code
+	 *     .subFlowMapping(false, sf -> sf.gateway(evenFlow())))
+	 * }
+	 *
* @param key the key. * @param subFlow the subFlow. * @return the router spec. diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java index 95d3661609..9868d2ffd4 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java @@ -16,8 +16,11 @@ package org.springframework.integration.dsl.flowservices; +import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import java.util.Collection; import java.util.Collections; @@ -30,11 +33,15 @@ import java.util.stream.Collectors; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.aop.framework.Advised; +import org.springframework.aop.support.AopUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.SmartLifecycle; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; +import org.springframework.core.Ordered; import org.springframework.integration.annotation.Aggregator; import org.springframework.integration.annotation.CorrelationStrategy; import org.springframework.integration.annotation.Filter; @@ -70,21 +77,29 @@ import org.springframework.util.StringUtils; @DirtiesContext public class FlowServiceTests { + @Autowired(required = false) + @Qualifier("flowServiceTests.MyFlow") + private IntegrationFlow myFlow; + @Autowired @Qualifier("flowServiceTests.MyFlow.input") private MessageChannel input; - @Autowired(required = false) - private MyFlow myFlow; - @Autowired private PollableChannel myFlowAdapterOutput; @Test - public void testFlowServiceAndLogAsLastNoError() { + public void testFlowServiceAndLogAsLastNoError() throws Exception { assertNotNull(this.myFlow); + assertTrue(AopUtils.isAopProxy(this.myFlow)); + assertThat(this.myFlow, instanceOf(Advised.class)); + assertThat(this.myFlow, instanceOf(Ordered.class)); + assertThat(this.myFlow, instanceOf(SmartLifecycle.class)); + this.input.send(MessageBuilder.withPayload("foo").build()); - Object result = this.myFlow.resultOverLoggingHandler.get(); + + MyFlow myFlow = (MyFlow) ((Advised) this.myFlow).getTargetSource().getTarget(); + Object result = myFlow.resultOverLoggingHandler.get(); assertNotNull(result); assertEquals("FOO", result); } @@ -134,7 +149,7 @@ public class FlowServiceTests { } @Component - public static class MyFlow implements IntegrationFlow { + public static class MyFlow implements IntegrationFlow, Ordered { private final AtomicReference resultOverLoggingHandler = new AtomicReference<>(); @@ -147,6 +162,11 @@ public class FlowServiceTests { }); } + @Override + public int getOrder() { + return 0; + } + } @Component diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java index eafd8fe395..e744e486ac 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java @@ -204,32 +204,6 @@ public class ManualFlowTests { @Test public void testWrongLifecycle() { - - class MyIntegrationFlow implements IntegrationFlow { - - @Override - public void configure(IntegrationFlowDefinition flow) { - flow.bridge(); - } - - } - - IntegrationFlow testFlow = new MyIntegrationFlow(); - - // This is fine because we are not going to start it automatically. - assertNotNull(this.integrationFlowContext.registration(testFlow) - .autoStartup(false) - .register()); - - try { - this.integrationFlowContext.registration(testFlow).register(); - fail("IllegalStateException expected"); - } - catch (Exception e) { - assertThat(e, instanceOf(IllegalStateException.class)); - assertThat(e.getMessage(), containsString("Consider to implement it for [" + testFlow + "].")); - } - try { this.integrationFlowContext.remove("foo"); fail("IllegalStateException expected"); diff --git a/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt b/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt index 97e909ec6e..6b12b5cf09 100644 --- a/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt +++ b/spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt @@ -25,14 +25,17 @@ 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.QueueChannel import org.springframework.integration.config.EnableIntegration import org.springframework.integration.dsl.IntegrationFlow +import org.springframework.integration.support.MessageBuilder import org.springframework.messaging.MessageChannel import org.springframework.messaging.PollableChannel import org.springframework.messaging.support.GenericMessage import org.springframework.test.annotation.DirtiesContext import org.springframework.test.context.junit.jupiter.SpringJUnitConfig + /** * @author Artem Bilan * @@ -66,6 +69,30 @@ class RouterDslTests { } + @Autowired + @Qualifier("splitRouteAggregate.input") + private lateinit var splitRouteAggregateInput: MessageChannel + + @Test + fun `route to two subflows using them as bean references`() { + + val replyChannel = QueueChannel() + val message = MessageBuilder.withPayload(arrayOf(1, 2, 3)) + .setReplyChannel(replyChannel) + .build() + + this.splitRouteAggregateInput.send(message) + + val receive = replyChannel.receive(10000) + + val payload = receive?.payload + + assert(payload).isNotNull { + it.isInstanceOf(List::class.java) + it.isEqualTo(listOf("even", "odd", "even")) + } + } + @Configuration @EnableIntegration class Config { @@ -83,6 +110,29 @@ class RouterDslTests { .channel { c -> c.queue("routerTwoSubFlowsOutput") } } + @Bean + fun splitRouteAggregate() = + IntegrationFlow { f -> + f.split() + .route({ o -> o % 2 == 0 }, + { m -> + m.subFlowMapping(true) { sf -> sf.gateway(oddFlow()) } + .subFlowMapping(false) { sf -> sf.gateway(evenFlow()) } + }) + .aggregate() + } + + @Bean + fun oddFlow() = + IntegrationFlow { flow -> + flow.handle { _, _ -> "odd" } + } + + @Bean + fun evenFlow() = + IntegrationFlow { flow -> + flow.handle { _, _ -> "even" } + } } } diff --git a/src/reference/asciidoc/dsl.adoc b/src/reference/asciidoc/dsl.adoc index 0090c0af11..82d70a24af 100644 --- a/src/reference/asciidoc/dsl.adoc +++ b/src/reference/asciidoc/dsl.adoc @@ -578,6 +578,8 @@ The result of this definition is the same bunch of Integration components wired Only limitation is here, that this flow is started with named direct channel - `lambdaFlow.input`. And Lambda flow can't start from `MessageSource` or `MessageProducer`. +Starting _version 5.1_, this kind of `IntegrationFlow` are wrapped to the proxy for exposing lifecycle control and provide access to the `inputChannel` of the internally associated `StandardIntegrationFlow`. + [[java-dsl-function-expression]] === FunctionExpression @@ -643,6 +645,42 @@ public IntegrationFlow routeFlow() { The `.channelMapping()` continues to work as in regular `Router` mapping, but the `.subFlowMapping()` tied that subflow with main flow. In other words, any router's subflow returns to the main flow after `.route()`. +[IMPORTANT] +===== +Sometimes it is necessary to refer to an existing `IntegrationFlow` `@Bean` from the `.subFlowMapping()`: +[source,java] +---- +@Bean +public IntegrationFlow splitRouteAggregate() { + return f -> f + .split() + .route(o -> o % 2 == 0, + m -> m + .subFlowMapping(true, oddFlow()) + .subFlowMapping(false, sf -> sf.gateway(evenFlow()))) + .aggregate(); +} + +@Bean +public IntegrationFlow oddFlow() { + return f -> f.handle(m -> System.out.println("odd")); +} + +@Bean +public IntegrationFlow evenFlow() { + return f -> f.handle((p, h) -> "even"); +} +---- + +In this case, when you need to receive a reply from such a sub-flow and continue the main flow, this `IntegrationFlow` bean reference (or its input channel) has to be wrapped with a `.gateway()` as shown above. +The `oddFlow()` referece in the sample above is not wrapped to the `.gateway()` - therefore we don't expect a reply from this routing branch. +Otherwise you end up with an exception like: +.... +Caused by: org.springframework.beans.factory.BeanCreationException: The 'currentComponent' (org.springframework.integration.router.MethodInvokingRouter@7965a51c) is a one-way 'MessageHandler' and it isn't appropriate to configure 'outputChannel'. This is the end of the integration flow. +.... +When a sub-flow is configured as a lambda, the Framework handles the request-reply interaction with the sub-flow and a gateway is not needed. +===== + Of course, subflows can be nested with any depth, but we don't recommend to do that because, in fact, even in the router case, adding complex subflows within a flow would quickly begin to look like a plate of spaghetti and difficult for a human to parse. [[java-dsl-protocol-adapters]]