From dd085c8945f67a2ac1c86c6cf8172a3f65bf06f4 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 6 Jun 2014 00:36:53 +0300 Subject: [PATCH] DSL: Fix `@Autowired` issue for `IntegrationFlow`s * Move DSL parser logic from `BFPP` to `BPP` as soon as the DSL doesn't populate any `BeanDefinition` - just real objects, which can be registered with `this.beanFactory.registerSingleton(beanName, component);` * Add more DSL for `Amqp` 'Namespace' factory and provide tests on the matter * Comment out `spring-integration-jdbc` and `spring-integration-jpa` dependecies because the provide `spring-jdbc` dependency, but it causes an issue in Boot: https://github.com/spring-projects/spring-boot/issues/1041 --- spring-integration-java-dsl/build.gradle | 4 +- .../integration/dsl/IntegrationFlow.java | 11 +- .../integration/dsl/IntegrationFlows.java | 5 + .../integration/dsl/amqp/Amqp.java | 45 ++++- .../amqp/AmqpInboundChannelAdapterSpec.java | 169 ++++++++++++++++++ .../dsl/amqp/AmqpOutboundEndpointSpec.java | 112 ++++++++++++ ...slIntegrationConfigurationInitializer.java | 126 ++----------- .../IntegrationFlowBeanPostProcessor.java | 148 +++++++++++++++ .../dsl/core/MessagingProducerSpec.java | 62 +++++++ .../dsl/test/IntegrationFlowTests.java | 58 +++++- 10 files changed, 606 insertions(+), 134 deletions(-) create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpInboundChannelAdapterSpec.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpOutboundEndpointSpec.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/MessagingProducerSpec.java diff --git a/spring-integration-java-dsl/build.gradle b/spring-integration-java-dsl/build.gradle index de85b58..e2da5b1 100644 --- a/spring-integration-java-dsl/build.gradle +++ b/spring-integration-java-dsl/build.gradle @@ -54,10 +54,10 @@ dependencies { , 'spring-integration-ftp' , 'spring-integration-gemfire' , 'spring-integration-http' - , 'spring-integration-jdbc' +// , 'spring-integration-jdbc' , 'spring-integration-jms' , 'spring-integration-jmx' - , 'spring-integration-jpa' +// , 'spring-integration-jpa' , 'spring-integration-mail' , 'spring-integration-mongodb' , 'spring-integration-mqtt' diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlow.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlow.java index e8316a4..de75671 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlow.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlow.java @@ -19,27 +19,22 @@ package org.springframework.integration.dsl; import java.util.LinkedHashSet; import java.util.Set; -import org.springframework.beans.factory.support.AbstractBeanDefinition; -import org.springframework.integration.dsl.config.InstanceBeanDefinition; - /** * @author Artem Bilan */ public final class IntegrationFlow { - private final Set integrationComponents = new LinkedHashSet(); + private final Set integrationComponents = new LinkedHashSet(); IntegrationFlow() { } - public Set getIntegrationComponents() { + public Set getIntegrationComponents() { return integrationComponents; } IntegrationFlow addComponent(Object component) { - AbstractBeanDefinition beanDefinition = component instanceof AbstractBeanDefinition - ? (AbstractBeanDefinition) component : new InstanceBeanDefinition(component); - this.integrationComponents.add(beanDefinition); + this.integrationComponents.add(component); return this; } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java index 47f791e..4ce57f7 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java @@ -22,6 +22,7 @@ import org.springframework.integration.config.SourcePollingChannelAdapterFactory import org.springframework.integration.core.MessageSource; import org.springframework.integration.dsl.channel.MessageChannelSpec; import org.springframework.integration.dsl.core.MessagingGatewaySpec; +import org.springframework.integration.dsl.core.MessagingProducerSpec; import org.springframework.integration.dsl.support.EndpointConfigurer; import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype; import org.springframework.integration.dsl.support.MessageChannelReference; @@ -79,6 +80,10 @@ public final class IntegrationFlows { .currentComponent(sourcePollingChannelAdapterFactoryBean); } + public static IntegrationFlowBuilder from(MessagingProducerSpec messagingProducerSpec) { + return from(messagingProducerSpec.get()); + } + public static IntegrationFlowBuilder from(MessageProducerSupport messageProducer) { DirectFieldAccessor dfa = new DirectFieldAccessor(messageProducer); MessageChannel outputChannel = (MessageChannel) dfa.getPropertyValue("outputChannel"); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java index cf4e342..14998d3 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java @@ -16,32 +16,65 @@ package org.springframework.integration.dsl.amqp; +import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter; import org.springframework.integration.amqp.inbound.AmqpInboundGateway; +import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint; import org.springframework.integration.dsl.core.MessagingGatewaySpec; +import org.springframework.integration.dsl.core.MessagingProducerSpec; /** * @author Artem Bilan */ public abstract class Amqp { - public static MessagingGatewaySpec inboundGateway( - SimpleMessageListenerContainer listenerContainer) { - return new AmqpInboundGatewaySpec(listenerContainer); - } - public static AmqpInboundGatewaySpec inboundGateway(ConnectionFactory connectionFactory, String... queueNames) { SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory); listenerContainer.setQueueNames(queueNames); - return new AmqpInboundGatewaySpec(listenerContainer); + return (AmqpInboundGatewaySpec) inboundGateway(listenerContainer); } public static AmqpInboundGatewaySpec inboundGateway(ConnectionFactory connectionFactory, Queue... queues) { SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory); listenerContainer.setQueues(queues); + return (AmqpInboundGatewaySpec) inboundGateway(listenerContainer); + } + + public static MessagingGatewaySpec inboundGateway( + SimpleMessageListenerContainer listenerContainer) { return new AmqpInboundGatewaySpec(listenerContainer); } + public static AmqpInboundChannelAdapterSpec inboundAdapter(ConnectionFactory connectionFactory, String... queueNames) { + SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory); + listenerContainer.setQueueNames(queueNames); + return (AmqpInboundChannelAdapterSpec) inboundAdapter(listenerContainer); + } + + public static AmqpInboundChannelAdapterSpec inboundAdapter(ConnectionFactory connectionFactory, Queue... queues) { + SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory); + listenerContainer.setQueues(queues); + return (AmqpInboundChannelAdapterSpec) inboundAdapter(listenerContainer); + } + + public static MessagingProducerSpec inboundAdapter( + SimpleMessageListenerContainer listenerContainer) { + return new AmqpInboundChannelAdapterSpec(listenerContainer); + } + + public static AmqpOutboundEndpointSpec outboundAdapter(AmqpTemplate amqpTemplate) { + return outboundEndpoint(new AmqpOutboundEndpoint(amqpTemplate), false); + } + + public static AmqpOutboundEndpointSpec outboundGateway(AmqpTemplate amqpTemplate) { + return outboundEndpoint(new AmqpOutboundEndpoint(amqpTemplate), true); + } + + private static AmqpOutboundEndpointSpec outboundEndpoint(AmqpOutboundEndpoint endpoint, boolean expectReply) { + return new AmqpOutboundEndpointSpec(endpoint, expectReply); + } + } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpInboundChannelAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpInboundChannelAdapterSpec.java new file mode 100644 index 0000000..f0cb84f --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpInboundChannelAdapterSpec.java @@ -0,0 +1,169 @@ +/* + * Copyright 2014 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.amqp; + +import java.util.concurrent.Executor; + +import org.aopalliance.aop.Advice; + +import org.springframework.amqp.core.AcknowledgeMode; +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.amqp.support.converter.MessageConverter; +import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter; +import org.springframework.integration.amqp.support.AmqpHeaderMapper; +import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; +import org.springframework.integration.dsl.core.MessagingProducerSpec; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.util.ErrorHandler; + +/** + * @author Artem Bilan + */ +public class AmqpInboundChannelAdapterSpec extends MessagingProducerSpec { + + private final SimpleMessageListenerContainer listenerContainer; + + private final DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper(); + + public AmqpInboundChannelAdapterSpec(SimpleMessageListenerContainer listenerContainer) { + super(new AmqpInboundChannelAdapter(listenerContainer)); + this.listenerContainer = listenerContainer; + this.target.setHeaderMapper(headerMapper); + } + + public AmqpInboundChannelAdapterSpec acknowledgeMode(AcknowledgeMode acknowledgeMode) { + this.listenerContainer.setAcknowledgeMode(acknowledgeMode); + return this; + } + + public AmqpInboundChannelAdapterSpec addQueueNames(String... queueName) { + this.listenerContainer.addQueueNames(queueName); + return this; + } + + public AmqpInboundChannelAdapterSpec addQueues(Queue... queues) { + this.listenerContainer.addQueues(queues); + return this; + } + + public AmqpInboundChannelAdapterSpec errorHandler(ErrorHandler errorHandler) { + this.listenerContainer.setErrorHandler(errorHandler); + return this; + } + + public AmqpInboundChannelAdapterSpec channelTransacted(boolean transactional) { + this.listenerContainer.setChannelTransacted(transactional); + return this; + } + + public AmqpInboundChannelAdapterSpec adviceChain(Advice... adviceChain) { + this.listenerContainer.setAdviceChain(adviceChain); + return this; + } + + public AmqpInboundChannelAdapterSpec recoveryInterval(long recoveryInterval) { + this.listenerContainer.setRecoveryInterval(recoveryInterval); + return this; + } + + public AmqpInboundChannelAdapterSpec concurrentConsumers(int concurrentConsumers) { + this.listenerContainer.setConcurrentConsumers(concurrentConsumers); + return this; + } + + public AmqpInboundChannelAdapterSpec maxConcurrentConsumers(int maxConcurrentConsumers) { + this.listenerContainer.setMaxConcurrentConsumers(maxConcurrentConsumers); + return this; + } + + public AmqpInboundChannelAdapterSpec exclusive(boolean exclusive) { + this.listenerContainer.setExclusive(exclusive); + return this; + } + + public AmqpInboundChannelAdapterSpec startConsumerMinInterval(long startConsumerMinInterval) { + this.listenerContainer.setStartConsumerMinInterval(startConsumerMinInterval); + return this; + } + + public AmqpInboundChannelAdapterSpec stopConsumerMinInterval(long stopConsumerMinInterval) { + this.listenerContainer.setStopConsumerMinInterval(stopConsumerMinInterval); + return this; + } + + public AmqpInboundChannelAdapterSpec consecutiveActiveTrigger(int consecutiveActiveTrigger) { + this.listenerContainer.setConsecutiveActiveTrigger(consecutiveActiveTrigger); + return this; + } + + public AmqpInboundChannelAdapterSpec consecutiveIdleTrigger(int consecutiveIdleTrigger) { + this.listenerContainer.setConsecutiveIdleTrigger(consecutiveIdleTrigger); + return this; + } + + public AmqpInboundChannelAdapterSpec receiveTimeout(long receiveTimeout) { + this.listenerContainer.setReceiveTimeout(receiveTimeout); + return this; + } + + public AmqpInboundChannelAdapterSpec shutdownTimeout(long shutdownTimeout) { + this.listenerContainer.setShutdownTimeout(shutdownTimeout); + return this; + } + + public AmqpInboundChannelAdapterSpec taskExecutor(Executor taskExecutor) { + this.listenerContainer.setTaskExecutor(taskExecutor); + return this; + } + + public AmqpInboundChannelAdapterSpec prefetchCount(int prefetchCount) { + this.listenerContainer.setPrefetchCount(prefetchCount); + return this; + } + + public AmqpInboundChannelAdapterSpec txSize(int txSize) { + this.listenerContainer.setTxSize(txSize); + return this; + } + + public AmqpInboundChannelAdapterSpec transactionManager(PlatformTransactionManager transactionManager) { + this.listenerContainer.setTransactionManager(transactionManager); + return this; + } + + public AmqpInboundChannelAdapterSpec defaultRequeueRejected(boolean defaultRequeueRejected) { + this.listenerContainer.setDefaultRequeueRejected(defaultRequeueRejected); + return this; + } + + public AmqpInboundChannelAdapterSpec messageConverter(MessageConverter messageConverter) { + this.target.setMessageConverter(messageConverter); + return this; + } + + public AmqpInboundChannelAdapterSpec headerMapper(AmqpHeaderMapper headerMapper) { + this.target.setHeaderMapper(headerMapper); + return this; + } + + public AmqpInboundChannelAdapterSpec mappedRequestHeaders(String... headers) { + this.headerMapper.setRequestHeaderNames(headers); + return this; + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpOutboundEndpointSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpOutboundEndpointSpec.java new file mode 100644 index 0000000..c244d74 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpOutboundEndpointSpec.java @@ -0,0 +1,112 @@ +/* + * Copyright 2014 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.amqp; + +import org.springframework.amqp.core.MessageDeliveryMode; +import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint; +import org.springframework.integration.amqp.support.AmqpHeaderMapper; +import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; +import org.springframework.integration.dsl.core.IntegrationComponentSpec; +import org.springframework.messaging.MessageChannel; +import org.springframework.util.Assert; + +/** + * @author Artem Bilan + */ +public class AmqpOutboundEndpointSpec extends IntegrationComponentSpec { + + private final AmqpOutboundEndpoint endpoint; + + private final boolean expectReply; + + private final DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper(); + + AmqpOutboundEndpointSpec(AmqpOutboundEndpoint endpoint, boolean expectReply) { + this.endpoint = endpoint; + this.expectReply = expectReply; + this.endpoint.setExpectReply(expectReply); + this.endpoint.setHeaderMapper(this.headerMapper); + } + + public AmqpOutboundEndpointSpec headerMapper(AmqpHeaderMapper headerMapper) { + endpoint.setHeaderMapper(headerMapper); + return this; + } + + public AmqpOutboundEndpointSpec routingKey(String routingKey) { + endpoint.setRoutingKey(routingKey); + return this; + } + + public AmqpOutboundEndpointSpec defaultDeliveryMode(MessageDeliveryMode defaultDeliveryMode) { + endpoint.setDefaultDeliveryMode(defaultDeliveryMode); + return this; + } + + public AmqpOutboundEndpointSpec exchangeName(String exchangeName) { + endpoint.setExchangeName(exchangeName); + return this; + } + + public AmqpOutboundEndpointSpec routingKeyExpression(String routingKeyExpression) { + endpoint.setRoutingKeyExpression(routingKeyExpression); + return this; + } + + public AmqpOutboundEndpointSpec returnChannel(MessageChannel returnChannel) { + endpoint.setReturnChannel(returnChannel); + return this; + } + + public AmqpOutboundEndpointSpec confirmAckChannel(MessageChannel ackChannel) { + endpoint.setConfirmAckChannel(ackChannel); + return this; + } + + public AmqpOutboundEndpointSpec exchangeNameExpression(String exchangeNameExpression) { + endpoint.setExchangeNameExpression(exchangeNameExpression); + return this; + } + + public AmqpOutboundEndpointSpec confirmNackChannel(MessageChannel nackChannel) { + endpoint.setConfirmNackChannel(nackChannel); + return this; + } + + public AmqpOutboundEndpointSpec confirmCorrelationExpression(String confirmCorrelationExpression) { + endpoint.setConfirmCorrelationExpression(confirmCorrelationExpression); + return this; + } + + public AmqpOutboundEndpointSpec mappedRequestHeaders(String... headers) { + this.headerMapper.setRequestHeaderNames(headers); + return this; + } + + public AmqpOutboundEndpointSpec mappedReplyHeaders(String... headers) { + Assert.isTrue(expectReply, "'mappedReplyHeaders' can be applied on for gateway"); + this.headerMapper.setReplyHeaderNames(headers); + return this; + } + + + @Override + protected AmqpOutboundEndpoint doGet() { + return this.endpoint; + } + +} 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 b275391..4f18767 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 @@ -17,31 +17,14 @@ package org.springframework.integration.dsl.core; import java.util.Arrays; -import java.util.Collection; -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; -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.channel.FixedSubscriberChannel; -import org.springframework.integration.config.ConsumerEndpointFactoryBean; -import org.springframework.integration.config.IntegrationConfigUtils; +import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.integration.config.IntegrationConfigurationInitializer; -import org.springframework.integration.dsl.IntegrationFlow; -import org.springframework.integration.dsl.config.InstanceBeanDefinition; -import org.springframework.integration.dsl.support.MessageChannelReference; -import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; /** @@ -51,14 +34,23 @@ import org.springframework.util.Assert; */ public class DslIntegrationConfigurationInitializer implements IntegrationConfigurationInitializer { + private static final String INTEGRATION_FLOW_BPP_BEAN_NAME = IntegrationFlowBeanPostProcessor + .class.getName(); + @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.checkSpecBeans(configurableListableBeanFactory); - this.initializeIntegrationFlows(configurableListableBeanFactory); + + checkSpecBeans(configurableListableBeanFactory); + + BeanDefinitionRegistry registry = (BeanDefinitionRegistry) configurableListableBeanFactory; + if (!registry.containsBeanDefinition(INTEGRATION_FLOW_BPP_BEAN_NAME)) { + registry.registerBeanDefinition(INTEGRATION_FLOW_BPP_BEAN_NAME, + new RootBeanDefinition(IntegrationFlowBeanPostProcessor.class)); + } } private void checkSpecBeans(ConfigurableListableBeanFactory beanFactory) { @@ -71,98 +63,4 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig } } - private void initializeIntegrationFlows(ConfigurableListableBeanFactory beanFactory) { - AutowiredAnnotationBeanPostProcessor autowiredAnnotationBeanPostProcessor = - beanFactory.getBean(AutowiredAnnotationBeanPostProcessor.class); - BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory; - 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 = beanFactory.getBean(flowName, IntegrationFlow.class); - int channelNameIndex = 0; - 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) { - String channelBeanName = ((AbstractMessageChannel) instance).getComponentName(); - if (channelBeanName == null) { - channelBeanName = flowNamePrefix + "channel" + - BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++; - } - registry.registerBeanDefinition(channelBeanName, component); - } - else if (instance instanceof ConsumerEndpointSpec) { - ConsumerEndpointSpec endpointSpec = (ConsumerEndpointSpec) instance; - MessageHandler messageHandler = endpointSpec.get().getT2(); - ConsumerEndpointFactoryBean endpoint = endpointSpec.get().getT1(); - String id = endpointSpec.getId(); - - Collection messageHandlers = - beanFactory.getBeansOfType(messageHandler.getClass(), false, false).values(); - - if (!messageHandlers.contains(messageHandler)) { - String handlerBeanName = generateInstanceBeanDefinitionName(registry, messageHandler); - String[] handlerAlias = id != null - ? new String[]{id + IntegrationConfigUtils.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 if (instance instanceof MessageChannelReference) { - String channelName = ((MessageChannelReference) instance).getName(); - if (!registry.containsBeanDefinition(channelName)) { - IntegrationConfigUtils.autoCreateDirectChannel(channelName, registry); - } - } - else if (instance instanceof FixedSubscriberChannel) { - FixedSubscriberChannel fixedSubscriberChannel = (FixedSubscriberChannel) instance; - String channelBeanName = fixedSubscriberChannel.getComponentName(); - if ("Unnamed fixed subscriber channel".equals(channelBeanName)) { - channelBeanName = flowNamePrefix + "channel" + - BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++; - } - registry.registerBeanDefinition(channelBeanName, component); - } - else { - String beanName = generateInstanceBeanDefinitionName(registry, instance); - registry.registerBeanDefinition(beanName, component); - } - } - } - else { - BeanDefinitionReaderUtils.registerWithGeneratedName(component, registry); - } - } - } - - } - - @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/core/IntegrationFlowBeanPostProcessor.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java new file mode 100644 index 0000000..8aad6aa --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/IntegrationFlowBeanPostProcessor.java @@ -0,0 +1,148 @@ +/* + * Copyright 2014 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.core; + +import java.util.Collection; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.beans.factory.BeanFactoryUtils; +import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.FixedSubscriberChannel; +import org.springframework.integration.config.ConsumerEndpointFactoryBean; +import org.springframework.integration.config.IntegrationConfigUtils; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.support.MessageChannelReference; +import org.springframework.messaging.MessageHandler; +import org.springframework.util.Assert; + +/** + * @author Artem Bilan + */ +public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, BeanFactoryAware { + + private ConfigurableListableBeanFactory beanFactory; + + @Override + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + Assert.isInstanceOf(ConfigurableListableBeanFactory.class, beanFactory, + "To use Spring Integration Java DSL the 'beanFactory' has to be an instance of " + + "'ConfigurableListableBeanFactory'. Consider using 'GenericApplicationContext' implementation." + ); + + this.beanFactory = (ConfigurableListableBeanFactory) beanFactory; + } + + @Override + public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { + return bean; + } + + @Override + public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { + if (bean instanceof IntegrationFlow) { + IntegrationFlow flow = (IntegrationFlow) bean; + String flowNamePrefix = beanName + ":"; + int channelNameIndex = 0; + for (Object component : flow.getIntegrationComponents()) { + if (component instanceof ConsumerEndpointSpec) { + ConsumerEndpointSpec endpointSpec = (ConsumerEndpointSpec) component; + MessageHandler messageHandler = endpointSpec.get().getT2(); + ConsumerEndpointFactoryBean endpoint = endpointSpec.get().getT1(); + String id = endpointSpec.getId(); + + Collection messageHandlers = + this.beanFactory.getBeansOfType(messageHandler.getClass(), false, false).values(); + + if (!messageHandlers.contains(messageHandler)) { + String handlerBeanName = generateBeanName(messageHandler); + String[] handlerAlias = id != null + ? new String[] {id + IntegrationConfigUtils.HANDLER_ALIAS_SUFFIX} + : null; + + registerComponent(messageHandler, handlerBeanName); + if (handlerAlias != null) { + for (String alias : handlerAlias) { + this.beanFactory.registerAlias(handlerBeanName, alias); + } + } + } + + String endpointBeanName = id; + if (endpointBeanName == null) { + endpointBeanName = generateBeanName(endpoint); + } + registerComponent(endpoint, endpointBeanName); + } + else { + Collection values = this.beanFactory.getBeansOfType(component.getClass(), false, false).values(); + if (!values.contains(component)) { + if (component instanceof AbstractMessageChannel) { + String channelBeanName = ((AbstractMessageChannel) component).getComponentName(); + if (channelBeanName == null) { + channelBeanName = flowNamePrefix + "channel" + + BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++; + } + registerComponent(component, channelBeanName); + } + else if (component instanceof MessageChannelReference) { + String channelBeanName = ((MessageChannelReference) component).getName(); + if (!this.beanFactory.containsBean(channelBeanName)) { + DirectChannel directChannel = new DirectChannel(); + registerComponent(directChannel, channelBeanName); + } + } + else if (component instanceof FixedSubscriberChannel) { + FixedSubscriberChannel fixedSubscriberChannel = (FixedSubscriberChannel) component; + String channelBeanName = fixedSubscriberChannel.getComponentName(); + if ("Unnamed fixed subscriber channel".equals(channelBeanName)) { + channelBeanName = flowNamePrefix + "channel" + + BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++; + } + registerComponent(component, channelBeanName); + } + else { + registerComponent(component, generateBeanName(component)); + } + } + } + } + } + return bean; + } + + private void registerComponent(Object component, String beanName) { + this.beanFactory.registerSingleton(beanName, component); + this.beanFactory.initializeBean(component, beanName); + } + + private String generateBeanName(Object instance) { + String generatedBeanName = instance.getClass().getName(); + String id = instance.getClass().getName(); + int counter = -1; + while (counter == -1 || this.beanFactory.containsBean(id)) { + counter++; + id = generatedBeanName + BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + counter; + } + return id; + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/MessagingProducerSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/MessagingProducerSpec.java new file mode 100644 index 0000000..41848b4 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/MessagingProducerSpec.java @@ -0,0 +1,62 @@ +/* + * Copyright 2014 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.core; + +import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.messaging.MessageChannel; + +/** + * @author Artem Bilan + */ +public abstract class MessagingProducerSpec, P extends MessageProducerSupport> + extends IntegrationComponentSpec { + + public MessagingProducerSpec(P producer) { + this.target = producer; + } + + public S id(String id) { + this.target.setBeanName(id); + return super.id(id); + } + + public S phase(int phase) { + this.target.setPhase(phase); + return _this(); + } + + public S autoStartup(boolean autoStartup) { + this.target.setAutoStartup(autoStartup); + return _this(); + } + + public S outputChannel(MessageChannel outputChannel) { + target.setOutputChannel(outputChannel); + return _this(); + } + + public S errorChannel(MessageChannel errorChannel) { + target.setErrorChannel(errorChannel); + return _this(); + } + + @Override + protected final P doGet() { + throw new UnsupportedOperationException(); + } + +} 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 17c3368..56c3760 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 @@ -65,7 +65,7 @@ import org.springframework.context.annotation.AnnotationConfigApplicationContext import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; -import org.springframework.context.annotation.Lazy; +import org.springframework.context.annotation.DependsOn; import org.springframework.data.mongodb.MongoDbFactory; import org.springframework.data.mongodb.core.SimpleMongoDbFactory; import org.springframework.integration.IntegrationMessageHeaderAccessor; @@ -294,6 +294,7 @@ public class IntegrationFlowTests { private AmqpTemplate amqpTemplate; @Autowired + @Qualifier("queue") private Queue amqpQueue; @Autowired @@ -331,7 +332,7 @@ public class IntegrationFlowTests { } assertTrue(this.outputChannel.getChannelInterceptors().contains(this.testChannelInterceptor)); - assertEquals(new Integer(10), this.testChannelInterceptor.getInvoked()); + assertThat(this.testChannelInterceptor.getInvoked(), Matchers.greaterThanOrEqualTo(10)); } @Test @@ -895,6 +896,26 @@ public class IntegrationFlowTests { assertThat(((Exception) receive.getPayload()).getMessage(), Matchers.containsString("' rejected Message")); } + @Autowired + @Qualifier("amqpOutboundInput") + private MessageChannel amqpOutboundInput; + + @Autowired + @Qualifier("amqpReplyChannel") + private PollableChannel amqpReplyChannel; + + @Test + public void testAmqpOutboundFlow() throws Exception { + this.amqpOutboundInput.send(MessageBuilder.withPayload("hello through the amqp") + .setHeader("routingKey", "foo") + .build()); + Message receive = this.amqpReplyChannel.receive(5000); + assertNotNull(receive); + assertEquals("HELLO THROUGH THE AMQP", receive.getPayload()); + } + + + @MessagingGateway(defaultRequestChannel = "controlBus") private static interface ControlBusGateway { @@ -977,11 +998,16 @@ public class IntegrationFlowTests { c -> c.autoStartup(false).id("payloadSerializingTransformer")) .channel(MessageChannels.queue(new SimpleMessageStore(), "fooQueue")) .transform(new PayloadDeserializingTransformer()) - .channel(MessageChannels.publishSubscribe("publishSubscribeChannel")) + .channel(publishSubscribeChannel()) .transform((Integer p) -> p * 2, c -> c.advice(this.expressionAdvice())) .get(); } + @Bean + public MessageChannel publishSubscribeChannel() { + return MessageChannels.publishSubscribe().get(); + } + @Bean public MongoDbFactory mongoDbFactory() throws Exception { return new SimpleMongoDbFactory(new MongoClient("localhost", mongoPort), "local"); @@ -1146,7 +1172,6 @@ public class IntegrationFlowTests { } @Autowired - @Lazy private GreetingService greetingService; @Bean @@ -1165,6 +1190,7 @@ public class IntegrationFlowTests { } @Bean + @DependsOn("enrichFlow") public IntegrationFlow enricherFlow() { return IntegrationFlows.fromFixedMessageChannel("enricherInput") .enrich(e -> e.requestChannel("enrichChannel") @@ -1308,6 +1334,9 @@ public class IntegrationFlowTests { @Autowired private ConnectionFactory rabbitConnectionFactory; + @Autowired + private AmqpTemplate amqpTemplate; + @Bean public Queue queue() { return new AnonymousQueue(); @@ -1322,6 +1351,27 @@ public class IntegrationFlowTests { } @Bean + public IntegrationFlow amqpOutboundFlow() { + return IntegrationFlows.from("amqpOutboundInput") + .handle(Amqp.outboundAdapter(this.amqpTemplate).routingKeyExpression("headers.routingKey").get()) + .get(); + } + + @Bean + public Queue fooQueue() { + return new Queue("foo"); + } + + @Bean + public IntegrationFlow amqpInboundFlow() { + return IntegrationFlows.from(Amqp.inboundAdapter(this.rabbitConnectionFactory, fooQueue())) + .transform(String.class, String::toUpperCase) + .channel(MessageChannels.queue("amqpReplyChannel")) + .get(); + } + + @Bean + @DependsOn("gatewayRequestFlow") public IntegrationFlow gatewayFlow() { return IntegrationFlows.from("gatewayInput") .gateway("gatewayRequest", g -> g.errorChannel("gatewayError").replyTimeout(10L))