INTEXT-96: The Java DSL initial implementation

JIRA: https://jira.springsource.org/browse/INTEXT-96

* `MessageChannels` Builder
* Configuration infrastructure
* `Pollers` Builder
* Initial `IntegrationFlows` Builder
* `EndpointConfigurer` Specs
* EIP-methods `from(MessageSource)`, `from(MessageChannel)`, `transform`, `filter`

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

@@ -0,0 +1,97 @@
+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;
+
+/**
+ * The Java DSL Integration infrastructure {@code beanFactory} initializer.
+ *
+ * @author Artem Bilan
+ */
+public class DslIntegrationConfigurationInitializer implements IntegrationConfigurationInitializer {
+
+	@Override
+	public void initialize(ConfigurableListableBeanFactory configurableListableBeanFactory) throws BeansException {
+		this.initializeIntegrationFlows(configurableListableBeanFactory);
+	}
+
+	private void initializeIntegrationFlows(ConfigurableListableBeanFactory beanFactory) {
+		Map<String, IntegrationFlow> integrationFlows = beanFactory.getBeansOfType(IntegrationFlow.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);
+		}
+
+	}
+
+	@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

@@ -0,0 +1,82 @@
+package org.springframework.integration.dsl;
+
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+
+import org.aopalliance.aop.Advice;
+
+import org.springframework.integration.config.ConsumerEndpointFactoryBean;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.scheduling.PollerMetadata;
+import org.springframework.messaging.MessageHandler;
+
+/**
+ * @author Artem Bilan
+ * @since 4.0
+ */
+public abstract class EndpointSpec<S extends EndpointSpec<S, C>, C extends MessageHandler> {
+
+	private final ConsumerEndpointFactoryBean endpointFactoryBean = new ConsumerEndpointFactoryBean();
+
+	private final C messageHandler;
+
+	private final List<Advice> adviceChain = new LinkedList<Advice>();
+
+	private String id;
+
+	EndpointSpec(C messageHandler) {
+		this.messageHandler = messageHandler;
+		this.endpointFactoryBean.setHandler(this.messageHandler);
+		if (this.messageHandler instanceof AbstractReplyProducingMessageHandler) {
+			((AbstractReplyProducingMessageHandler) this.messageHandler).setAdviceChain(this.adviceChain);
+		}
+		else {
+			this.endpointFactoryBean.setAdviceChain(this.adviceChain);
+		}
+	}
+
+	public S id(String id) {
+		this.id = id;
+		this.endpointFactoryBean.setBeanName(id);
+		return _this();
+	}
+
+	public S phase(int phase) {
+		this.endpointFactoryBean.setPhase(phase);
+		return _this();
+	}
+
+	public S autoStartup(boolean autoStartup) {
+		this.endpointFactoryBean.setAutoStartup(autoStartup);
+		return _this();
+	}
+
+	public S advice(Advice... advice) {
+		this.adviceChain.addAll(Arrays.asList(advice));
+		return _this();
+	}
+
+	public S poller(PollerMetadata pollerMetadata) {
+		this.endpointFactoryBean.setPollerMetadata(pollerMetadata);
+		return _this();
+	}
+
+	String getId() {
+		return id;
+	}
+
+	ConsumerEndpointFactoryBean getEndpoint() {
+		return this.endpointFactoryBean;
+	}
+
+	C getHandler() {
+		return this.messageHandler;
+	}
+
+	@SuppressWarnings("unchecked")
+	protected S _this() {
+		return (S) this;
+	}
+
+}

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

@@ -0,0 +1,31 @@
+package org.springframework.integration.dsl;
+
+import org.springframework.integration.filter.MessageFilter;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * @author Artem Bilan
+ * @since 4.0
+ */
+public final class FilterEndpointSpec extends EndpointSpec<FilterEndpointSpec, MessageFilter> {
+
+	FilterEndpointSpec(MessageFilter messageFilter) {
+		super(messageFilter);
+	}
+
+	public FilterEndpointSpec throwExceptionOnRejection(boolean throwExceptionOnRejection) {
+		this.getHandler().setThrowExceptionOnRejection(throwExceptionOnRejection);
+		return _this();
+	}
+
+	public FilterEndpointSpec discardChannel(MessageChannel discardChannel) {
+		this.getHandler().setDiscardChannel(discardChannel);
+		return _this();
+	}
+
+	public FilterEndpointSpec discardWithinAdvice(boolean discardWithinAdvice) {
+		this.getHandler().setDiscardWithinAdvice(discardWithinAdvice);
+		return _this();
+	}
+
+}

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

@@ -0,0 +1,15 @@
+package org.springframework.integration.dsl;
+
+import org.springframework.messaging.MessageHandler;
+
+/**
+ * @author Artem Bilan
+ * @since 4.0
+ */
+public final class GenericEndpointSpec<C extends MessageHandler> extends EndpointSpec<GenericEndpointSpec<C>, C> {
+
+	GenericEndpointSpec(C messageHandler) {
+		super(messageHandler);
+	}
+
+}

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

@@ -0,0 +1,46 @@
+/*
+ * 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;
+
+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>();
+
+	IntegrationFlow() {
+	}
+
+	public Set<AbstractBeanDefinition> getIntegrationComponents() {
+		return integrationComponents;
+	}
+
+	IntegrationFlow addComponent(Object component) {
+		AbstractBeanDefinition beanDefinition = component instanceof AbstractBeanDefinition
+				? (AbstractBeanDefinition) component : new InstanceBeanDefinition(component);
+		this.integrationComponents.add(beanDefinition);
+		return this;
+	}
+
+}

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

@@ -0,0 +1,154 @@
+/*
+ * 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;
+
+import org.springframework.expression.Expression;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean;
+import org.springframework.integration.core.GenericSelector;
+import org.springframework.integration.core.MessageProducer;
+import org.springframework.integration.core.MessageSelector;
+import org.springframework.integration.dsl.support.EndpointConfigurer;
+import org.springframework.integration.filter.ExpressionEvaluatingSelector;
+import org.springframework.integration.filter.MessageFilter;
+import org.springframework.integration.filter.MethodInvokingSelector;
+import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.transformer.ExpressionEvaluatingTransformer;
+import org.springframework.integration.transformer.GenericTransformer;
+import org.springframework.integration.transformer.MessageTransformingHandler;
+import org.springframework.integration.transformer.MethodInvokingTransformer;
+import org.springframework.integration.transformer.Transformer;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * @author Artem Bilan
+ */
+public final class IntegrationFlowBuilder {
+
+	private final static SpelExpressionParser PARSER = new SpelExpressionParser();
+
+	private final IntegrationFlow flow = new IntegrationFlow();
+
+	private MessageChannel currentMessageChannel;
+
+	private Object currentComponent;
+
+	IntegrationFlowBuilder() {
+	}
+
+	IntegrationFlowBuilder addComponent(Object component) {
+		this.flow.addComponent(component);
+		return this;
+	}
+
+	IntegrationFlowBuilder currentComponent(Object component) {
+		this.currentComponent = component;
+		return this;
+	}
+
+	public IntegrationFlowBuilder channel(MessageChannel messageChannel) {
+		this.currentMessageChannel = messageChannel;
+		return this.addComponent(this.currentMessageChannel).registerOutputChannelIfCan(this.currentMessageChannel);
+	}
+
+	private IntegrationFlowBuilder registerOutputChannelIfCan(MessageChannel outputChannel) {
+		this.flow.addComponent(outputChannel);
+		if (this.currentComponent != null) {
+			if (this.currentComponent instanceof MessageProducer) {
+				((MessageProducer) this.currentComponent).setOutputChannel(outputChannel);
+			}
+			if (this.currentComponent instanceof AbstractReplyProducingMessageHandler) {
+				((AbstractReplyProducingMessageHandler) this.currentComponent).setOutputChannel(outputChannel);
+			}
+			else if (this.currentComponent instanceof SourcePollingChannelAdapterFactoryBean) {
+				((SourcePollingChannelAdapterFactoryBean) this.currentComponent).setOutputChannel(outputChannel);
+			}
+			this.currentComponent = null;
+		}
+		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) {
+		MessageChannel inputChannel = this.currentMessageChannel;
+		this.currentMessageChannel = null;
+		if (inputChannel == null) {
+			inputChannel = new DirectChannel();
+			this.registerOutputChannelIfCan(inputChannel);
+		}
+
+		endpointSpec.getEndpoint().setInputChannel(inputChannel);
+
+		return this.addComponent(endpointSpec).currentComponent(endpointSpec.getHandler());
+	}
+
+	public IntegrationFlow get() {
+		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

@@ -0,0 +1,55 @@
+/*
+ * 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;
+
+import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean;
+import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.endpoint.AbstractEndpoint;
+import org.springframework.integration.scheduling.PollerMetadata;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * @author Artem Bilan
+ */
+public final class IntegrationFlows {
+
+	public static IntegrationFlowBuilder from(MessageChannel messageChannel) {
+		return new IntegrationFlowBuilder().channel(messageChannel);
+	}
+
+	public static IntegrationFlowBuilder from(MessageSource<?> messageSource) {
+		return from(messageSource, null);
+	}
+
+	public static IntegrationFlowBuilder from(MessageSource<?> messageSource, PollerMetadata pollerMetadata) {
+		SourcePollingChannelAdapterFactoryBean factoryBean = new SourcePollingChannelAdapterFactoryBean();
+		factoryBean.setSource(messageSource);
+		factoryBean.setPollerMetadata(pollerMetadata);
+		return new IntegrationFlowBuilder()
+				.addComponent(messageSource)
+				.addComponent(factoryBean)
+				.currentComponent(factoryBean);
+	}
+
+	public static IntegrationFlowBuilder from(AbstractEndpoint endpoint) {
+		return new IntegrationFlowBuilder();
+	}
+
+	private IntegrationFlows() {
+	}
+
+}

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

@@ -0,0 +1,66 @@
+/*
+ * 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.channel;
+
+import java.util.Arrays;
+
+import org.springframework.integration.channel.AbstractMessageChannel;
+import org.springframework.messaging.support.ChannelInterceptor;
+
+/**
+ * @author Artem Bilan
+ */
+public abstract class ChannelSpecSupport<S extends ChannelSpecSupport<S, C>, C extends AbstractMessageChannel> {
+
+	protected C channel;
+
+	private String id;
+
+	private Class<?>[] datatypes;
+
+	private ChannelInterceptor[] interceptors;
+
+	public S datatypes(Class<?>... datatypes) {
+		this.datatypes = datatypes;
+		return _this();
+	}
+
+	public S interceptors(ChannelInterceptor... interceptors) {
+		this.interceptors = interceptors;
+		return _this();
+	}
+
+	public S id(String id) {
+		this.id = id;
+		return _this();
+	}
+
+	public C get() {
+		this.channel.setDatatypes(this.datatypes);
+		this.channel.setBeanName(this.id);
+		if (this.interceptors != null) {
+			this.channel.setInterceptors(Arrays.asList(this.interceptors));
+		}
+		return this.channel;
+	}
+
+	@SuppressWarnings("unchecked")
+	protected S _this() {
+		return (S) this;
+	}
+
+}

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

@@ -0,0 +1,40 @@
+/*
+ * 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.channel;
+
+import org.springframework.integration.channel.DirectChannel;
+
+/**
+ * @author Artem Bilan
+ */
+public class DirectChannelSpec extends LoadBalancingChannelSpecSupport<DirectChannelSpec, DirectChannel> {
+
+	@Override
+	public DirectChannel get() {
+		this.channel = new DirectChannel(this.loadBalancingStrategy);
+		if (this.failover != null) {
+			this.channel.setFailover(this.failover);
+		}
+		if (this.maxSubscribers != null) {
+			this.channel.setMaxSubscribers(this.maxSubscribers);
+		}
+		return super.get();
+	}
+
+	DirectChannelSpec() {
+	}
+}

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

@@ -0,0 +1,45 @@
+/*
+ * 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.channel;
+
+import java.util.concurrent.Executor;
+
+import org.springframework.integration.channel.ExecutorChannel;
+
+/**
+ * @author Artem Bilan
+ */
+public class ExecutorChannelSpec extends LoadBalancingChannelSpecSupport<ExecutorChannelSpec, ExecutorChannel> {
+
+	private final Executor executor;
+
+	ExecutorChannelSpec(Executor executor) {
+		this.executor = executor;
+	}
+
+	public ExecutorChannel get() {
+		this.channel = new ExecutorChannel(this.executor, this.loadBalancingStrategy);
+		if (this.failover != null) {
+			this.channel.setFailover(this.failover);
+		}
+		if (this.maxSubscribers != null) {
+			this.channel.setMaxSubscribers(this.maxSubscribers);
+		}
+		return super.get();
+	}
+
+}

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

@@ -0,0 +1,49 @@
+/*
+ * 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.channel;
+
+import org.springframework.integration.channel.AbstractMessageChannel;
+import org.springframework.integration.dispatcher.LoadBalancingStrategy;
+import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
+
+/**
+ * @author Artem Bilan
+ */
+public class LoadBalancingChannelSpecSupport<S extends ChannelSpecSupport<S, C>, C extends AbstractMessageChannel> extends ChannelSpecSupport<S, C> {
+
+	protected LoadBalancingStrategy loadBalancingStrategy = new RoundRobinLoadBalancingStrategy();
+
+	protected Boolean failover;
+
+	protected Integer maxSubscribers;
+
+	public S loadBalancer(LoadBalancingStrategy loadBalancingStrategy) {
+		this.loadBalancingStrategy = loadBalancingStrategy;
+		return _this();
+	}
+
+	public S failover(Boolean failover) {
+		this.failover = failover;
+		return _this();
+	}
+
+	public S maxSubscribers(Integer maxSubscribers) {
+		this.maxSubscribers = maxSubscribers;
+		return _this();
+	}
+
+}

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

@@ -0,0 +1,78 @@
+/*
+ * 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.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;
+
+/**
+ * @author Artem Bilan
+ */
+public final class MessageChannels {
+
+	public static DirectChannelSpec direct() {
+		return new DirectChannelSpec();
+	}
+
+	public static DirectChannelSpec direct(LoadBalancingStrategy loadBalancingStrategy) {
+		return direct().loadBalancer(loadBalancingStrategy);
+	}
+
+	public static QueueChannelSpec queue() {
+		return new QueueChannelSpec();
+	}
+
+	public static QueueChannelSpec queue(BlockingQueue<Message<?>> queue) {
+		return new QueueChannelSpec(queue);
+	}
+
+	public static QueueChannelSpec queue(Integer capacity) {
+		return new QueueChannelSpec(capacity);
+	}
+
+	public static QueueChannelSpec.MessageStoreSpec queue(MessageGroupStore messageGroupStore, Object groupId) {
+		return new QueueChannelSpec.MessageStoreSpec(messageGroupStore, groupId);
+	}
+
+	public static ExecutorChannelSpec executor(Executor executor) {
+		return new ExecutorChannelSpec(executor);
+	}
+
+	public static RendezvousChannelSpec rendezvous() {
+		return new RendezvousChannelSpec();
+	}
+
+	public static PriorityChannelSpec priority() {
+		return new PriorityChannelSpec();
+	}
+
+	public static PublishSubscribeChannelSpec publishSubscribe() {
+		return new PublishSubscribeChannelSpec();
+	}
+
+	public static PublishSubscribeChannelSpec publishSubscribe(Executor executor) {
+		return new PublishSubscribeChannelSpec(executor);
+	}
+
+	private MessageChannels() {
+	}
+
+}

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

@@ -0,0 +1,52 @@
+/*
+ * 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.channel;
+
+import java.util.Comparator;
+
+import org.springframework.integration.channel.PriorityChannel;
+import org.springframework.messaging.Message;
+
+/**
+ * @author Artem Bilan
+ */
+public class PriorityChannelSpec extends ChannelSpecSupport<PriorityChannelSpec, PriorityChannel> {
+
+	private int capacity;
+
+	Comparator<Message<?>> comparator;
+
+	public PriorityChannelSpec setCapacity(int capacity) {
+		this.capacity = capacity;
+		return this;
+	}
+
+	public PriorityChannelSpec setComparator(Comparator<Message<?>> comparator) {
+		this.comparator = comparator;
+		return this;
+	}
+
+	@Override
+	public PriorityChannel get() {
+		this.channel = new PriorityChannel(this.capacity, this.comparator);
+		return super.get();
+	}
+
+
+	PriorityChannelSpec() {
+	}
+}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PublishSubscribeChannelSpec.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.channel;
+
+import java.util.concurrent.Executor;
+
+import org.springframework.integration.channel.PublishSubscribeChannel;
+import org.springframework.util.ErrorHandler;
+
+/**
+ * @author Artem Bilan
+ */
+public class PublishSubscribeChannelSpec extends ChannelSpecSupport<PublishSubscribeChannelSpec, PublishSubscribeChannel> {
+
+	PublishSubscribeChannelSpec() {
+		this.channel = new PublishSubscribeChannel();
+	}
+
+	PublishSubscribeChannelSpec(Executor executor) {
+		this.channel = new PublishSubscribeChannel(executor);
+	}
+
+	PublishSubscribeChannelSpec errorHandler(ErrorHandler errorHandler) {
+		this.channel.setErrorHandler(errorHandler);
+		return this;
+	}
+
+	public PublishSubscribeChannelSpec ignoreFailures(boolean ignoreFailures) {
+		this.channel.setIgnoreFailures(ignoreFailures);
+		return this;
+	}
+
+	public PublishSubscribeChannelSpec applySequence(boolean applySequence) {
+		this.channel.setApplySequence(applySequence);
+		return this;
+	}
+
+	public PublishSubscribeChannelSpec maxSubscribers(Integer maxSubscribers) {
+		this.channel.setMaxSubscribers(maxSubscribers);
+		return this;
+	}
+
+	public PublishSubscribeChannelSpec minSubscribers(int minSubscribers) {
+		this.channel.setMinSubscribers(minSubscribers);
+		return this;
+	}
+
+}

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

@@ -0,0 +1,106 @@
+/*
+ * 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.channel;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.locks.Lock;
+
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.store.MessageGroupQueue;
+import org.springframework.integration.store.MessageGroupStore;
+import org.springframework.messaging.Message;
+
+/**
+ * @author Artem Bilan
+ */
+public class QueueChannelSpec extends ChannelSpecSupport<QueueChannelSpec, QueueChannel> {
+
+	protected BlockingQueue<Message<?>> queue;
+
+	protected Integer capacity;
+
+	QueueChannelSpec() {
+	}
+
+	QueueChannelSpec(BlockingQueue<Message<?>> queue) {
+		this.queue = queue;
+	}
+
+	QueueChannelSpec(Integer capacity) {
+		this.capacity = capacity;
+	}
+
+	@Override
+	public QueueChannel get() {
+		if (this.queue != null) {
+			this.channel = new QueueChannel(this.queue);
+		}
+		else if (this.capacity != null) {
+			this.channel = new QueueChannel(this.capacity);
+		}
+		else {
+			this.channel = new QueueChannel();
+		}
+		return super.get();
+	}
+
+	public static class MessageStoreSpec extends QueueChannelSpec {
+
+		private final MessageGroupStore messageGroupStore;
+
+		private final Object groupId;
+
+		private Lock storeLock;
+
+		MessageStoreSpec(MessageGroupStore messageGroupStore, Object groupId) {
+			super();
+			this.messageGroupStore = messageGroupStore;
+			this.groupId = groupId;
+		}
+
+		public MessageStoreSpec capacity(Integer capacity) {
+			this.capacity = capacity;
+			return this;
+		}
+
+		public MessageStoreSpec storeLock(Lock storeLock) {
+			this.storeLock = storeLock;
+			return this;
+		}
+
+		@Override
+		public QueueChannel get() {
+			if (this.capacity != null) {
+				if (this.storeLock != null) {
+					this.queue = new MessageGroupQueue(messageGroupStore, groupId, this.capacity, this.storeLock);
+				}
+				else {
+					this.queue = new MessageGroupQueue(messageGroupStore, groupId, this.capacity);
+				}
+			}
+			else if (this.storeLock != null) {
+				this.queue = new MessageGroupQueue(messageGroupStore, groupId, this.storeLock);
+			}
+			else {
+				this.queue = new MessageGroupQueue(messageGroupStore, groupId);
+			}
+			return super.get();
+		}
+
+	}
+
+}

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

@@ -0,0 +1,30 @@
+/*
+ * 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.channel;
+
+import org.springframework.integration.channel.RendezvousChannel;
+
+/**
+ * @author Artem Bilan
+ */
+public class RendezvousChannelSpec extends ChannelSpecSupport<RendezvousChannelSpec, RendezvousChannel> {
+
+	RendezvousChannelSpec() {
+		this.channel = new RendezvousChannel();
+	}
+
+}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/InstanceBeanDefinition.java

@@ -0,0 +1,53 @@
+/*
+ * 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.config;
+
+import org.springframework.beans.factory.config.ConstructorArgumentValues;
+import org.springframework.beans.factory.support.GenericBeanDefinition;
+
+/**
+ * @author Artem Bilan
+ */
+@SuppressWarnings("serial")
+public class InstanceBeanDefinition extends GenericBeanDefinition {
+
+	private final Object instance;
+
+	public InstanceBeanDefinition(Object instance) {
+		this.instance = instance;
+		ConstructorArgumentValues args = new ConstructorArgumentValues();
+		args.addGenericArgumentValue(this.instance);
+		this.setConstructorArgumentValues(args);
+		this.setBeanClass(SimpleFactoryBean.class);
+	}
+
+	public InstanceBeanDefinition(InstanceBeanDefinition original) {
+		super(original);
+		this.instance = original.instance;
+	}
+
+	@Override
+	public InstanceBeanDefinition cloneBeanDefinition() {
+		return new InstanceBeanDefinition(this);
+	}
+
+	@Override
+	public Object getSource() {
+		return this.instance;
+	}
+
+}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/SimpleFactoryBean.java

@@ -0,0 +1,95 @@
+/*
+ * 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.config;
+
+import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.beans.factory.config.AbstractFactoryBean;
+import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
+import org.springframework.context.Lifecycle;
+import org.springframework.context.SmartLifecycle;
+
+/**
+ * @author Artem Bilan
+ */
+class SimpleFactoryBean<T> extends AbstractFactoryBean<T> implements BeanNameAware, SmartLifecycle {
+
+	private final T target;
+
+	private String name;
+
+	public SimpleFactoryBean(T target) {
+		this.target = target;
+	}
+
+	@Override
+	public void setBeanName(String name) {
+		this.name = name;
+	}
+
+	@Override
+	public Class<?> getObjectType() {
+		return this.target.getClass();
+	}
+
+	@Override
+	protected T createInstance() throws Exception {
+		((AutowireCapableBeanFactory) this.getBeanFactory()).initializeBean(this.target, this.name);
+		return this.target;
+	}
+
+	@Override
+	public boolean isAutoStartup() {
+		return this.target instanceof SmartLifecycle && ((SmartLifecycle) this.target).isAutoStartup();
+	}
+
+	@Override
+	public void stop(Runnable callback) {
+		if (this.target instanceof SmartLifecycle) {
+			((SmartLifecycle) this.target).stop(callback);
+		}
+	}
+
+	@Override
+	public void start() {
+		if (this.target instanceof Lifecycle) {
+			((Lifecycle) this.target).start();
+		}
+	}
+
+	@Override
+	public void stop() {
+		if (this.target instanceof Lifecycle) {
+			((Lifecycle) this.target).start();
+		}
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.target instanceof SmartLifecycle && ((SmartLifecycle) this.target).isRunning();
+	}
+
+	@Override
+	public int getPhase() {
+		if (this.target instanceof SmartLifecycle) {
+			return ((SmartLifecycle) this.target).getPhase();
+		}
+		else {
+			return 0;
+		}
+	}
+
+}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides configuration classes of the Spring Integration Java DSL.
+ */
+package org.springframework.integration.dsl.config;

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

@@ -0,0 +1,13 @@
+package org.springframework.integration.dsl.support;
+
+import org.springframework.integration.dsl.EndpointSpec;
+
+/**
+ * @author Artem Bilan
+ * @since 4.0
+ */
+public interface EndpointConfigurer<S extends EndpointSpec<?, ?>> {
+
+	void configure(S spec);
+
+}

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

@@ -0,0 +1,122 @@
+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
+ */
+public final class Pollers {
+
+	public static PollerSpec trigger(Trigger trigger) {
+		return new PollerSpec(trigger);
+	}
+
+	public static PollerSpec fixedRate(long period) {
+		return fixedRate(period, null);
+	}
+
+	public static PollerSpec fixedRate(long period, TimeUnit timeUnit) {
+		return periodicTrigger(period, timeUnit, true);
+	}
+
+	public static PollerSpec fixedDelay(long period) {
+		return fixedDelay(period, null);
+	}
+
+	public static PollerSpec fixedDelay(long period, TimeUnit timeUnit) {
+		return periodicTrigger(period, timeUnit, false);
+	}
+
+	private static PollerSpec periodicTrigger(long period, TimeUnit timeUnit, boolean fixedRate) {
+		PeriodicTrigger periodicTrigger = new PeriodicTrigger(period, timeUnit);
+		periodicTrigger.setFixedRate(fixedRate);
+		return new PollerSpec(periodicTrigger);
+	}
+
+	public static PollerSpec cron(String cronExpression) {
+		return cron(cronExpression, TimeZone.getDefault());
+	}
+
+	public static PollerSpec cron(String cronExpression, TimeZone timeZone) {
+		return new PollerSpec(new CronTrigger(cronExpression, timeZone));
+	}
+
+	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/main/resources/META-INF/spring.factories

@@ -0,0 +1,2 @@
+org.springframework.integration.config.IntegrationConfigurationInitializer=\
+org.springframework.integration.dsl.DslIntegrationConfigurationInitializer

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

@@ -0,0 +1,175 @@
+/*
+ * 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.test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.aopalliance.aop.Advice;
+import org.hamcrest.Matchers;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.Lifecycle;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.MessageDispatchingException;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.QueueChannel;
+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.MessageChannels;
+import org.springframework.integration.dsl.support.Pollers;
+import org.springframework.integration.endpoint.MethodInvokingMessageSource;
+import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice;
+import org.springframework.integration.scheduling.PollerMetadata;
+import org.springframework.integration.store.SimpleMessageStore;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.transformer.PayloadDeserializingTransformer;
+import org.springframework.integration.transformer.PayloadSerializingTransformer;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageDeliveryException;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.support.AnnotationConfigContextLoader;
+
+/**
+ * @author Artem Bilan
+ */
+@ContextConfiguration(loader = AnnotationConfigContextLoader.class)
+@RunWith(SpringJUnit4ClassRunner.class)
+public class IntegrationFlowTests {
+
+	@Autowired
+	@Qualifier("flow1QueueChannel")
+	private PollableChannel outputChannel;
+
+	@Autowired
+	private DirectChannel inputChannel;
+
+	@Autowired
+	private PollableChannel successChannel;
+
+	@Autowired
+	private BeanFactory beanFactory;
+
+	@Test
+	public void testPollingFlow() {
+		for (int i = 0; i < 10; i++) {
+			Message<?> message = this.outputChannel.receive(5000);
+			assertNotNull(message);
+			assertEquals("" + i, message.getPayload());
+		}
+	}
+
+	@Test
+	public void testDirectFlow() {
+		assertTrue(this.beanFactory.containsBean("filter"));
+		assertTrue(this.beanFactory.containsBean("filter.handler"));
+		QueueChannel replyChannel = new QueueChannel();
+		Message<String> message = MessageBuilder.withPayload("100").setReplyChannel(replyChannel).build();
+		try {
+			this.inputChannel.send(message);
+			fail("Expected MessageDispatchingException");
+		}
+		catch (Exception e) {
+			assertThat(e, Matchers.instanceOf(MessageDeliveryException.class));
+			assertThat(e.getCause(), Matchers.instanceOf(MessageDispatchingException.class));
+			assertThat(e.getMessage(), Matchers.containsString("Dispatcher has no subscribers"));
+		}
+		this.beanFactory.getBean("payloadSerializingTransformer", Lifecycle.class).start();
+		this.inputChannel.send(message);
+		Message<?> reply = replyChannel.receive(5000);
+		assertNotNull(reply);
+		assertEquals(200, reply.getPayload());
+
+		Message<?> successMessage = this.successChannel.receive(5000);
+		assertNotNull(successMessage);
+		assertEquals(100, successMessage.getPayload());
+	}
+
+
+	@Configuration
+	@EnableIntegration
+	public static class ContextConfiguration {
+
+		@Bean
+		public MessageSource<?> integerMessageSource() {
+			MethodInvokingMessageSource source = new MethodInvokingMessageSource();
+			source.setObject(new AtomicInteger());
+			source.setMethodName("getAndIncrement");
+			return source;
+		}
+
+		@Bean
+		public IntegrationFlow flow1() {
+			return IntegrationFlows.from(this.integerMessageSource(), Pollers.fixedRate(100).get())
+					.transform("payload.toString()")
+					.channel(MessageChannels.queue().id("flow1QueueChannel").get())
+					.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();
+		}
+
+		@Bean
+		public Advice expressionAdvice() {
+			ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
+			advice.setOnSuccessExpression("payload");
+			advice.setSuccessChannel(this.successChannel());
+			return advice;
+		}
+
+		@Bean
+		public IntegrationFlow flow2() {
+			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())
+					.transform(new PayloadDeserializingTransformer())
+					.transform((Integer p) -> p * 2, c -> c.advice(this.expressionAdvice()))
+					.get();
+		}
+
+	}
+
+}