From abab1af3d5ada37fac46eca8ade8267bb9172c1d Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 19 Feb 2014 11:24:51 +0200 Subject: [PATCH] Support @Autowired for Configurations with DSL --- ...slIntegrationConfigurationInitializer.java | 58 ++++++++++------- .../dsl/test/IntegrationFlowTests.java | 63 ++++++++++++++----- 2 files changed, 82 insertions(+), 39 deletions(-) diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java index 7f505e6..5e28cb9 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/DslIntegrationConfigurationInitializer.java @@ -16,11 +16,17 @@ package org.springframework.integration.dsl.core; +import java.util.Arrays; import java.util.Collection; -import java.util.Map; +import java.util.HashSet; +import java.util.List; +import java.util.Set; import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.BeanFactoryUtils; +import org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor; +import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.parsing.BeanComponentDefinition; import org.springframework.beans.factory.support.AbstractBeanDefinition; @@ -49,21 +55,35 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig "To use Spring Integration Java DSL the 'beanFactory' has to be an instance of 'BeanDefinitionRegistry'." + "Consider using 'GenericApplicationContext' implementation." ); + this.checkSpecBeans(configurableListableBeanFactory); this.initializeIntegrationFlows(configurableListableBeanFactory); - this.populateBeansFromSpecs(configurableListableBeanFactory); + } + + private void checkSpecBeans(ConfigurableListableBeanFactory beanFactory) { + List specBeanNames = Arrays.asList(beanFactory.getBeanNamesForType(IntegrationComponentSpec.class, false, false)); + if (!specBeanNames.isEmpty()) { + throw new BeanCreationException("'IntegrationComponentSpec' beans: '" + specBeanNames + "' must be populated " + + "to target objects via 'get()' method call. It is important for @Autowired injections."); + } } private void initializeIntegrationFlows(ConfigurableListableBeanFactory beanFactory) { - Map integrationFlows = beanFactory.getBeansOfType(IntegrationFlow.class, false, false); + AutowiredAnnotationBeanPostProcessor autowiredAnnotationBeanPostProcessor = beanFactory.getBean(AutowiredAnnotationBeanPostProcessor.class); BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory; - for (Map.Entry integrationFlowEntry : integrationFlows.entrySet()) { - String flowName = integrationFlowEntry.getKey(); + String[] integrationFlowBeanNames = beanFactory.getBeanNamesForType(IntegrationFlow.class, false, false); + Set processedConfigurations = new HashSet(); + for (String flowName : integrationFlowBeanNames) { + BeanDefinition flowBeanDefinition = beanFactory.getBeanDefinition(flowName); + String configurationBeanName = flowBeanDefinition.getFactoryBeanName(); + if (processedConfigurations.add(configurationBeanName)) { + autowiredAnnotationBeanPostProcessor.processInjection(beanFactory.getBean(configurationBeanName)); + } String flowNamePrefix = flowName + ":"; - IntegrationFlow flow = integrationFlowEntry.getValue(); + IntegrationFlow flow = beanFactory.getBean(flowName, IntegrationFlow.class); int channelNameIndex = 0; - for (AbstractBeanDefinition beanDefinition : flow.getIntegrationComponents()) { - if (beanDefinition instanceof InstanceBeanDefinition) { - final Object instance = beanDefinition.getSource(); + for (AbstractBeanDefinition component : flow.getIntegrationComponents()) { + if (component instanceof InstanceBeanDefinition) { + final Object instance = component.getSource(); Collection values = beanFactory.getBeansOfType(instance.getClass(), false, false).values(); if (!values.contains(instance)) { if (instance instanceof AbstractMessageChannel) { @@ -71,7 +91,7 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig if (channelBeanName == null) { channelBeanName = flowNamePrefix + "channel" + BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++; } - registry.registerBeanDefinition(channelBeanName, beanDefinition); + registry.registerBeanDefinition(channelBeanName, component); } else if (instance instanceof ConsumerEndpointSpec) { ConsumerEndpointSpec endpointSpec = (ConsumerEndpointSpec) instance; @@ -97,30 +117,20 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig } else { String beanName = generateInstanceBeanDefinitionName(registry, instance); - registry.registerBeanDefinition(beanName, beanDefinition); + registry.registerBeanDefinition(beanName, component); } } } else { - BeanDefinitionReaderUtils.registerWithGeneratedName(beanDefinition, registry); + BeanDefinitionReaderUtils.registerWithGeneratedName(component, registry); } } - registry.removeBeanDefinition(flowName); - beanFactory.destroyBean(flowName); +// registry.removeBeanDefinition(flowName); +// beanFactory.destroyBean(flowName); } } - private void populateBeansFromSpecs(ConfigurableListableBeanFactory beanFactory) { - Map specs = beanFactory.getBeansOfType(IntegrationComponentSpec.class, false, false); - BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory; - for (Map.Entry specEntry : specs.entrySet()) { - String id = specEntry.getKey(); - IntegrationComponentSpec spec = (IntegrationComponentSpec) specEntry.getValue(); - registry.registerBeanDefinition(id, new InstanceBeanDefinition(spec.get())); - } - } - @SuppressWarnings("serial") private static String generateInstanceBeanDefinitionName(BeanDefinitionRegistry registry, final Object instance) { return BeanDefinitionReaderUtils.generateBeanName(new GenericBeanDefinition() { diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java index fe74cc8..1309bca 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java @@ -53,8 +53,6 @@ import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.channel.DirectChannelSpec; import org.springframework.integration.dsl.channel.MessageChannels; -import org.springframework.integration.dsl.channel.QueueChannelSpec; -import org.springframework.integration.dsl.support.PollerSpec; import org.springframework.integration.dsl.support.Pollers; import org.springframework.integration.endpoint.MethodInvokingMessageSource; import org.springframework.integration.event.core.MessagingEvent; @@ -69,6 +67,7 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.transformer.PayloadDeserializingTransformer; import org.springframework.integration.transformer.PayloadSerializingTransformer; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessageHeaders; @@ -209,6 +208,7 @@ public class IntegrationFlowTests { public void testWrongLastComponent() { try { new AnnotationConfigApplicationContext(InvalidLastComponentFlowContext.class); + fail("BeanCreationException expected"); } catch (Exception e) { assertThat(e, Matchers.instanceOf(BeanCreationException.class)); @@ -245,6 +245,19 @@ public class IntegrationFlowTests { assertEquals("Hello, world", receive.getPayload()); } + @Test + public void testWrongConfigurationWithSpecBean() { + try { + new AnnotationConfigApplicationContext(InvalidConfigurationWithSpec.class); + fail("BeanCreationException expected"); + } + catch (Exception e) { + assertThat(e, Matchers.instanceOf(IllegalArgumentException.class)); + assertThat(e.getCause(), Matchers.instanceOf(BeanCreationException.class)); + assertThat(e.getCause().getMessage(), Matchers.containsString("must be populated to target objects via 'get()' method call")); + } + } + @Configuration @EnableIntegration @@ -267,8 +280,13 @@ public class IntegrationFlowTests { } @Bean(name = PollerMetadata.DEFAULT_POLLER_METADATA_BEAN_NAME) - public PollerSpec poller() { - return Pollers.fixedRate(500); + public PollerMetadata poller() { + return Pollers.fixedRate(500).get(); + } + + @Bean + public DirectChannel inputChannel() { + return MessageChannels.direct().get(); } } @@ -277,27 +295,26 @@ public class IntegrationFlowTests { @ComponentScan public static class ContextConfiguration2 { - @Bean - public QueueChannelSpec successChannel() { - return MessageChannels.queue(); - } + @Autowired + @Qualifier("inputChannel") + private MessageChannel inputChannel; + + @Autowired + @Qualifier("successChannel") + private PollableChannel successChannel; - @Bean - public DirectChannelSpec inputChannel() { - return MessageChannels.direct(); - } @Bean public Advice expressionAdvice() { ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice(); advice.setOnSuccessExpression("payload"); - advice.setSuccessChannel(this.successChannel().get()); + advice.setSuccessChannel(this.successChannel); return advice; } @Bean public IntegrationFlow flow2() { - return IntegrationFlows.from(this.inputChannel()) + return IntegrationFlows.from(this.inputChannel) .filter(p -> p instanceof String, c -> c.id("filter")) .transform(Integer::parseInt) .transform(new PayloadSerializingTransformer(), @@ -323,6 +340,11 @@ public class IntegrationFlowTests { @Configuration public static class ContextConfiguration3 { + @Bean + public QueueChannel successChannel() { + return MessageChannels.queue().get(); + } + @Bean public AtomicReference eventHolder() { return new AtomicReference<>(); @@ -377,7 +399,8 @@ public class IntegrationFlowTests { .handle(this.fileWritingMessageHandler(), c -> { FileWritingMessageHandler handler = c.get().getT2(); handler.setFileNameGenerator(message -> null); - handler.setExpectReply(false); }) + handler.setExpectReply(false); + }) .get(); } @@ -410,4 +433,14 @@ public class IntegrationFlowTests { } + @EnableIntegration + public static class InvalidConfigurationWithSpec { + + @Bean + public DirectChannelSpec invalidBean() { + return MessageChannels.direct(); + } + + } + }