DSL: Fix @Autowired issue for IntegrationFlows

* 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/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<AbstractBeanDefinition> integrationComponents = new LinkedHashSet<AbstractBeanDefinition>();
+	private final Set<Object> integrationComponents = new LinkedHashSet<Object>();
 
 	IntegrationFlow() {
 	}
 
-	public Set<AbstractBeanDefinition> getIntegrationComponents() {
+	public Set<Object> 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;
 	}
 

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");

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<AmqpInboundGatewaySpec, AmqpInboundGateway> 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<AmqpInboundGatewaySpec, AmqpInboundGateway> 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<AmqpInboundChannelAdapterSpec, AmqpInboundChannelAdapter> 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);
+	}
+
 }

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<AmqpInboundChannelAdapterSpec, AmqpInboundChannelAdapter> {
+
+	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;
+	}
+
+}

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<AmqpOutboundEndpointSpec, AmqpOutboundEndpoint> {
+
+	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;
+	}
+
+}

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<String> processedConfigurations = new HashSet<String>();
-		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);
-	}
-
 }

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;
+	}
+
+}

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<S extends MessagingProducerSpec<S, P>, P extends MessageProducerSupport>
+		extends IntegrationComponentSpec<S, P> {
+
+	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();
+	}
+
+}

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))