diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/DslIntegrationConfigurationInitializer.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/DslIntegrationConfigurationInitializer.java index e7700be..2aaeeb4 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/DslIntegrationConfigurationInitializer.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/DslIntegrationConfigurationInitializer.java @@ -16,13 +16,24 @@ package org.springframework.integration.dsl; +import java.util.Collection; import java.util.Map; import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactoryUtils; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.parsing.BeanComponentDefinition; +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.beans.factory.support.GenericBeanDefinition; +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.config.ConsumerEndpointFactoryBean; import org.springframework.integration.config.IntegrationConfigurationInitializer; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.dsl.config.InstanceBeanDefinition; import org.springframework.integration.dsl.core.Spec; +import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; /** @@ -34,24 +45,87 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig @Override public void initialize(ConfigurableListableBeanFactory configurableListableBeanFactory) throws BeansException { + Assert.isInstanceOf(BeanDefinitionRegistry.class, configurableListableBeanFactory, + "To use Spring Integration Java DSL the 'beanFactory' has to be an instance of 'BeanDefinitionRegistry'." + + "Consider using 'GenericApplicationContext' implementation."); + this.initializeIntegrationFlows(configurableListableBeanFactory); this.populateBeansFromSpecs(configurableListableBeanFactory); - configurableListableBeanFactory.addBeanPostProcessor(new IntegrationFlowBeanPostProcessor(configurableListableBeanFactory)); + } + + private void initializeIntegrationFlows(ConfigurableListableBeanFactory beanFactory) { + Map integrationFlows = beanFactory.getBeansOfType(IntegrationFlow.class, false, false); + BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory; + for (Map.Entry integrationFlowEntry : integrationFlows.entrySet()) { + String flowName = integrationFlowEntry.getKey(); + String flowNamePrefix = flowName + ":"; + IntegrationFlow flow = integrationFlowEntry.getValue(); + int channelNameIndex = 0; + for (AbstractBeanDefinition beanDefinition : flow.getIntegrationComponents()) { + if (beanDefinition instanceof InstanceBeanDefinition) { + final Object instance = beanDefinition.getSource(); + Collection values = beanFactory.getBeansOfType(instance.getClass(), false, false).values(); + if (!values.contains(instance)) { + if (instance instanceof AbstractMessageChannel) { + String channelBeanName = ((AbstractMessageChannel) instance).getComponentName(); + if (channelBeanName == null) { + channelBeanName = flowNamePrefix + "channel" + BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++; + } + registry.registerBeanDefinition(channelBeanName, beanDefinition); + } + else if (instance instanceof EndpointSpec) { + EndpointSpec endpointSpec = (EndpointSpec) instance; + MessageHandler messageHandler = endpointSpec.getHandler(); + ConsumerEndpointFactoryBean endpoint = endpointSpec.getEndpoint(); + String id = endpointSpec.getId(); + + String handlerBeanName = generateInstanceBeanDefinitionName(registry, messageHandler); + String[] handlerAlias = id != null ? new String[]{id + IntegrationNamespaceUtils.HANDLER_ALIAS_SUFFIX} : null; + BeanComponentDefinition definitionHolder = new BeanComponentDefinition(new InstanceBeanDefinition(messageHandler), handlerBeanName, handlerAlias); + BeanDefinitionReaderUtils.registerBeanDefinition(definitionHolder, registry); + + String endpointBeanName = id; + if (endpointBeanName == null) { + endpointBeanName = generateInstanceBeanDefinitionName(registry, endpoint); + } + registry.registerBeanDefinition(endpointBeanName, new InstanceBeanDefinition(endpoint)); + } + else { + String beanName = generateInstanceBeanDefinitionName(registry, instance); + registry.registerBeanDefinition(beanName, beanDefinition); + } + } + } + else { + BeanDefinitionReaderUtils.registerWithGeneratedName(beanDefinition, registry); + } + } + registry.removeBeanDefinition(flowName); + beanFactory.destroyBean(flowName); + } + } private void populateBeansFromSpecs(ConfigurableListableBeanFactory beanFactory) { - Assert.isInstanceOf(BeanDefinitionRegistry.class, beanFactory, - "To use Spring Integration Java DSL the 'beanFactory' has to be an instance of 'BeanDefinitionRegistry'." + - "Consider using 'GenericApplicationContext' implementation." - ); Map specs = beanFactory.getBeansOfType(Spec.class, false, false); BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory; for (Map.Entry specEntry : specs.entrySet()) { String id = specEntry.getKey(); Spec spec = specEntry.getValue(); registry.removeBeanDefinition(id); - beanFactory.destroyBean(id); beanFactory.registerSingleton(id, spec.get()); + beanFactory.initializeBean(spec.get(), id); } } + @SuppressWarnings("serial") + private static String generateInstanceBeanDefinitionName(BeanDefinitionRegistry registry, final Object instance) { + return BeanDefinitionReaderUtils.generateBeanName(new GenericBeanDefinition() { + + @Override + public String getBeanClassName() { + return instance.getClass().getName(); + } + }, registry); + } + } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java deleted file mode 100644 index 30d448a..0000000 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java +++ /dev/null @@ -1,106 +0,0 @@ -package org.springframework.integration.dsl; - -import java.util.Collection; - -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.BeanFactoryUtils; -import org.springframework.beans.factory.config.BeanPostProcessor; -import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; -import org.springframework.beans.factory.parsing.BeanComponentDefinition; -import org.springframework.beans.factory.support.AbstractBeanDefinition; -import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; -import org.springframework.beans.factory.support.BeanDefinitionRegistry; -import org.springframework.beans.factory.support.GenericBeanDefinition; -import org.springframework.integration.channel.AbstractMessageChannel; -import org.springframework.integration.config.ConsumerEndpointFactoryBean; -import org.springframework.integration.config.xml.IntegrationNamespaceUtils; -import org.springframework.integration.dsl.config.InstanceBeanDefinition; -import org.springframework.messaging.MessageHandler; -import org.springframework.util.Assert; - -/** - * @author Artem Bilan - * @since 4.0 - */ -public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor { - - private final ConfigurableListableBeanFactory beanFactory; - - private final BeanDefinitionRegistry registry; - - - public IntegrationFlowBeanPostProcessor(ConfigurableListableBeanFactory beanFactory) { - Assert.isInstanceOf(BeanDefinitionRegistry.class, beanFactory, - "To use Spring Integration Java DSL the 'beanFactory' has to be an instance of 'BeanDefinitionRegistry'." + - "Consider using 'GenericApplicationContext' implementation." - ); - this.beanFactory = beanFactory; - this.registry = (BeanDefinitionRegistry) beanFactory; - } - - @Override - public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { - if (bean instanceof IntegrationFlow) { - String flowNamePrefix = beanName + ":"; - int channelNameIndex = 0; - for (AbstractBeanDefinition beanDefinition : ((IntegrationFlow) bean).getIntegrationComponents()) { - if (beanDefinition instanceof InstanceBeanDefinition) { - final Object instance = beanDefinition.getSource(); - Collection values = this.beanFactory.getBeansOfType(instance.getClass(), false, false).values(); - if (!values.contains(instance)) { - if (instance instanceof AbstractMessageChannel) { - String channelBeanName = ((AbstractMessageChannel) instance).getComponentName(); - if (channelBeanName == null) { - channelBeanName = flowNamePrefix + "channel" + BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++; - } - registry.registerBeanDefinition(channelBeanName, beanDefinition); - } - else if (instance instanceof EndpointSpec) { - EndpointSpec endpointSpec = (EndpointSpec) instance; - MessageHandler messageHandler = endpointSpec.getHandler(); - ConsumerEndpointFactoryBean endpoint = endpointSpec.getEndpoint(); - String id = endpointSpec.getId(); - - String handlerBeanName = generateInstanceBeanDefinitionName(registry, messageHandler); - String[] handlerAlias = id != null ? new String[]{id + IntegrationNamespaceUtils.HANDLER_ALIAS_SUFFIX} : null; - BeanComponentDefinition definitionHolder = new BeanComponentDefinition( - new InstanceBeanDefinition(messageHandler), handlerBeanName, handlerAlias); - BeanDefinitionReaderUtils.registerBeanDefinition(definitionHolder, registry); - - String endpointBeanName = id; - if (endpointBeanName == null) { - endpointBeanName = generateInstanceBeanDefinitionName(registry, endpoint); - } - registry.registerBeanDefinition(endpointBeanName, new InstanceBeanDefinition(endpoint)); - } - else { - String name = generateInstanceBeanDefinitionName(registry, instance); - registry.registerBeanDefinition(name, beanDefinition); - } - } - } - else { - BeanDefinitionReaderUtils.registerWithGeneratedName(beanDefinition, registry); - } - } - } - return bean; - } - - @Override - public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { - return bean; - } - - @SuppressWarnings("serial") - private static String generateInstanceBeanDefinitionName(BeanDefinitionRegistry registry, final Object instance) { - return BeanDefinitionReaderUtils.generateBeanName(new GenericBeanDefinition() { - - @Override - public String getBeanClassName() { - return instance.getClass().getName(); - } - }, registry); - } - -} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java index cbd6483..fd4a6d0 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java @@ -28,7 +28,6 @@ import org.springframework.integration.dsl.support.EndpointConfigurer; import org.springframework.integration.filter.ExpressionEvaluatingSelector; import org.springframework.integration.filter.MessageFilter; import org.springframework.integration.filter.MethodInvokingSelector; -import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.transformer.ExpressionEvaluatingTransformer; import org.springframework.integration.transformer.GenericTransformer; import org.springframework.integration.transformer.MessageTransformingHandler; @@ -117,9 +116,6 @@ public final class IntegrationFlowBuilder { if (this.currentComponent instanceof MessageProducer) { ((MessageProducer) this.currentComponent).setOutputChannel(outputChannel); } - if (this.currentComponent instanceof AbstractReplyProducingMessageHandler) { - ((AbstractReplyProducingMessageHandler) this.currentComponent).setOutputChannel(outputChannel); - } else if (this.currentComponent instanceof SourcePollingChannelAdapterFactoryBean) { ((SourcePollingChannelAdapterFactoryBean) this.currentComponent).setOutputChannel(outputChannel); } 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 8d25af4..c0af239 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 @@ -22,7 +22,6 @@ import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; -import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; import org.aopalliance.aop.Advice; @@ -35,8 +34,11 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.Lifecycle; import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; import org.springframework.integration.MessageDispatchingException; +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; @@ -122,16 +124,6 @@ public class IntegrationFlowTests { @EnableIntegration public static class ContextConfiguration { - @Bean - public DirectChannelSpec inputChannel() { - return MessageChannels.direct(); - } - - @Bean - public QueueChannelSpec successChannel() { - return MessageChannels.queue(); - } - @Bean public MessageSource integerMessageSource() { MethodInvokingMessageSource source = new MethodInvokingMessageSource(); @@ -156,37 +148,50 @@ public class IntegrationFlowTests { } @Configuration + @ComponentScan public static class ContextConfiguration2 { - @Autowired - @Qualifier("inputChannel") - private DirectChannel inputChannel; + @Bean + public QueueChannelSpec successChannel() { + return MessageChannels.queue(); + } - @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); + advice.setSuccessChannel(this.successChannel().get()); 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(), c -> c.autoStartup(false).id("payloadSerializingTransformer")) .channel(MessageChannels.queue(new SimpleMessageStore(), "fooQueue")) .transform(new PayloadDeserializingTransformer()) - .channel(MessageChannels.executor("executor", Executors.newCachedThreadPool())) + .channel(MessageChannels.publishSubscribe("publishSubscribeChannel")) .transform((Integer p) -> p * 2, c -> c.advice(this.expressionAdvice())) .get(); } } + @MessageEndpoint + public static class AnnotationTestService { + + @ServiceActivator(inputChannel = "publishSubscribeChannel") + public void handle(Object payload) { + assertEquals(100, payload); + } + } + + }