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 e744bb7e25..ae85455e6d 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 @@ -16,7 +16,6 @@ package org.springframework.integration.dsl.context; -import java.util.Collection; import java.util.LinkedHashMap; import java.util.Map; @@ -34,10 +33,12 @@ import org.springframework.beans.factory.SmartInitializingSingleton; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.BeanDefinitionCustomizer; import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.beans.factory.config.ConfigurableBeanFactory; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.config.EmbeddedValueResolver; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.BeanDefinitionOverrideException; import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; @@ -163,7 +164,7 @@ public class IntegrationFlowBeanPostProcessor id = flowNamePrefix + id; } - if (noBeanPresentForComponent(messageHandler)) { + if (noBeanPresentForComponent(messageHandler, flowBeanName)) { String handlerBeanName = generateBeanName(messageHandler, flowNamePrefix); registerComponent(messageHandler, handlerBeanName, flowBeanName); @@ -174,7 +175,7 @@ public class IntegrationFlowBeanPostProcessor targetIntegrationComponents.put(endpoint, id); } else { - if (noBeanPresentForComponent(component)) { + if (noBeanPresentForComponent(component, flowBeanName)) { if (component instanceof AbstractMessageChannel) { String channelBeanName = ((AbstractMessageChannel) component).getComponentName(); if (channelBeanName == null) { @@ -211,7 +212,7 @@ public class IntegrationFlowBeanPostProcessor if (!CollectionUtils.isEmpty(componentsToRegister)) { componentsToRegister.entrySet() .stream() - .filter(o -> noBeanPresentForComponent(o.getKey())) + .filter(o -> noBeanPresentForComponent(o.getKey(), flowBeanName)) .forEach(o -> registerComponent(o.getKey(), generateBeanName(o.getKey(), flowNamePrefix, o.getValue(), @@ -231,7 +232,7 @@ public class IntegrationFlowBeanPostProcessor targetIntegrationComponents.put(pollingChannelAdapterFactoryBean, id); MessageSource messageSource = spec.get().getT2(); - if (noBeanPresentForComponent(messageSource)) { + if (noBeanPresentForComponent(messageSource, flowBeanName)) { String messageSourceId = id + ".source"; if (messageSource instanceof NamedComponent && ((NamedComponent) messageSource).getComponentName() != null) { @@ -277,7 +278,14 @@ public class IntegrationFlowBeanPostProcessor } } else { - targetIntegrationComponents.put(entry.getKey(), entry.getValue()); + Object componentToUse = entry.getKey(); + String beanNameToUse = entry.getValue(); + if (StringUtils.hasText(beanNameToUse) && + ConfigurableBeanFactory.SCOPE_PROTOTYPE.equals( + this.beanFactory.getBeanDefinition(beanNameToUse).getScope())) { + this.beanFactory.initializeBean(componentToUse, beanNameToUse); + } + targetIntegrationComponents.put(component, beanNameToUse); } } } @@ -337,7 +345,7 @@ public class IntegrationFlowBeanPostProcessor componentsToRegister.entrySet() .stream() - .filter(component -> noBeanPresentForComponent(component.getKey())) + .filter(component -> noBeanPresentForComponent(component.getKey(), beanName)) .forEach(component -> registerComponent(component.getKey(), generateBeanName(component.getKey(), component.getValue()))); @@ -378,17 +386,36 @@ public class IntegrationFlowBeanPostProcessor } } - private boolean noBeanPresentForComponent(Object instance) { + @SuppressWarnings("unchecked") + private boolean noBeanPresentForComponent(Object instance, String parentBeanName) { if (instance instanceof NamedComponent) { String beanName = ((NamedComponent) instance).getComponentName(); if (beanName != null) { - return !this.beanFactory.containsBean(beanName); + if (this.beanFactory.containsBean(beanName)) { + BeanDefinition existingBeanDefinition = this.beanFactory.getBeanDefinition(beanName); + if (!ConfigurableBeanFactory.SCOPE_PROTOTYPE.equals(existingBeanDefinition.getScope()) + && !instance.equals(this.beanFactory.getBean(beanName))) { + + AbstractBeanDefinition beanDefinition = + BeanDefinitionBuilder.genericBeanDefinition((Class) instance.getClass(), + () -> instance) + .getBeanDefinition(); + beanDefinition.setResourceDescription("the '" + parentBeanName + "' bean definition"); + throw new BeanDefinitionOverrideException(beanName, beanDefinition, existingBeanDefinition); + } + else { + return false; + } + } + else { + return true; + } } } - Collection beans = this.beanFactory.getBeansOfType(instance.getClass(), false, false).values(); - - return !beans.contains(instance); + return !this.beanFactory.getBeansOfType(instance.getClass(), false, false) + .values() + .contains(instance); } private void registerComponent(Object component, String beanName) { 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 363d688391..db332bf3a4 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 @@ -16,6 +16,7 @@ package org.springframework.integration.dsl.manualflow; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.lessThan; @@ -46,6 +47,7 @@ import org.hamcrest.Matchers; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.BeanCreationNotAllowedException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.DisposableBean; @@ -53,11 +55,13 @@ import org.springframework.beans.factory.ListableBeanFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.config.BeanFactoryPostProcessor; import org.springframework.beans.factory.config.ConfigurableBeanFactory; +import org.springframework.beans.factory.support.BeanDefinitionOverrideException; import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Scope; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.config.EnableIntegrationManagement; @@ -241,7 +245,8 @@ public class ManualFlowTests { assertThat(e.getMessage(), containsString("The 'receive()/receiveAndConvert()' isn't supported")); } - assertThat(this.beanFactory.getBeanNamesForType(MessageTransformingHandler.class)[0], startsWith(flowId + ".")); + assertThat(this.beanFactory.getBeanNamesForType(MessageTransformingHandler.class)[0], + startsWith(flowId + ".")); flowRegistration.destroy(); @@ -516,6 +521,18 @@ public class ManualFlowTests { flowRegistrations.forEach(IntegrationFlowRegistration::destroy); } + + @Test + public void testDisabledBeansOverride() { + assertThatThrownBy( + () -> this.integrationFlowContext + .registration(f -> f.channel(c -> c.direct("doNotOverrideChannel"))) + .register()) + .isExactlyInstanceOf(BeanCreationException.class) + .hasCauseExactlyInstanceOf(BeanDefinitionOverrideException.class) + .hasMessageContaining("Invalid bean definition with name 'doNotOverrideChannel'"); + } + @Configuration @EnableIntegration @EnableMessageHistory @@ -533,6 +550,12 @@ public class ManualFlowTests { return new Date(); } + + @Bean + public MessageChannel doNotOverrideChannel() { + return new DirectChannel(); + } + } private static class MyFlowAdapter extends IntegrationFlowAdapter { @@ -582,7 +605,7 @@ public class ManualFlowTests { } @Override - public void destroy() throws Exception { + public void destroy() { this.destroyed = true; } diff --git a/src/reference/asciidoc/dsl.adoc b/src/reference/asciidoc/dsl.adoc index 6527c37ab1..8922f1be85 100644 --- a/src/reference/asciidoc/dsl.adoc +++ b/src/reference/asciidoc/dsl.adoc @@ -115,28 +115,38 @@ The endpoints are automatically wired together by using direct channels. [[java-dsl-class-cast]] .Lambdas And `Message` Arguments -IMPORTANT: When using lambdas in EIP methods, the "input" argument is generally the message payload. +[IMPORTANT] +==== +When using lambdas in EIP methods, the "input" argument is generally the message payload. If you wish to access the entire message, use one of the overloaded methods that take a `Class` as the first parameter. For example, this won't work: -==== [source, java] ---- ., Foo>transform(m -> newFooFromMessage(m)) ---- -==== This will fail at runtime with a `ClassCastException` because the lambda doesn't retain the argument type and the framework will attempt to cast the payload to a `Message`. Instead, use: -==== [source, java] ---- .(Message.class, m -> newFooFromMessage(m)) ---- ==== +[[bean-definitions-override]] +.Bean Definitions override +[IMPORTANT] +==== +The Java DSL can register beans for the object defined in-line in the flow definition, as well as can reuse existing, injected beans. +In case of the same bean name defined for in-line object and existing bean definition, a `BeanDefinitionOverrideException` is thrown indicating that such a configuration is wrong. +However when you deal with `prototype` beans, there is no way to detect from the integration flow processor an existing bean definition because every time we call a `prototype` bean from the `BeanFactory` we get a new instance. +This way a provided instance is used in the `IntegrationFlow` as is without any bean registration and any possible check against existing `prototype` bean definition. +However `BeanFactory.initializeBean()` is called for this object if it has an explicit `id` and bean definition for this name is in `prototype` scope. +==== + [[java-dsl-channels]] === Message Channels @@ -316,7 +326,7 @@ It avoids inconvenient coding using setters and makes the flow definition more s Note that you can use `Transformers` to declare target `Transformer` instances as `@Bean` instances and, again, use them from `IntegrationFlow` definition as bean methods. Nevertheless, the DSL parser takes care of bean declarations for inline objects, if they are not yet defined as beans. -See [https://docs.spring.io/spring-integration/api/org/springframework/integration/dsl/Transformers.html] in the Javadoc for more information and supported factory methods. +See https://docs.spring.io/spring-integration/api/org/springframework/integration/dsl/Transformers.html[Transformers] in the Javadoc for more information and supported factory methods. Also see <>. @@ -789,15 +799,18 @@ public IntegrationFlow evenFlow() { } ---- +{empty} + 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 in the preceding example. The `oddFlow()` reference in the preceding example is not wrapped to the `.gateway()`. Therefore, we do not expect a reply from this routing branch. Otherwise, you end up with an exception similar to the following: -[source] ----- -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. ----- +.... +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 you configure a sub-flow as a lambda, the framework handles the request-reply interaction with the sub-flow and a gateway is not needed. ====