Spec<?, ?> improvements

* Provide `id`-aware DSL methods for `MessageChannelSpec`
* Rework `DslIntegrationConfigurationInitializer` to delegate to `IntegrationFlowBeanPostProcessor`
to allow to use components from `IntegrationFlowBuilder` via `@Autowired`

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/DslIntegrationConfigurationInitializer.java

@@ -16,23 +16,14 @@
 
 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.messaging.MessageHandler;
+import org.springframework.integration.dsl.core.Spec;
+import org.springframework.util.Assert;
 
 /**
  * The Java DSL Integration infrastructure {@code beanFactory} initializer.
@@ -43,71 +34,24 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig
 
 	@Override
 	public void initialize(ConfigurableListableBeanFactory configurableListableBeanFactory) throws BeansException {
-		this.initializeIntegrationFlows(configurableListableBeanFactory);
+		this.populateBeansFromSpecs(configurableListableBeanFactory);
+		configurableListableBeanFactory.addBeanPostProcessor(new IntegrationFlowBeanPostProcessor(configurableListableBeanFactory));
 	}
 
-	private void initializeIntegrationFlows(ConfigurableListableBeanFactory beanFactory) {
-		Map<String, IntegrationFlow> integrationFlows = beanFactory.getBeansOfType(IntegrationFlow.class, false, false);
+	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<String, Spec> specs = beanFactory.getBeansOfType(Spec.class, false, false);
 		BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory;
-		for (Map.Entry<String, IntegrationFlow> 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);
+		for (Map.Entry<String, Spec> specEntry : specs.entrySet()) {
+			String id = specEntry.getKey();
+			Spec<?, ?> spec = specEntry.getValue();
+			registry.removeBeanDefinition(id);
+			beanFactory.destroyBean(id);
+			beanFactory.registerSingleton(id, spec.get());
 		}
-
-	}
-
-	@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/EndpointSpec.java

@@ -23,6 +23,7 @@ import java.util.List;
 import org.aopalliance.aop.Advice;
 
 import org.springframework.integration.config.ConsumerEndpointFactoryBean;
+import org.springframework.integration.dsl.support.PollerSpec;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.messaging.MessageHandler;
@@ -31,17 +32,17 @@ import org.springframework.messaging.MessageHandler;
  * @author Artem Bilan
  * @since 4.0
  */
-public abstract class EndpointSpec<S extends EndpointSpec<S, C>, C extends MessageHandler> {
+public abstract class EndpointSpec<S extends EndpointSpec<S, H>, H extends MessageHandler> {
 
 	private final ConsumerEndpointFactoryBean endpointFactoryBean = new ConsumerEndpointFactoryBean();
 
-	private final C messageHandler;
+	private final H messageHandler;
 
 	private final List<Advice> adviceChain = new LinkedList<Advice>();
 
 	private String id;
 
-	EndpointSpec(C messageHandler) {
+	EndpointSpec(H messageHandler) {
 		this.messageHandler = messageHandler;
 		this.endpointFactoryBean.setHandler(this.messageHandler);
 		if (this.messageHandler instanceof AbstractReplyProducingMessageHandler) {
@@ -78,6 +79,11 @@ public abstract class EndpointSpec<S extends EndpointSpec<S, C>, C extends Messa
 		return _this();
 	}
 
+	public S poller(PollerSpec pollerMetadataSpec) {
+		this.endpointFactoryBean.setPollerMetadata(pollerMetadataSpec.get());
+		return _this();
+	}
+
 	String getId() {
 		return id;
 	}
@@ -86,7 +92,7 @@ public abstract class EndpointSpec<S extends EndpointSpec<S, C>, C extends Messa
 		return this.endpointFactoryBean;
 	}
 
-	C getHandler() {
+	H getHandler() {
 		return this.messageHandler;
 	}
 

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java

@@ -22,9 +22,9 @@ import org.springframework.messaging.MessageHandler;
  * @author Artem Bilan
  * @since 4.0
  */
-public final class GenericEndpointSpec<C extends MessageHandler> extends EndpointSpec<GenericEndpointSpec<C>, C> {
+public final class GenericEndpointSpec<H extends MessageHandler> extends EndpointSpec<GenericEndpointSpec<H>, H> {
 
-	GenericEndpointSpec(C messageHandler) {
+	GenericEndpointSpec(H messageHandler) {
 		super(messageHandler);
 	}
 

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java

@@ -0,0 +1,106 @@
+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);
+	}
+
+}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java

@@ -23,6 +23,7 @@ import org.springframework.integration.config.SourcePollingChannelAdapterFactory
 import org.springframework.integration.core.GenericSelector;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.core.MessageSelector;
+import org.springframework.integration.dsl.channel.MessageChannelSpec;
 import org.springframework.integration.dsl.support.EndpointConfigurer;
 import org.springframework.integration.filter.ExpressionEvaluatingSelector;
 import org.springframework.integration.filter.MessageFilter;
@@ -34,6 +35,7 @@ import org.springframework.integration.transformer.MessageTransformingHandler;
 import org.springframework.integration.transformer.MethodInvokingTransformer;
 import org.springframework.integration.transformer.Transformer;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.util.Assert;
 
 /**
  * @author Artem Bilan
@@ -62,10 +64,53 @@ public final class IntegrationFlowBuilder {
 	}
 
 	public IntegrationFlowBuilder channel(MessageChannel messageChannel) {
+		Assert.notNull(messageChannel);
 		this.currentMessageChannel = messageChannel;
 		return this.addComponent(this.currentMessageChannel).registerOutputChannelIfCan(this.currentMessageChannel);
 	}
 
+	public IntegrationFlowBuilder channel(MessageChannelSpec<?, ?> messageChannelSpec) {
+		Assert.notNull(messageChannelSpec);
+		return this.channel(messageChannelSpec.get());
+	}
+
+	public IntegrationFlowBuilder transform(String expression) {
+		return this.transform(PARSER.parseExpression(expression));
+	}
+
+	public IntegrationFlowBuilder transform(Expression expression) {
+		return this.transform(new ExpressionEvaluatingTransformer(expression));
+	}
+
+	public <S, T> IntegrationFlowBuilder transform(GenericTransformer<S, T> genericTransformer) {
+		return this.transform(genericTransformer, null);
+	}
+
+	public <S, T> IntegrationFlowBuilder transform(GenericTransformer<S, T> genericTransformer,
+												   EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
+		Transformer transformer = genericTransformer instanceof Transformer
+				? (Transformer) genericTransformer : new MethodInvokingTransformer(genericTransformer);
+		return this.register(new GenericEndpointSpec<MessageTransformingHandler>(new MessageTransformingHandler(transformer)), endpointConfigurer);
+	}
+
+	public IntegrationFlowBuilder filter(String expression) {
+		return this.filter(PARSER.parseExpression(expression));
+	}
+
+	public IntegrationFlowBuilder filter(Expression expression) {
+		return this.filter(new ExpressionEvaluatingSelector(expression));
+	}
+
+	public <S> IntegrationFlowBuilder filter(GenericSelector<S> genericSelector) {
+		return this.filter(genericSelector, null);
+	}
+
+	public <S> IntegrationFlowBuilder filter(GenericSelector<S> genericSelector, EndpointConfigurer<FilterEndpointSpec> endpointConfigurer) {
+		MessageSelector selector = genericSelector instanceof MessageSelector
+				? (MessageSelector) genericSelector : new MethodInvokingSelector(genericSelector);
+		return this.register(new FilterEndpointSpec(new MessageFilter(selector)), endpointConfigurer);
+	}
+
 	private IntegrationFlowBuilder registerOutputChannelIfCan(MessageChannel outputChannel) {
 		this.flow.addComponent(outputChannel);
 		if (this.currentComponent != null) {
@@ -83,50 +128,10 @@ public final class IntegrationFlowBuilder {
 		return this;
 	}
 
-	public IntegrationFlowBuilder transform(String expression) {
-		return this.transform(PARSER.parseExpression(expression));
-	}
-
-	public IntegrationFlowBuilder transform(Expression expression) {
-		return this.transform(new ExpressionEvaluatingTransformer(expression));
-	}
-
-	public <S, T> IntegrationFlowBuilder transform(GenericTransformer<S, T> genericTransformer) {
-		Transformer transformer = genericTransformer instanceof Transformer
-				? (Transformer) genericTransformer : new MethodInvokingTransformer(genericTransformer);
-		return this.transform(genericTransformer, new DefaultEndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>>());
-	}
-
-	public <S, T> IntegrationFlowBuilder transform(GenericTransformer<S, T> genericTransformer,
-												   EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
-		Transformer transformer = genericTransformer instanceof Transformer
-				? (Transformer) genericTransformer : new MethodInvokingTransformer(genericTransformer);
-		GenericEndpointSpec<MessageTransformingHandler> spec = new GenericEndpointSpec<MessageTransformingHandler>(new MessageTransformingHandler(transformer));
-		endpointConfigurer.configure(spec);
-		return this.register(spec);
-	}
-
-	public IntegrationFlowBuilder filter(String expression) {
-		return this.filter(PARSER.parseExpression(expression));
-	}
-
-	public IntegrationFlowBuilder filter(Expression expression) {
-		return this.filter(new ExpressionEvaluatingSelector(expression));
-	}
-
-	public <S> IntegrationFlowBuilder filter(GenericSelector<S> genericSelector) {
-		return this.filter(genericSelector, new DefaultEndpointConfigurer<FilterEndpointSpec>());
-	}
-
-	public <S> IntegrationFlowBuilder filter(GenericSelector<S> genericSelector, EndpointConfigurer<FilterEndpointSpec> endpointConfigurer) {
-		MessageSelector selector = genericSelector instanceof MessageSelector
-				? (MessageSelector) genericSelector : new MethodInvokingSelector(genericSelector);
-		FilterEndpointSpec spec = new FilterEndpointSpec(new MessageFilter(selector));
-		endpointConfigurer.configure(spec);
-		return this.register(spec);
-	}
-
-	private IntegrationFlowBuilder register(EndpointSpec<?, ?> endpointSpec) {
+	private <S extends EndpointSpec<?, ?>> IntegrationFlowBuilder register(S endpointSpec, EndpointConfigurer<S> endpointConfigurer) {
+		if (endpointConfigurer != null) {
+			endpointConfigurer.configure(endpointSpec);
+		}
 		MessageChannel inputChannel = this.currentMessageChannel;
 		this.currentMessageChannel = null;
 		if (inputChannel == null) {
@@ -143,12 +148,4 @@ public final class IntegrationFlowBuilder {
 		return this.flow;
 	}
 
-	private class DefaultEndpointConfigurer<S extends EndpointSpec<?, ?>> implements EndpointConfigurer<S> {
-
-		@Override
-		public void configure(S spec) {
-
-		}
-	}
-
 }

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java

@@ -18,9 +18,11 @@ package org.springframework.integration.dsl;
 
 import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean;
 import org.springframework.integration.core.MessageSource;
-import org.springframework.integration.endpoint.AbstractEndpoint;
+import org.springframework.integration.dsl.channel.MessageChannelSpec;
+import org.springframework.integration.dsl.support.PollerSpec;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.util.Assert;
 
 /**
  * @author Artem Bilan
@@ -31,8 +33,17 @@ public final class IntegrationFlows {
 		return new IntegrationFlowBuilder().channel(messageChannel);
 	}
 
+	public static IntegrationFlowBuilder from(MessageChannelSpec<?, ?> messageChannelSpec) {
+		return from(messageChannelSpec.get());
+	}
+
 	public static IntegrationFlowBuilder from(MessageSource<?> messageSource) {
-		return from(messageSource, null);
+		return from(messageSource, (PollerMetadata) null);
+	}
+
+	public static IntegrationFlowBuilder from(MessageSource<?> messageSource, PollerSpec pollerSpec) {
+		Assert.notNull(pollerSpec);
+		return from(messageSource, pollerSpec.get());
 	}
 
 	public static IntegrationFlowBuilder from(MessageSource<?> messageSource, PollerMetadata pollerMetadata) {
@@ -45,9 +56,9 @@ public final class IntegrationFlows {
 				.currentComponent(factoryBean);
 	}
 
-	public static IntegrationFlowBuilder from(AbstractEndpoint endpoint) {
+	/*public static IntegrationFlowBuilder from(AbstractEndpoint endpoint) {
 		return new IntegrationFlowBuilder();
-	}
+	}*/
 
 	private IntegrationFlows() {
 	}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/DirectChannelSpec.java

@@ -21,10 +21,10 @@ import org.springframework.integration.channel.DirectChannel;
 /**
  * @author Artem Bilan
  */
-public class DirectChannelSpec extends LoadBalancingChannelSpecSupport<DirectChannelSpec, DirectChannel> {
+public class DirectChannelSpec extends LoadBalancingChannelSpec<DirectChannelSpec, DirectChannel> {
 
 	@Override
-	public DirectChannel get() {
+	protected DirectChannel doGet() {
 		this.channel = new DirectChannel(this.loadBalancingStrategy);
 		if (this.failover != null) {
 			this.channel.setFailover(this.failover);
@@ -32,7 +32,7 @@ public class DirectChannelSpec extends LoadBalancingChannelSpecSupport<DirectCha
 		if (this.maxSubscribers != null) {
 			this.channel.setMaxSubscribers(this.maxSubscribers);
 		}
-		return super.get();
+		return super.doGet();
 	}
 
 	DirectChannelSpec() {

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/ExecutorChannelSpec.java

@@ -23,7 +23,7 @@ import org.springframework.integration.channel.ExecutorChannel;
 /**
  * @author Artem Bilan
  */
-public class ExecutorChannelSpec extends LoadBalancingChannelSpecSupport<ExecutorChannelSpec, ExecutorChannel> {
+public class ExecutorChannelSpec extends LoadBalancingChannelSpec<ExecutorChannelSpec, ExecutorChannel> {
 
 	private final Executor executor;
 
@@ -31,7 +31,8 @@ public class ExecutorChannelSpec extends LoadBalancingChannelSpecSupport<Executo
 		this.executor = executor;
 	}
 
-	public ExecutorChannel get() {
+	@Override
+	protected ExecutorChannel doGet() {
 		this.channel = new ExecutorChannel(this.executor, this.loadBalancingStrategy);
 		if (this.failover != null) {
 			this.channel.setFailover(this.failover);
@@ -39,7 +40,7 @@ public class ExecutorChannelSpec extends LoadBalancingChannelSpecSupport<Executo
 		if (this.maxSubscribers != null) {
 			this.channel.setMaxSubscribers(this.maxSubscribers);
 		}
-		return super.get();
+		return super.doGet();
 	}
 
 }

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/LoadBalancingChannelSpec.java

@@ -23,7 +23,7 @@ import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrateg
 /**
  * @author Artem Bilan
  */
-public class LoadBalancingChannelSpecSupport<S extends ChannelSpecSupport<S, C>, C extends AbstractMessageChannel> extends ChannelSpecSupport<S, C> {
+public abstract class LoadBalancingChannelSpec<S extends MessageChannelSpec<S, C>, C extends AbstractMessageChannel> extends MessageChannelSpec<S, C> {
 
 	protected LoadBalancingStrategy loadBalancingStrategy = new RoundRobinLoadBalancingStrategy();
 

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannelSpec.java

@@ -19,12 +19,13 @@ package org.springframework.integration.dsl.channel;
 import java.util.Arrays;
 
 import org.springframework.integration.channel.AbstractMessageChannel;
+import org.springframework.integration.dsl.core.Spec;
 import org.springframework.messaging.support.ChannelInterceptor;
 
 /**
  * @author Artem Bilan
  */
-public abstract class ChannelSpecSupport<S extends ChannelSpecSupport<S, C>, C extends AbstractMessageChannel> {
+public abstract class MessageChannelSpec<S extends MessageChannelSpec<S, C>, C extends AbstractMessageChannel> extends Spec<S, C> {
 
 	protected C channel;
 
@@ -34,6 +35,11 @@ public abstract class ChannelSpecSupport<S extends ChannelSpecSupport<S, C>, C e
 
 	private ChannelInterceptor[] interceptors;
 
+	S id(String id) {
+		this.id = id;
+		return _this();
+	}
+
 	public S datatypes(Class<?>... datatypes) {
 		this.datatypes = datatypes;
 		return _this();
@@ -44,12 +50,8 @@ public abstract class ChannelSpecSupport<S extends ChannelSpecSupport<S, C>, C e
 		return _this();
 	}
 
-	public S id(String id) {
-		this.id = id;
-		return _this();
-	}
-
-	public C get() {
+	@Override
+	protected C doGet() {
 		this.channel.setDatatypes(this.datatypes);
 		this.channel.setBeanName(this.id);
 		if (this.interceptors != null) {
@@ -58,9 +60,5 @@ public abstract class ChannelSpecSupport<S extends ChannelSpecSupport<S, C>, C e
 		return this.channel;
 	}
 
-	@SuppressWarnings("unchecked")
-	protected S _this() {
-		return (S) this;
-	}
 
 }

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java

@@ -19,7 +19,6 @@ package org.springframework.integration.dsl.channel;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Executor;
 
-import org.springframework.integration.dispatcher.LoadBalancingStrategy;
 import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.messaging.Message;
 
@@ -32,46 +31,82 @@ public final class MessageChannels {
 		return new DirectChannelSpec();
 	}
 
-	public static DirectChannelSpec direct(LoadBalancingStrategy loadBalancingStrategy) {
-		return direct().loadBalancer(loadBalancingStrategy);
+	public static DirectChannelSpec direct(String id) {
+		return direct().id(id);
 	}
 
 	public static QueueChannelSpec queue() {
 		return new QueueChannelSpec();
 	}
 
+	public static QueueChannelSpec queue(String id) {
+		return queue().id(id);
+	}
+
 	public static QueueChannelSpec queue(BlockingQueue<Message<?>> queue) {
 		return new QueueChannelSpec(queue);
 	}
 
+	public static QueueChannelSpec queue(String id, BlockingQueue<Message<?>> queue) {
+		return queue(queue).id(id);
+	}
+
 	public static QueueChannelSpec queue(Integer capacity) {
 		return new QueueChannelSpec(capacity);
 	}
 
+	public static QueueChannelSpec queue(String id, Integer capacity) {
+		return queue(capacity).id(id);
+	}
+
 	public static QueueChannelSpec.MessageStoreSpec queue(MessageGroupStore messageGroupStore, Object groupId) {
 		return new QueueChannelSpec.MessageStoreSpec(messageGroupStore, groupId);
 	}
 
+	public static QueueChannelSpec.MessageStoreSpec queue(String id, MessageGroupStore messageGroupStore, Object groupId) {
+		return queue(messageGroupStore, groupId).id(id);
+	}
+
 	public static ExecutorChannelSpec executor(Executor executor) {
 		return new ExecutorChannelSpec(executor);
 	}
 
+	public static ExecutorChannelSpec executor(String id, Executor executor) {
+		return executor(executor).id(id);
+	}
+
 	public static RendezvousChannelSpec rendezvous() {
 		return new RendezvousChannelSpec();
 	}
 
+	public static RendezvousChannelSpec rendezvous(String id) {
+		return rendezvous().id(id);
+	}
+
 	public static PriorityChannelSpec priority() {
 		return new PriorityChannelSpec();
 	}
 
+	public static PriorityChannelSpec priority(String id) {
+		return priority().id(id);
+	}
+
 	public static PublishSubscribeChannelSpec publishSubscribe() {
 		return new PublishSubscribeChannelSpec();
 	}
 
+	public static PublishSubscribeChannelSpec publishSubscribe(String id) {
+		return publishSubscribe().id(id);
+	}
+
 	public static PublishSubscribeChannelSpec publishSubscribe(Executor executor) {
 		return new PublishSubscribeChannelSpec(executor);
 	}
 
+	public static PublishSubscribeChannelSpec publishSubscribe(String id, Executor executor) {
+		return publishSubscribe(executor).id(id);
+	}
+
 	private MessageChannels() {
 	}
 

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PriorityChannelSpec.java

@@ -24,7 +24,7 @@ import org.springframework.messaging.Message;
 /**
  * @author Artem Bilan
  */
-public class PriorityChannelSpec extends ChannelSpecSupport<PriorityChannelSpec, PriorityChannel> {
+public class PriorityChannelSpec extends MessageChannelSpec<PriorityChannelSpec, PriorityChannel> {
 
 	private int capacity;
 
@@ -41,9 +41,9 @@ public class PriorityChannelSpec extends ChannelSpecSupport<PriorityChannelSpec,
 	}
 
 	@Override
-	public PriorityChannel get() {
+	protected PriorityChannel doGet() {
 		this.channel = new PriorityChannel(this.capacity, this.comparator);
-		return super.get();
+		return super.doGet();
 	}
 
 

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PublishSubscribeChannelSpec.java

@@ -24,7 +24,7 @@ import org.springframework.util.ErrorHandler;
 /**
  * @author Artem Bilan
  */
-public class PublishSubscribeChannelSpec extends ChannelSpecSupport<PublishSubscribeChannelSpec, PublishSubscribeChannel> {
+public class PublishSubscribeChannelSpec extends MessageChannelSpec<PublishSubscribeChannelSpec, PublishSubscribeChannel> {
 
 	PublishSubscribeChannelSpec() {
 		this.channel = new PublishSubscribeChannel();

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java

@@ -27,7 +27,7 @@ import org.springframework.messaging.Message;
 /**
  * @author Artem Bilan
  */
-public class QueueChannelSpec extends ChannelSpecSupport<QueueChannelSpec, QueueChannel> {
+public class QueueChannelSpec extends MessageChannelSpec<QueueChannelSpec, QueueChannel> {
 
 	protected BlockingQueue<Message<?>> queue;
 
@@ -45,7 +45,7 @@ public class QueueChannelSpec extends ChannelSpecSupport<QueueChannelSpec, Queue
 	}
 
 	@Override
-	public QueueChannel get() {
+	protected QueueChannel doGet() {
 		if (this.queue != null) {
 			this.channel = new QueueChannel(this.queue);
 		}
@@ -55,7 +55,7 @@ public class QueueChannelSpec extends ChannelSpecSupport<QueueChannelSpec, Queue
 		else {
 			this.channel = new QueueChannel();
 		}
-		return super.get();
+		return super.doGet();
 	}
 
 	public static class MessageStoreSpec extends QueueChannelSpec {
@@ -72,6 +72,11 @@ public class QueueChannelSpec extends ChannelSpecSupport<QueueChannelSpec, Queue
 			this.groupId = groupId;
 		}
 
+		@Override
+		MessageStoreSpec id(String id) {
+			return (MessageStoreSpec) super.id(id);
+		}
+
 		public MessageStoreSpec capacity(Integer capacity) {
 			this.capacity = capacity;
 			return this;
@@ -83,7 +88,7 @@ public class QueueChannelSpec extends ChannelSpecSupport<QueueChannelSpec, Queue
 		}
 
 		@Override
-		public QueueChannel get() {
+		protected QueueChannel doGet() {
 			if (this.capacity != null) {
 				if (this.storeLock != null) {
 					this.queue = new MessageGroupQueue(messageGroupStore, groupId, this.capacity, this.storeLock);
@@ -98,7 +103,7 @@ public class QueueChannelSpec extends ChannelSpecSupport<QueueChannelSpec, Queue
 			else {
 				this.queue = new MessageGroupQueue(messageGroupStore, groupId);
 			}
-			return super.get();
+			return super.doGet();
 		}
 
 	}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/RendezvousChannelSpec.java

@@ -21,7 +21,7 @@ import org.springframework.integration.channel.RendezvousChannel;
 /**
  * @author Artem Bilan
  */
-public class RendezvousChannelSpec extends ChannelSpecSupport<RendezvousChannelSpec, RendezvousChannel> {
+public class RendezvousChannelSpec extends MessageChannelSpec<RendezvousChannelSpec, RendezvousChannel> {
 
 	RendezvousChannelSpec() {
 		this.channel = new RendezvousChannel();

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/Spec.java

@@ -0,0 +1,41 @@
+/*
+ * 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;
+
+/**
+ * @author Artem Bilan
+ * @since 4.0
+ */
+public abstract class Spec<S extends Spec<S, T>, T> {
+
+	private volatile T target;
+
+	public final T get() {
+		if (this.target == null) {
+			this.target = this.doGet();
+		}
+		return this.target;
+	}
+
+	protected abstract T doGet();
+
+	@SuppressWarnings("unchecked")
+	protected S _this() {
+		return (S) this;
+	}
+
+}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/PollerSpec.java

@@ -0,0 +1,94 @@
+/*
+ * 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.support;
+
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.Executor;
+
+import org.aopalliance.aop.Advice;
+
+import org.springframework.integration.dsl.core.Spec;
+import org.springframework.integration.scheduling.PollerMetadata;
+import org.springframework.integration.transaction.TransactionSynchronizationFactory;
+import org.springframework.scheduling.Trigger;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource;
+import org.springframework.transaction.interceptor.TransactionInterceptor;
+import org.springframework.util.ErrorHandler;
+
+/**
+* @author Artem Bilan
+* @since 4.0
+*/
+public final class PollerSpec extends Spec<PollerSpec, PollerMetadata> {
+
+	private final PollerMetadata pollerMetadata = new PollerMetadata();
+
+	private final List<Advice> adviceChain = new LinkedList<Advice>();
+
+	PollerSpec(Trigger trigger) {
+		this.pollerMetadata.setTrigger(trigger);
+	}
+
+	public PollerSpec transactionSynchronizationFactory(TransactionSynchronizationFactory transactionSynchronizationFactory) {
+		pollerMetadata.setTransactionSynchronizationFactory(transactionSynchronizationFactory);
+		return this;
+	}
+
+	public PollerSpec errorHandler(ErrorHandler errorHandler) {
+		pollerMetadata.setErrorHandler(errorHandler);
+		return this;
+	}
+
+	public PollerSpec maxMessagesPerPoll(long maxMessagesPerPoll) {
+		pollerMetadata.setMaxMessagesPerPoll(maxMessagesPerPoll);
+		return this;
+	}
+
+	public PollerSpec receiveTimeout(long receiveTimeout) {
+		pollerMetadata.setReceiveTimeout(receiveTimeout);
+		return this;
+	}
+
+	public PollerSpec advice(Advice... advice) {
+		this.adviceChain.addAll(Arrays.asList(advice));
+		return this;
+	}
+
+	public PollerSpec transactional(PlatformTransactionManager transactionManager) {
+		return this.advice(new TransactionInterceptor(transactionManager, new MatchAlwaysTransactionAttributeSource()));
+	}
+
+	public PollerSpec taskExecutor(Executor taskExecutor) {
+		pollerMetadata.setTaskExecutor(taskExecutor);
+		return this;
+	}
+
+	public PollerSpec sendTimeout(long sendTimeout) {
+		pollerMetadata.setSendTimeout(sendTimeout);
+		return this;
+	}
+
+	@Override
+	protected PollerMetadata doGet() {
+		pollerMetadata.setAdviceChain(this.adviceChain);
+		return this.pollerMetadata;
+	}
+
+}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Pollers.java

@@ -16,24 +16,12 @@
 
 package org.springframework.integration.dsl.support;
 
-import java.util.Arrays;
-import java.util.LinkedList;
-import java.util.List;
 import java.util.TimeZone;
-import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 
-import org.aopalliance.aop.Advice;
-
-import org.springframework.integration.scheduling.PollerMetadata;
-import org.springframework.integration.transaction.TransactionSynchronizationFactory;
 import org.springframework.scheduling.Trigger;
 import org.springframework.scheduling.support.CronTrigger;
 import org.springframework.scheduling.support.PeriodicTrigger;
-import org.springframework.transaction.PlatformTransactionManager;
-import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource;
-import org.springframework.transaction.interceptor.TransactionInterceptor;
-import org.springframework.util.ErrorHandler;
 
 /**
  * @author Artem Bilan
@@ -77,62 +65,4 @@ public final class Pollers {
 	private Pollers() {
 	}
 
-	public static final class PollerSpec {
-
-		private final PollerMetadata pollerMetadata = new PollerMetadata();
-
-		private final List<Advice> adviceChain = new LinkedList<Advice>();
-
-		private PollerSpec(Trigger trigger) {
-			this.pollerMetadata.setTrigger(trigger);
-		}
-
-		public PollerSpec transactionSynchronizationFactory(TransactionSynchronizationFactory transactionSynchronizationFactory) {
-			pollerMetadata.setTransactionSynchronizationFactory(transactionSynchronizationFactory);
-			return this;
-		}
-
-		public PollerSpec errorHandler(ErrorHandler errorHandler) {
-			pollerMetadata.setErrorHandler(errorHandler);
-			return this;
-		}
-
-		public PollerSpec maxMessagesPerPoll(long maxMessagesPerPoll) {
-			pollerMetadata.setMaxMessagesPerPoll(maxMessagesPerPoll);
-			return this;
-		}
-
-		public PollerSpec receiveTimeout(long receiveTimeout) {
-			pollerMetadata.setReceiveTimeout(receiveTimeout);
-			return this;
-		}
-
-		public PollerSpec advice(Advice... advice) {
-			this.adviceChain.addAll(Arrays.asList(advice));
-			return this;
-		}
-
-		public PollerSpec transactional(PlatformTransactionManager transactionManager) {
-			return this.advice(new TransactionInterceptor(transactionManager, new MatchAlwaysTransactionAttributeSource()));
-		}
-
-		public PollerSpec taskExecutor(Executor taskExecutor) {
-			pollerMetadata.setTaskExecutor(taskExecutor);
-			return this;
-		}
-
-		public PollerSpec sendTimeout(long sendTimeout) {
-			pollerMetadata.setSendTimeout(sendTimeout);
-			return this;
-		}
-
-		public PollerMetadata get() {
-			pollerMetadata.setAdviceChain(this.adviceChain);
-			return this.pollerMetadata;
-		}
-
-
-
-	}
-
 }

spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java

@@ -22,6 +22,7 @@ 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;
@@ -42,7 +43,10 @@ import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.channel.DirectChannelSpec;
 import org.springframework.integration.dsl.channel.MessageChannels;
+import org.springframework.integration.dsl.channel.QueueChannelSpec;
+import org.springframework.integration.dsl.support.PollerSpec;
 import org.springframework.integration.dsl.support.Pollers;
 import org.springframework.integration.endpoint.MethodInvokingMessageSource;
 import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice;
@@ -118,6 +122,16 @@ 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();
@@ -128,48 +142,51 @@ public class IntegrationFlowTests {
 
 		@Bean
 		public IntegrationFlow flow1() {
-			return IntegrationFlows.from(this.integerMessageSource(), Pollers.fixedRate(100).get())
+			return IntegrationFlows.from(this.integerMessageSource(), Pollers.fixedRate(100))
 					.transform("payload.toString()")
-					.channel(MessageChannels.queue().id("flow1QueueChannel").get())
+					.channel(MessageChannels.queue("flow1QueueChannel"))
 					.get();
 		}
 
-		@Bean
-		public DirectChannel inputChannel() {
-			return MessageChannels.direct().get();
-		}
-
-		@Bean
-		public QueueChannel successChannel() {
-			return MessageChannels.queue().get();
-		}
-
 		@Bean(name = PollerMetadata.DEFAULT_POLLER_METADATA_BEAN_NAME)
-		public PollerMetadata poller() {
-			return Pollers.fixedRate(500).get();
+		public PollerSpec poller() {
+			return Pollers.fixedRate(500);
 		}
 
+	}
+
+	@Configuration
+	public static class ContextConfiguration2 {
+
+		@Autowired
+		@Qualifier("inputChannel")
+		private DirectChannel inputChannel;
+
+		@Autowired
+		@Qualifier("successChannel")
+		private PollableChannel successChannel;
+
 		@Bean
 		public Advice expressionAdvice() {
 			ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
 			advice.setOnSuccessExpression("payload");
-			advice.setSuccessChannel(this.successChannel());
+			advice.setSuccessChannel(this.successChannel);
 			return advice;
 		}
 
 		@Bean
 		public IntegrationFlow flow2() {
-			return IntegrationFlows.from(this.inputChannel())
+			return IntegrationFlows.from(this.inputChannel)
 					.filter(p -> p instanceof String, c -> c.id("filter"))
 					.<String, Integer>transform(Integer::parseInt)
 					.transform(new PayloadSerializingTransformer(),
 							c -> c.autoStartup(false).id("payloadSerializingTransformer"))
-					.channel(MessageChannels.queue(new SimpleMessageStore(), "fooQueue").get())
+					.channel(MessageChannels.queue(new SimpleMessageStore(), "fooQueue"))
 					.transform(new PayloadDeserializingTransformer())
+					.channel(MessageChannels.executor("executor", Executors.newCachedThreadPool()))
 					.transform((Integer p) -> p * 2, c -> c.advice(this.expressionAdvice()))
 					.get();
 		}
-
 	}
 
 }