Java DSL: Lambdas for Namespace Factories

* Add `Function` and `Consumer` copies from Java 8
* Rework some functional interfaces to `Function` or `Consumer`
* Introduce `Channels`, `MessageHandlers`, `MessageProducers`, `MessageSources`, `MessagingGateways`
- some convenient wrappers for Namespace Factories to be used from Lambdas
* Add Lambda-methods for those new wrappers
* Provide some refactoring
* Upgrade to Boot `1.1.7`

Java DSL: Lambda_Factories

* Change all `IntegrationFlows` methods to `.from()`
* Introduce internal `Function<?, ?>` extensions for particular cases, e.g. `MessageSourcesFunction`.
Now end-user has to cast Lambda parameter to concrete type to get desired factory, e.g.:
```
return IntegrationFlows.from((MessageProducers mp) -> mp.imap("imap://user:pw@host/INBOX"))
```
* rename `.fromFixedMessageChannel` to `.from()` with additional boolean flag
* Introduce `PollerFactory` Lambda for `.poller()` EIP-method on `EndpointSpec`
* Move `Pollers` stuff to the `.core` package to fix package tangle
* Refactoring for some code style
* Fix some typos

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

@@ -0,0 +1,151 @@
+/*
+ * 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.io.File;
+
+import javax.jms.ConnectionFactory;
+
+import org.apache.commons.net.ftp.FTPFile;
+
+import org.springframework.amqp.core.AmqpTemplate;
+import org.springframework.integration.dsl.amqp.Amqp;
+import org.springframework.integration.dsl.amqp.AmqpOutboundEndpointSpec;
+import org.springframework.integration.dsl.file.FileWritingMessageHandlerSpec;
+import org.springframework.integration.dsl.file.Files;
+import org.springframework.integration.dsl.ftp.Ftp;
+import org.springframework.integration.dsl.ftp.FtpMessageHandlerSpec;
+import org.springframework.integration.dsl.ftp.FtpOutboundGatewaySpec;
+import org.springframework.integration.dsl.jms.Jms;
+import org.springframework.integration.dsl.jms.JmsOutboundChannelAdapterSpec;
+import org.springframework.integration.dsl.jms.JmsOutboundGatewaySpec;
+import org.springframework.integration.dsl.mail.Mail;
+import org.springframework.integration.dsl.mail.MailSendingMessageHandlerSpec;
+import org.springframework.integration.dsl.sftp.Sftp;
+import org.springframework.integration.dsl.sftp.SftpMessageHandlerSpec;
+import org.springframework.integration.dsl.sftp.SftpOutboundGatewaySpec;
+import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.file.support.FileExistsMode;
+import org.springframework.jms.core.JmsTemplate;
+
+import com.jcraft.jsch.ChannelSftp;
+
+/**
+ * @author Artem Bilan
+ */
+public class Adapters {
+
+	public AmqpOutboundEndpointSpec amqp(AmqpTemplate amqpTemplate) {
+		return Amqp.outboundAdapter(amqpTemplate);
+	}
+
+	public AmqpOutboundEndpointSpec amqpGateway(AmqpTemplate amqpTemplate) {
+		return Amqp.outboundGateway(amqpTemplate);
+	}
+
+	public FileWritingMessageHandlerSpec file(File destinationDirectory) {
+		return Files.outboundAdapter(destinationDirectory);
+	}
+
+	public FileWritingMessageHandlerSpec file(String directoryExpression) {
+		return Files.outboundAdapter(directoryExpression);
+	}
+
+	public FileWritingMessageHandlerSpec fileGateway(File destinationDirectory) {
+		return Files.outboundGateway(destinationDirectory);
+	}
+
+	public FileWritingMessageHandlerSpec fileGateway(String directoryExpression) {
+		return Files.outboundGateway(directoryExpression);
+	}
+
+	public FtpMessageHandlerSpec ftp(SessionFactory<FTPFile> sessionFactory) {
+		return Ftp.outboundAdapter(sessionFactory);
+	}
+
+	public FtpMessageHandlerSpec ftp(SessionFactory<FTPFile> sessionFactory, FileExistsMode fileExistsMode) {
+		return Ftp.outboundAdapter(sessionFactory, fileExistsMode);
+	}
+
+	public FtpMessageHandlerSpec ftp(RemoteFileTemplate<FTPFile> remoteFileTemplate) {
+		return Ftp.outboundAdapter(remoteFileTemplate);
+	}
+
+	public FtpMessageHandlerSpec ftp(RemoteFileTemplate<FTPFile> remoteFileTemplate, FileExistsMode fileExistsMode) {
+		return Ftp.outboundAdapter(remoteFileTemplate, fileExistsMode);
+	}
+
+	public FtpOutboundGatewaySpec ftpGateway(SessionFactory<FTPFile> sessionFactory,
+			AbstractRemoteFileOutboundGateway.Command command, String expression) {
+		return Ftp.outboundGateway(sessionFactory, command, expression);
+	}
+
+	public FtpOutboundGatewaySpec ftpGateway(SessionFactory<FTPFile> sessionFactory, String command,
+			String expression) {
+		return Ftp.outboundGateway(sessionFactory, command, expression);
+	}
+
+	public SftpMessageHandlerSpec ftps(SessionFactory<ChannelSftp.LsEntry> sessionFactory) {
+		return Sftp.outboundAdapter(sessionFactory);
+	}
+
+	public SftpMessageHandlerSpec sftp(SessionFactory<ChannelSftp.LsEntry> sessionFactory,
+			FileExistsMode fileExistsMode) {
+		return Sftp.outboundAdapter(sessionFactory, fileExistsMode);
+	}
+
+	public SftpMessageHandlerSpec sftp(RemoteFileTemplate<ChannelSftp.LsEntry> remoteFileTemplate) {
+		return Sftp.outboundAdapter(remoteFileTemplate);
+	}
+
+	public SftpMessageHandlerSpec sftp(RemoteFileTemplate<ChannelSftp.LsEntry> remoteFileTemplate,
+			FileExistsMode fileExistsMode) {
+		return Sftp.outboundAdapter(remoteFileTemplate, fileExistsMode);
+	}
+
+	public SftpOutboundGatewaySpec sftpGateway(SessionFactory<ChannelSftp.LsEntry> sessionFactory,
+			AbstractRemoteFileOutboundGateway.Command command, String expression) {
+		return Sftp.outboundGateway(sessionFactory, command, expression);
+	}
+
+	public SftpOutboundGatewaySpec sftpGateway(SessionFactory<ChannelSftp.LsEntry> sessionFactory, String command,
+			String expression) {
+		return Sftp.outboundGateway(sessionFactory, command, expression);
+	}
+
+	public JmsOutboundChannelAdapterSpec.JmsOutboundChannelSpecTemplateAware jms(ConnectionFactory connectionFactory) {
+		return Jms.outboundAdapter(connectionFactory);
+	}
+
+	public JmsOutboundChannelAdapterSpec<? extends JmsOutboundChannelAdapterSpec<?>> jms(JmsTemplate jmsTemplate) {
+		return Jms.outboundAdapter(jmsTemplate);
+	}
+
+	public JmsOutboundGatewaySpec jmsGateway(ConnectionFactory connectionFactory) {
+		return Jms.outboundGateway(connectionFactory);
+	}
+
+	public MailSendingMessageHandlerSpec mail(String host) {
+		return Mail.outboundAdapter(host);
+	}
+
+	Adapters() {
+	}
+
+}

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

@@ -0,0 +1,197 @@
+/*
+ * 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.concurrent.BlockingQueue;
+import java.util.concurrent.Executor;
+
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.integration.dsl.amqp.Amqp;
+import org.springframework.integration.dsl.amqp.AmqpMessageChannelSpec;
+import org.springframework.integration.dsl.amqp.AmqpPollableMessageChannelSpec;
+import org.springframework.integration.dsl.amqp.AmqpPublishSubscribeMessageChannelSpec;
+import org.springframework.integration.dsl.channel.DirectChannelSpec;
+import org.springframework.integration.dsl.channel.ExecutorChannelSpec;
+import org.springframework.integration.dsl.channel.MessageChannels;
+import org.springframework.integration.dsl.channel.PriorityChannelSpec;
+import org.springframework.integration.dsl.channel.PublishSubscribeChannelSpec;
+import org.springframework.integration.dsl.channel.QueueChannelSpec;
+import org.springframework.integration.dsl.channel.RendezvousChannelSpec;
+import org.springframework.integration.dsl.jms.Jms;
+import org.springframework.integration.dsl.jms.JmsMessageChannelSpec;
+import org.springframework.integration.dsl.jms.JmsPollableMessageChannelSpec;
+import org.springframework.integration.dsl.jms.JmsPublishSubscribeMessageChannelSpec;
+import org.springframework.integration.store.ChannelMessageStore;
+import org.springframework.integration.store.PriorityCapableChannelMessageStore;
+import org.springframework.messaging.Message;
+
+/**
+ * @author Artem Bilan
+ */
+public class Channels {
+
+	public DirectChannelSpec direct() {
+		return MessageChannels.direct();
+	}
+
+	public DirectChannelSpec direct(String id) {
+		return MessageChannels.direct(id);
+	}
+
+	public QueueChannelSpec queue() {
+		return MessageChannels.queue();
+	}
+
+	public QueueChannelSpec queue(String id) {
+		return MessageChannels.queue(id);
+	}
+
+	public QueueChannelSpec queue(Integer capacity) {
+		return MessageChannels.queue(capacity);
+	}
+
+	public QueueChannelSpec queue(String id, Integer capacity) {
+		return MessageChannels.queue(id, capacity);
+	}
+
+	public QueueChannelSpec queue(BlockingQueue<Message<?>> queue) {
+		return MessageChannels.queue(queue);
+	}
+
+	public QueueChannelSpec queue(String id, BlockingQueue<Message<?>> queue) {
+		return MessageChannels.queue(id, queue);
+	}
+
+	public QueueChannelSpec.MessageStoreSpec queue(ChannelMessageStore messageGroupStore, Object groupId) {
+		return MessageChannels.queue(messageGroupStore, groupId);
+	}
+
+	public QueueChannelSpec.MessageStoreSpec queue(String id, ChannelMessageStore messageGroupStore, Object groupId) {
+		return MessageChannels.queue(id, messageGroupStore, groupId);
+	}
+
+	public PriorityChannelSpec priority() {
+		return MessageChannels.priority();
+	}
+
+	public PriorityChannelSpec priority(String id) {
+		return MessageChannels.priority(id);
+	}
+
+	public QueueChannelSpec.MessageStoreSpec priority(String id, PriorityCapableChannelMessageStore messageGroupStore,
+			Object groupId) {
+		return MessageChannels.priority(id, messageGroupStore, groupId);
+	}
+
+	public QueueChannelSpec.MessageStoreSpec priority(PriorityCapableChannelMessageStore messageGroupStore,
+			Object groupId) {
+		return MessageChannels.priority(messageGroupStore, groupId);
+	}
+
+	public RendezvousChannelSpec rendezvous() {
+		return MessageChannels.rendezvous();
+	}
+
+	public RendezvousChannelSpec rendezvous(String id) {
+		return MessageChannels.rendezvous(id);
+	}
+
+	public PublishSubscribeChannelSpec publishSubscribe() {
+		return MessageChannels.publishSubscribe();
+	}
+
+	public PublishSubscribeChannelSpec publishSubscribe(Executor executor) {
+		return MessageChannels.publishSubscribe(executor);
+	}
+
+	public ExecutorChannelSpec executor(Executor executor) {
+		return MessageChannels.executor(executor);
+	}
+
+	public ExecutorChannelSpec executor(String id, Executor executor) {
+		return MessageChannels.executor(id, executor);
+	}
+
+	public PublishSubscribeChannelSpec publishSubscribe(String id, Executor executor) {
+		return MessageChannels.publishSubscribe(id, executor);
+	}
+
+	public PublishSubscribeChannelSpec publishSubscribe(String id) {
+		return MessageChannels.publishSubscribe(id);
+	}
+
+	public AmqpPollableMessageChannelSpec<? extends AmqpPollableMessageChannelSpec<?>> amqpPollable(
+			ConnectionFactory connectionFactory) {
+		return Amqp.pollableChannel(connectionFactory);
+	}
+
+	public AmqpPollableMessageChannelSpec<? extends AmqpPollableMessageChannelSpec<?>> amqpPollable(String id,
+			ConnectionFactory connectionFactory) {
+		return Amqp.pollableChannel(id, connectionFactory);
+	}
+
+	public AmqpMessageChannelSpec<? extends AmqpMessageChannelSpec<?>> amqp(ConnectionFactory connectionFactory) {
+		return Amqp.channel(connectionFactory);
+	}
+
+	public AmqpMessageChannelSpec<? extends AmqpMessageChannelSpec<?>> amqp(String id,
+			ConnectionFactory connectionFactory) {
+		return Amqp.channel(id, connectionFactory);
+	}
+
+	public static AmqpPublishSubscribeMessageChannelSpec amqpPublishSubscribe(ConnectionFactory connectionFactory) {
+		return Amqp.publishSubscribeChannel(connectionFactory);
+	}
+
+	public static AmqpPublishSubscribeMessageChannelSpec amqpPublishSubscribe(String id,
+			ConnectionFactory connectionFactory) {
+		return Amqp.publishSubscribeChannel(id, connectionFactory);
+	}
+
+	public JmsPollableMessageChannelSpec<? extends JmsPollableMessageChannelSpec<?>> jmsPollable(
+			javax.jms.ConnectionFactory connectionFactory) {
+		return Jms.pollableChannel(connectionFactory);
+	}
+
+	public JmsPollableMessageChannelSpec<? extends JmsPollableMessageChannelSpec<?>> jmsPollable(String id,
+			javax.jms.ConnectionFactory connectionFactory) {
+		return Jms.pollableChannel(id, connectionFactory);
+	}
+
+	public JmsMessageChannelSpec<? extends JmsMessageChannelSpec<?>> jms(
+			javax.jms.ConnectionFactory connectionFactory) {
+		return Jms.channel(connectionFactory);
+	}
+
+	public JmsMessageChannelSpec<? extends JmsMessageChannelSpec<?>> jms(String id,
+			javax.jms.ConnectionFactory connectionFactory) {
+		return Jms.channel(id, connectionFactory);
+	}
+
+	public JmsPublishSubscribeMessageChannelSpec jmsPublishSubscribe(javax.jms.ConnectionFactory connectionFactory) {
+		return Jms.publishSubscribeChannel(connectionFactory);
+	}
+
+	public JmsPublishSubscribeMessageChannelSpec jmsPublishSubscribe(String id,
+			javax.jms.ConnectionFactory connectionFactory) {
+		return Jms.publishSubscribeChannel(id, connectionFactory);
+	}
+
+	Channels() {
+	}
+
+}

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

@@ -34,7 +34,8 @@ import org.springframework.util.StringUtils;
 /**
  * @author Artem Bilan
  */
-public abstract class CorrelationHandlerSpec<S extends CorrelationHandlerSpec<S, H>, H extends AbstractCorrelatingMessageHandler>
+public abstract class
+		CorrelationHandlerSpec<S extends CorrelationHandlerSpec<S, H>, H extends AbstractCorrelatingMessageHandler>
 		extends MessageHandlerSpec<S, H> {
 
 	protected MessageGroupStore messageStore;

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

@@ -69,7 +69,8 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec<HeaderEnricherS
 	}
 
 	public HeaderEnricherSpec messageProcessor(String expression) {
-		return this.messageProcessor(new ExpressionEvaluatingMessageProcessor<Object>(PARSER.parseExpression(expression)));
+		return this.messageProcessor(new ExpressionEvaluatingMessageProcessor<Object>(
+				PARSER.parseExpression(expression)));
 	}
 
 	public HeaderEnricherSpec messageProcessor(String beanName, String methodName) {
@@ -99,7 +100,8 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec<HeaderEnricherS
 		return headerExpressions(headers.get());
 	}
 
-	public HeaderEnricherSpec headerExpressions(MapBuilderConfigurer<StringStringMapBuilder, String, String> configurer) {
+	public HeaderEnricherSpec headerExpressions(
+			MapBuilderConfigurer<StringStringMapBuilder, String, String> configurer) {
 		StringStringMapBuilder builder = new StringStringMapBuilder();
 		configurer.configure(builder);
 		return headerExpressions(builder.get());

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

@@ -37,8 +37,8 @@ public final class IntegrationFlowBuilder extends IntegrationFlowDefinition<Inte
 		if (this.integrationComponents.size() == 1) {
 			if (this.currentComponent != null) {
 				if (this.currentComponent instanceof SourcePollingChannelAdapterSpec) {
-					throw new BeanCreationException("The 'SourcePollingChannelAdapter' (" + this.currentComponent + ") " +
-							"must be configured with at least one 'MessageChanel' or 'MessageHandler'.");
+					throw new BeanCreationException("The 'SourcePollingChannelAdapter' (" + this.currentComponent
+							+ ") " + "must be configured with at least one 'MessageChanel' or 'MessageHandler'.");
 				}
 			}
 			else if (this.currentMessageChannel != null) {

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

@@ -37,9 +37,9 @@ import org.springframework.integration.dsl.channel.MessageChannelSpec;
 import org.springframework.integration.dsl.core.ConsumerEndpointSpec;
 import org.springframework.integration.dsl.core.MessageHandlerSpec;
 import org.springframework.integration.dsl.support.BeanNameMessageProcessor;
-import org.springframework.integration.dsl.support.ComponentConfigurer;
-import org.springframework.integration.dsl.support.EndpointConfigurer;
+import org.springframework.integration.dsl.support.Consumer;
 import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype;
+import org.springframework.integration.dsl.support.Function;
 import org.springframework.integration.dsl.support.GenericHandler;
 import org.springframework.integration.dsl.support.GenericRouter;
 import org.springframework.integration.dsl.support.GenericSplitter;
@@ -135,6 +135,11 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 		return this.registerOutputChannelIfCan(this.currentMessageChannel);
 	}
 
+	public B channel(Function<Channels, MessageChannelSpec<?, ?>> channels) {
+		Assert.notNull(channels);
+		return channel(channels.apply(new Channels()));
+	}
+
 	public B channel(MessageChannelSpec<?, ?> messageChannelSpec) {
 		Assert.notNull(messageChannelSpec);
 		return this.channel(messageChannelSpec.get());
@@ -144,9 +149,9 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 		return controlBus(null);
 	}
 
-	public B controlBus(EndpointConfigurer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
-		return this.handle(new ServiceActivatingHandler(new ExpressionCommandMessageProcessor(new ControlBusMethodFilter())),
-				endpointConfigurer);
+	public B controlBus(Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
+		return this.handle(new ServiceActivatingHandler(new ExpressionCommandMessageProcessor(
+						new ControlBusMethodFilter())), endpointConfigurer);
 	}
 
 	public B transform(String expression) {
@@ -163,12 +168,12 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public <S, T> B transform(GenericTransformer<S, T> genericTransformer,
-			EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
+			Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 		return this.transform(null, genericTransformer, endpointConfigurer);
 	}
 
 	public <P, T> B transform(Class<P> payloadType, GenericTransformer<P, T> genericTransformer,
-			EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
+			Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 		Assert.notNull(genericTransformer);
 		Transformer transformer = genericTransformer instanceof Transformer ? (Transformer) genericTransformer :
 				(isLambda(genericTransformer)
@@ -192,12 +197,12 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public <P> B filter(GenericSelector<P> genericSelector,
-			EndpointConfigurer<FilterEndpointSpec> endpointConfigurer) {
+			Consumer<FilterEndpointSpec> endpointConfigurer) {
 		return filter(null, genericSelector, endpointConfigurer);
 	}
 
 	public <P> B filter(Class<P> payloadType, GenericSelector<P> genericSelector,
-			EndpointConfigurer<FilterEndpointSpec> endpointConfigurer) {
+			Consumer<FilterEndpointSpec> endpointConfigurer) {
 		Assert.notNull(genericSelector);
 		MessageSelector selector = genericSelector instanceof MessageSelector ? (MessageSelector) genericSelector :
 				(isLambda(genericSelector)
@@ -206,9 +211,19 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 		return this.register(new FilterEndpointSpec(new MessageFilter(selector)), endpointConfigurer);
 	}
 
-	public <S extends MessageHandlerSpec<S, ? extends MessageHandler>> B handle(S messageHandlerSpec) {
-		Assert.notNull(messageHandlerSpec);
-		return handle(messageHandlerSpec.get());
+	public <H extends MessageHandler> B handleWithAdapter(
+			Function<Adapters, MessageHandlerSpec<?, H>> adapters) {
+		return handleWithAdapter(adapters, null);
+	}
+
+	public <H extends MessageHandler> B handleWithAdapter(
+			Function<Adapters, MessageHandlerSpec<?, H>> adapters,
+			Consumer<GenericEndpointSpec<H>> endpointConfigurer) {
+		return handle(adapters.apply(new Adapters()), endpointConfigurer);
+	}
+
+	public B handle(MessageHandlerSpec<?, ? extends MessageHandler> messageHandlerSpec) {
+		return handle(messageHandlerSpec, null);
 	}
 
 	public B handle(MessageHandler messageHandler) {
@@ -220,7 +235,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public B handle(String beanName, String methodName,
-			EndpointConfigurer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
+			Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
 		return this.handle(new ServiceActivatingHandler(new BeanNameMessageProcessor<Object>(beanName, methodName)),
 				endpointConfigurer);
 	}
@@ -230,7 +245,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public <P> B handle(GenericHandler<P> handler,
-			EndpointConfigurer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
+			Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
 		return this.handle(null, handler, endpointConfigurer);
 	}
 
@@ -239,7 +254,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public <P> B handle(Class<P> payloadType, GenericHandler<P> handler,
-			EndpointConfigurer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
+			Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
 		ServiceActivatingHandler serviceActivatingHandler = null;
 		if (isLambda(handler)) {
 			serviceActivatingHandler = new ServiceActivatingHandler(new LambdaMessageProcessor(handler, payloadType));
@@ -250,19 +265,19 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 		return this.handle(serviceActivatingHandler, endpointConfigurer);
 	}
 
-	public <H extends MessageHandler, S extends MessageHandlerSpec<S, H>>
-	B handle(S messageHandlerSpec, EndpointConfigurer<GenericEndpointSpec<H>> endpointConfigurer) {
+	public <H extends MessageHandler> B handle(MessageHandlerSpec<?, H> messageHandlerSpec,
+			Consumer<GenericEndpointSpec<H>> endpointConfigurer) {
 		Assert.notNull(messageHandlerSpec);
 		return handle(messageHandlerSpec.get(), endpointConfigurer);
 	}
 
 	public <H extends MessageHandler> B handle(H messageHandler,
-			EndpointConfigurer<GenericEndpointSpec<H>> endpointConfigurer) {
+			Consumer<GenericEndpointSpec<H>> endpointConfigurer) {
 		Assert.notNull(messageHandler);
 		return this.register(new GenericEndpointSpec<H>(messageHandler), endpointConfigurer);
 	}
 
-	public B bridge(EndpointConfigurer<GenericEndpointSpec<BridgeHandler>> endpointConfigurer) {
+	public B bridge(Consumer<GenericEndpointSpec<BridgeHandler>> endpointConfigurer) {
 		return this.register(new GenericEndpointSpec<BridgeHandler>(new BridgeHandler()), endpointConfigurer);
 	}
 
@@ -271,7 +286,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public B delay(String groupId, String expression,
-			EndpointConfigurer<DelayerEndpointSpec> endpointConfigurer) {
+			Consumer<DelayerEndpointSpec> endpointConfigurer) {
 		DelayHandler delayHandler = new DelayHandler(groupId);
 		if (StringUtils.hasText(expression)) {
 			delayHandler.setDelayExpression(PARSER.parseExpression(expression));
@@ -279,15 +294,15 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 		return this.register(new DelayerEndpointSpec(delayHandler), endpointConfigurer);
 	}
 
-	public B enrich(ComponentConfigurer<EnricherSpec> enricherConfigurer) {
+	public B enrich(Consumer<EnricherSpec> enricherConfigurer) {
 		return this.enrich(enricherConfigurer, null);
 	}
 
-	public B enrich(ComponentConfigurer<EnricherSpec> enricherConfigurer,
-			EndpointConfigurer<GenericEndpointSpec<ContentEnricher>> endpointConfigurer) {
+	public B enrich(Consumer<EnricherSpec> enricherConfigurer,
+			Consumer<GenericEndpointSpec<ContentEnricher>> endpointConfigurer) {
 		Assert.notNull(enricherConfigurer);
 		EnricherSpec enricherSpec = new EnricherSpec();
-		enricherConfigurer.configure(enricherSpec);
+		enricherConfigurer.accept(enricherSpec);
 		return this.handle(enricherSpec.get(), endpointConfigurer);
 	}
 
@@ -296,7 +311,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public B enrichHeaders(MapBuilder<?, String, Object> headers,
-			EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
+			Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 		return enrichHeaders(headers.get(), endpointConfigurer);
 	}
 
@@ -313,33 +328,35 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public B enrichHeaders(final Map<String, Object> headers,
-			EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
-		return enrichHeaders(new ComponentConfigurer<HeaderEnricherSpec>() {
+			Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
+		return enrichHeaders(new Consumer<HeaderEnricherSpec>() {
+
 			@Override
-			public void configure(HeaderEnricherSpec spec) {
+			public void accept(HeaderEnricherSpec spec) {
 				spec.headers(headers);
 			}
+
 		}, endpointConfigurer);
 	}
 
-	public B enrichHeaders(ComponentConfigurer<HeaderEnricherSpec> headerEnricherConfigurer) {
+	public B enrichHeaders(Consumer<HeaderEnricherSpec> headerEnricherConfigurer) {
 		return this.enrichHeaders(headerEnricherConfigurer, null);
 	}
 
-	public B enrichHeaders(ComponentConfigurer<HeaderEnricherSpec> headerEnricherConfigurer,
-			EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
+	public B enrichHeaders(Consumer<HeaderEnricherSpec> headerEnricherConfigurer,
+			Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 		Assert.notNull(headerEnricherConfigurer);
 		HeaderEnricherSpec headerEnricherSpec = new HeaderEnricherSpec();
-		headerEnricherConfigurer.configure(headerEnricherSpec);
+		headerEnricherConfigurer.accept(headerEnricherSpec);
 		return transform(headerEnricherSpec.get(), endpointConfigurer);
 	}
 
-	public B split(EndpointConfigurer<SplitterEndpointSpec<DefaultMessageSplitter>> endpointConfigurer) {
+	public B split(Consumer<SplitterEndpointSpec<DefaultMessageSplitter>> endpointConfigurer) {
 		return this.split(new DefaultMessageSplitter(), endpointConfigurer);
 	}
 
 	public B split(String expression,
-			EndpointConfigurer<SplitterEndpointSpec<ExpressionEvaluatingSplitter>> endpointConfigurer) {
+			Consumer<SplitterEndpointSpec<ExpressionEvaluatingSplitter>> endpointConfigurer) {
 		return this.split(new ExpressionEvaluatingSplitter(PARSER.parseExpression(expression)), endpointConfigurer);
 	}
 
@@ -348,7 +365,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public B split(String beanName, String methodName,
-			EndpointConfigurer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
+			Consumer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
 		return this.split(new MethodInvokingSplitter(new BeanNameMessageProcessor<Collection<?>>(beanName, methodName)),
 				endpointConfigurer);
 	}
@@ -358,12 +375,12 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public <T> B split(GenericSplitter<T> splitter,
-			EndpointConfigurer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
+			Consumer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
 		return split(null, splitter, endpointConfigurer);
 	}
 
 	public <P> B split(Class<P> payloadType, GenericSplitter<P> splitter,
-			EndpointConfigurer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
+			Consumer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
 		MethodInvokingSplitter split = isLambda(splitter)
 				? new MethodInvokingSplitter(new LambdaMessageProcessor(splitter, payloadType))
 				: new MethodInvokingSplitter(splitter, "split");
@@ -371,13 +388,13 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public <S extends AbstractMessageSplitter> B split(S splitter,
-			EndpointConfigurer<SplitterEndpointSpec<S>> endpointConfigurer) {
+			Consumer<SplitterEndpointSpec<S>> endpointConfigurer) {
 		Assert.notNull(splitter);
 		return this.register(new SplitterEndpointSpec<S>(splitter), endpointConfigurer);
 	}
 
 	/**
-	 * Provides the {@link HeaderFilter} to the current {@link org.springframework.integration.dsl.IntegrationFlowBuilder.StandardIntegrationFlow}.
+	 * Provides the {@link HeaderFilter} to the current {@link IntegrationFlowBuilder.StandardIntegrationFlow}.
 	 * @param headersToRemove the array of headers (or patterns)
 	 * to remove from {@link org.springframework.messaging.MessageHeaders}.
 	 * @return this {@link IntegrationFlowDefinition}.
@@ -387,7 +404,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	/**
-	 * Provides the {@link HeaderFilter} to the current {@link org.springframework.integration.dsl.IntegrationFlowBuilder.StandardIntegrationFlow}.
+	 * Provides the {@link HeaderFilter} to the current {@link IntegrationFlowBuilder.StandardIntegrationFlow}.
 	 * @param headersToRemove the comma separated headers (or patterns) to remove from
 	 * {@link org.springframework.messaging.MessageHeaders}.
 	 * @param patternMatch    the {@code boolean} flag to indicate if {@code headersToRemove}
@@ -401,7 +418,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public B headerFilter(HeaderFilter headerFilter,
-			EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
+			Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 		return this.transform(headerFilter, endpointConfigurer);
 	}
 
@@ -410,7 +427,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public B claimCheckIn(MessageStore messageStore,
-			EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
+			Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 		return this.transform(new ClaimCheckInTransformer(messageStore), endpointConfigurer);
 	}
 
@@ -423,42 +440,42 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public B claimCheckOut(MessageStore messageStore, boolean removeMessage,
-			EndpointConfigurer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
+			Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
 		ClaimCheckOutTransformer claimCheckOutTransformer = new ClaimCheckOutTransformer(messageStore);
 		claimCheckOutTransformer.setRemoveMessage(removeMessage);
 		return this.transform(claimCheckOutTransformer, endpointConfigurer);
 	}
 
 	public B resequence() {
-		return this.resequence((EndpointConfigurer<GenericEndpointSpec<ResequencingMessageHandler>>) null);
+		return this.resequence((Consumer<GenericEndpointSpec<ResequencingMessageHandler>>) null);
 	}
 
-	public B resequence(EndpointConfigurer<GenericEndpointSpec<ResequencingMessageHandler>> endpointConfigurer) {
+	public B resequence(Consumer<GenericEndpointSpec<ResequencingMessageHandler>> endpointConfigurer) {
 		return this.handle(new ResequencerSpec().get(), endpointConfigurer);
 	}
 
-	public B resequence(ComponentConfigurer<ResequencerSpec> resequencerConfigurer,
-			EndpointConfigurer<GenericEndpointSpec<ResequencingMessageHandler>> endpointConfigurer) {
+	public B resequence(Consumer<ResequencerSpec> resequencerConfigurer,
+			Consumer<GenericEndpointSpec<ResequencingMessageHandler>> endpointConfigurer) {
 		Assert.notNull(resequencerConfigurer);
 		ResequencerSpec spec = new ResequencerSpec();
-		resequencerConfigurer.configure(spec);
+		resequencerConfigurer.accept(spec);
 		return this.handle(spec.get(), endpointConfigurer);
 	}
 
 	public B aggregate() {
-		return aggregate((EndpointConfigurer<GenericEndpointSpec<AggregatingMessageHandler>>) null);
+		return aggregate((Consumer<GenericEndpointSpec<AggregatingMessageHandler>>) null);
 	}
 
 	public B
-	aggregate(EndpointConfigurer<GenericEndpointSpec<AggregatingMessageHandler>> endpointConfigurer) {
+	aggregate(Consumer<GenericEndpointSpec<AggregatingMessageHandler>> endpointConfigurer) {
 		return handle(new AggregatorSpec().get(), endpointConfigurer);
 	}
 
-	public B aggregate(ComponentConfigurer<AggregatorSpec> aggregatorConfigurer,
-			EndpointConfigurer<GenericEndpointSpec<AggregatingMessageHandler>> endpointConfigurer) {
+	public B aggregate(Consumer<AggregatorSpec> aggregatorConfigurer,
+			Consumer<GenericEndpointSpec<AggregatingMessageHandler>> endpointConfigurer) {
 		Assert.notNull(aggregatorConfigurer);
 		AggregatorSpec spec = new AggregatorSpec();
-		aggregatorConfigurer.configure(spec);
+		aggregatorConfigurer.accept(spec);
 		return this.handle(spec.get(), endpointConfigurer);
 	}
 
@@ -467,29 +484,27 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public B route(String beanName, String method,
-			ComponentConfigurer<RouterSpec<MethodInvokingRouter>> routerConfigurer) {
+			Consumer<RouterSpec<MethodInvokingRouter>> routerConfigurer) {
 		return this.route(beanName, method, routerConfigurer, null);
 	}
 
-	public B route(String beanName, String method,
-			ComponentConfigurer<RouterSpec<MethodInvokingRouter>> routerConfigurer,
-			EndpointConfigurer<GenericEndpointSpec<MethodInvokingRouter>> endpointConfigurer) {
+	public B route(String beanName, String method, Consumer<RouterSpec<MethodInvokingRouter>> routerConfigurer,
+			Consumer<GenericEndpointSpec<MethodInvokingRouter>> endpointConfigurer) {
 		return this.route(new MethodInvokingRouter(new BeanNameMessageProcessor<Object>(beanName, method)),
 				routerConfigurer, endpointConfigurer);
 	}
 
 	public B route(String expression) {
-		return this.route(expression, (ComponentConfigurer<RouterSpec<ExpressionEvaluatingRouter>>) null);
+		return this.route(expression, (Consumer<RouterSpec<ExpressionEvaluatingRouter>>) null);
 	}
 
-	public B route(String expression,
-			ComponentConfigurer<RouterSpec<ExpressionEvaluatingRouter>> routerConfigurer) {
+	public B route(String expression, Consumer<RouterSpec<ExpressionEvaluatingRouter>> routerConfigurer) {
 		return this.route(expression, routerConfigurer, null);
 	}
 
 	public B route(String expression,
-			ComponentConfigurer<RouterSpec<ExpressionEvaluatingRouter>> routerConfigurer,
-			EndpointConfigurer<GenericEndpointSpec<ExpressionEvaluatingRouter>> endpointConfigurer) {
+			Consumer<RouterSpec<ExpressionEvaluatingRouter>> routerConfigurer,
+			Consumer<GenericEndpointSpec<ExpressionEvaluatingRouter>> endpointConfigurer) {
 		return this.route(new ExpressionEvaluatingRouter(PARSER.parseExpression(expression)), routerConfigurer,
 				endpointConfigurer);
 	}
@@ -499,7 +514,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public <S, T> B route(GenericRouter<S, T> router,
-			ComponentConfigurer<RouterSpec<MethodInvokingRouter>> routerConfigurer) {
+			Consumer<RouterSpec<MethodInvokingRouter>> routerConfigurer) {
 		return this.route(null, router, routerConfigurer);
 	}
 
@@ -508,19 +523,19 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public <P, T> B route(Class<P> payloadType, GenericRouter<P, T> router,
-			ComponentConfigurer<RouterSpec<MethodInvokingRouter>> routerConfigurer) {
+			Consumer<RouterSpec<MethodInvokingRouter>> routerConfigurer) {
 		return this.route(payloadType, router, routerConfigurer, null);
 	}
 
 	public <S, T> B route(GenericRouter<S, T> router,
-			ComponentConfigurer<RouterSpec<MethodInvokingRouter>> routerConfigurer,
-			EndpointConfigurer<GenericEndpointSpec<MethodInvokingRouter>> endpointConfigurer) {
+			Consumer<RouterSpec<MethodInvokingRouter>> routerConfigurer,
+			Consumer<GenericEndpointSpec<MethodInvokingRouter>> endpointConfigurer) {
 		return route(null, router, routerConfigurer, endpointConfigurer);
 	}
 
 	public <P, T> B route(Class<P> payloadType, GenericRouter<P, T> router,
-			ComponentConfigurer<RouterSpec<MethodInvokingRouter>> routerConfigurer,
-			EndpointConfigurer<GenericEndpointSpec<MethodInvokingRouter>> endpointConfigurer) {
+			Consumer<RouterSpec<MethodInvokingRouter>> routerConfigurer,
+			Consumer<GenericEndpointSpec<MethodInvokingRouter>> endpointConfigurer) {
 		MethodInvokingRouter methodInvokingRouter = isLambda(router)
 				? new MethodInvokingRouter(new LambdaMessageProcessor(router, payloadType))
 				: new MethodInvokingRouter(router);
@@ -528,24 +543,24 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public <R extends AbstractMappingMessageRouter> B route(R router,
-			ComponentConfigurer<RouterSpec<R>> routerConfigurer,
-			EndpointConfigurer<GenericEndpointSpec<R>> endpointConfigurer) {
+			Consumer<RouterSpec<R>> routerConfigurer,
+			Consumer<GenericEndpointSpec<R>> endpointConfigurer) {
 		if (routerConfigurer != null) {
 			RouterSpec<R> routerSpec = new RouterSpec<R>(router);
-			routerConfigurer.configure(routerSpec);
+			routerConfigurer.accept(routerSpec);
 		}
 		return this.route(router, endpointConfigurer);
 	}
 
-	public B routeToRecipients(ComponentConfigurer<RecipientListRouterSpec> routerConfigurer) {
+	public B routeToRecipients(Consumer<RecipientListRouterSpec> routerConfigurer) {
 		return this.routeToRecipients(routerConfigurer, null);
 	}
 
-	public B routeToRecipients(ComponentConfigurer<RecipientListRouterSpec> routerConfigurer,
-			EndpointConfigurer<GenericEndpointSpec<RecipientListRouter>> endpointConfigurer) {
+	public B routeToRecipients(Consumer<RecipientListRouterSpec> routerConfigurer,
+			Consumer<GenericEndpointSpec<RecipientListRouter>> endpointConfigurer) {
 		Assert.notNull(routerConfigurer);
 		RecipientListRouterSpec spec = new RecipientListRouterSpec();
-		routerConfigurer.configure(spec);
+		routerConfigurer.accept(spec);
 		DslRecipientListRouter recipientListRouter = (DslRecipientListRouter) spec.get();
 		Assert.notEmpty(recipientListRouter.get(), "recipient list must not be empty");
 		return this.route(recipientListRouter, endpointConfigurer);
@@ -556,7 +571,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public <R extends AbstractMessageRouter> B route(R router,
-			EndpointConfigurer<GenericEndpointSpec<R>> endpointConfigurer) {
+			Consumer<GenericEndpointSpec<R>> endpointConfigurer) {
 		return this.handle(router, endpointConfigurer);
 	}
 
@@ -565,7 +580,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public B gateway(String requestChannel,
-			EndpointConfigurer<GatewayEndpointSpec> endpointConfigurer) {
+			Consumer<GatewayEndpointSpec> endpointConfigurer) {
 		return register(new GatewayEndpointSpec(requestChannel), endpointConfigurer);
 	}
 
@@ -574,14 +589,14 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
 	}
 
 	public B gateway(MessageChannel requestChannel,
-			EndpointConfigurer<GatewayEndpointSpec> endpointConfigurer) {
+			Consumer<GatewayEndpointSpec> endpointConfigurer) {
 		return register(new GatewayEndpointSpec(requestChannel), endpointConfigurer);
 	}
 
 	private <S extends ConsumerEndpointSpec<S, ?>> B register(S endpointSpec,
-			EndpointConfigurer<S> endpointConfigurer) {
+			Consumer<S> endpointConfigurer) {
 		if (endpointConfigurer != null) {
-			endpointConfigurer.configure(endpointSpec);
+			endpointConfigurer.accept(endpointSpec);
 		}
 		MessageChannel inputChannel = this.currentMessageChannel;
 		this.currentMessageChannel = null;

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

@@ -24,12 +24,14 @@ import org.springframework.integration.dsl.core.ComponentsRegistration;
 import org.springframework.integration.dsl.core.MessageProducerSpec;
 import org.springframework.integration.dsl.core.MessageSourceSpec;
 import org.springframework.integration.dsl.core.MessagingGatewaySpec;
-import org.springframework.integration.dsl.support.EndpointConfigurer;
+import org.springframework.integration.dsl.support.Consumer;
 import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype;
+import org.springframework.integration.dsl.support.Function;
 import org.springframework.integration.dsl.support.MessageChannelReference;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.gateway.MessagingGatewaySupport;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.util.Assert;
 
 /**
  * The central factory for fluent {@link IntegrationFlowBuilder} API.
@@ -39,9 +41,9 @@ import org.springframework.messaging.MessageChannel;
 public final class IntegrationFlows {
 
 	/**
-	 * @param messageChannelName the name of existing {@link org.springframework.messaging.MessageChannel} bean.
-	 *                           The new {@link org.springframework.integration.channel.DirectChannel} bean will be
-	 *                           created on context startup, if there is no bean with this name.
+	 * @param messageChannelName the name of existing {@link MessageChannel} bean.
+	 * The new {@link DirectChannel} bean will be created on context startup
+	 * if there is no bean with this name.
 	 * @return new {@link IntegrationFlowBuilder}
 	 */
 	public static IntegrationFlowBuilder from(String messageChannelName) {
@@ -49,15 +51,28 @@ public final class IntegrationFlows {
 	}
 
 	/**
-	 * @param messageChannelName the name for {@link org.springframework.integration.channel.FixedSubscriberChannel}
-	 *                           to be created on context startup, not reference.
+	 * @param messageChannelName the name for {@link DirectChannel} or
+	 * {@link org.springframework.integration.channel.FixedSubscriberChannel}
+	 * to be created on context startup, not reference.
+	 * The {@link MessageChannel} depends on the {@code fixedSubscriber} boolean argument.
+	 * @param fixedSubscriber the boolean flag to determine if result {@link MessageChannel} should
+	 * be {@link DirectChannel}, if {@code false} or
+	 * {@link org.springframework.integration.channel.FixedSubscriberChannel}, if {@code true}.
 	 * @return new {@link IntegrationFlowBuilder}
 	 */
-	public static IntegrationFlowBuilder fromFixedMessageChannel(String messageChannelName) {
-		return from(new FixedSubscriberChannelPrototype(messageChannelName));
+	public static IntegrationFlowBuilder from(String messageChannelName, boolean fixedSubscriber) {
+		return fixedSubscriber
+				? from(new FixedSubscriberChannelPrototype(messageChannelName))
+				: from(messageChannelName);
+	}
+
+	public static IntegrationFlowBuilder from(ChannelsFunction channels) {
+		Assert.notNull(channels);
+		return from(channels.apply(new Channels()));
 	}
 
 	public static IntegrationFlowBuilder from(MessageChannelSpec<?, ?> messageChannelSpec) {
+		Assert.notNull(messageChannelSpec);
 		return from(messageChannelSpec.get());
 	}
 
@@ -65,13 +80,23 @@ public final class IntegrationFlows {
 		return new IntegrationFlowBuilder().channel(messageChannel);
 	}
 
-	public static <S extends MessageSourceSpec<S, ? extends MessageSource<?>>> IntegrationFlowBuilder
-	from(S messageSourceSpec) {
+	public static IntegrationFlowBuilder from(MessageSourcesFunction sources) {
+		return from(sources, null);
+	}
+
+	public static IntegrationFlowBuilder from(MessageSourcesFunction sources,
+			Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
+		Assert.notNull(sources);
+		return from(sources.apply(new MessageSources()), endpointConfigurer);
+	}
+
+	public static IntegrationFlowBuilder from(MessageSourceSpec<?, ? extends MessageSource<?>> messageSourceSpec) {
 		return from(messageSourceSpec, null);
 	}
 
-	public static <S extends MessageSourceSpec<S, ? extends MessageSource<?>>> IntegrationFlowBuilder
-	from(S messageSourceSpec, EndpointConfigurer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
+	public static IntegrationFlowBuilder from(MessageSourceSpec<?, ? extends MessageSource<?>> messageSourceSpec,
+			Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
+		Assert.notNull(messageSourceSpec);
 		return from(messageSourceSpec.get(), endpointConfigurer, registerComponents(messageSourceSpec));
 	}
 
@@ -80,16 +105,16 @@ public final class IntegrationFlows {
 	}
 
 	public static IntegrationFlowBuilder from(MessageSource<?> messageSource,
-			EndpointConfigurer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
+			Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
 		return from(messageSource, endpointConfigurer, null);
 	}
 
 	private static IntegrationFlowBuilder from(MessageSource<?> messageSource,
-			EndpointConfigurer<SourcePollingChannelAdapterSpec> endpointConfigurer,
+			Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer,
 			IntegrationFlowBuilder integrationFlowBuilder) {
 		SourcePollingChannelAdapterSpec spec = new SourcePollingChannelAdapterSpec(messageSource);
 		if (endpointConfigurer != null) {
-			endpointConfigurer.configure(spec);
+			endpointConfigurer.accept(spec);
 		}
 		if (integrationFlowBuilder == null) {
 			integrationFlowBuilder = new IntegrationFlowBuilder();
@@ -98,6 +123,10 @@ public final class IntegrationFlows {
 				.currentComponent(spec);
 	}
 
+	public static IntegrationFlowBuilder from(MessageProducersFunction producers) {
+		return from(producers.apply(new MessageProducers()));
+	}
+
 	public static IntegrationFlowBuilder from(MessageProducerSpec<?, ?> messageProducerSpec) {
 		return from(messageProducerSpec.get(), registerComponents(messageProducerSpec));
 	}
@@ -123,6 +152,10 @@ public final class IntegrationFlows {
 		return integrationFlowBuilder.addComponent(messageProducer);
 	}
 
+	public static IntegrationFlowBuilder from(MessagingGatewaysFunction gateways) {
+		return from(gateways.apply(new MessagingGateways()));
+	}
+
 	public static IntegrationFlowBuilder from(MessagingGatewaySpec<?, ?> inboundGatewaySpec) {
 		return from(inboundGatewaySpec.get(), registerComponents(inboundGatewaySpec));
 	}
@@ -165,4 +198,12 @@ public final class IntegrationFlows {
 	private IntegrationFlows() {
 	}
 
+	public interface ChannelsFunction extends Function<Channels, MessageChannelSpec<?, ?>> {}
+
+	public interface MessageSourcesFunction extends Function<MessageSources, MessageSourceSpec<?, ?>> {}
+
+	public interface MessageProducersFunction extends Function<MessageProducers, MessageProducerSpec<?, ?>> {}
+
+	public interface MessagingGatewaysFunction extends Function<MessagingGateways, MessagingGatewaySpec<?, ?>> {}
+
 }

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

@@ -55,16 +55,20 @@ class LambdaMessageProcessor implements MessageProcessor<Object>, BeanFactoryAwa
 		this.target = target;
 		final AtomicReference<Method> methodValue = new AtomicReference<Method>();
 		ReflectionUtils.doWithMethods(target.getClass(), new ReflectionUtils.MethodCallback() {
+
 			@Override
 			public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
 				methodValue.set(method);
 			}
+
 		}, new ReflectionUtils.MethodFilter() {
+
 			@Override
 			public boolean matches(Method method) {
 				return !method.isBridge() && method.getDeclaringClass() != Object.class &&
 						Modifier.isPublic(method.getModifiers()) && !Modifier.isStatic(method.getModifiers());
 			}
+
 		});
 
 		Assert.notNull(methodValue.get(), "LambdaMessageProcessor is applicable for inline or lambda " +
@@ -125,4 +129,5 @@ class LambdaMessageProcessor implements MessageProcessor<Object>, BeanFactoryAwa
 			throw new MessageHandlingException(message, e);
 		}
 	}
+
 }

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

@@ -0,0 +1,75 @@
+/*
+ * 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.io.File;
+
+import org.springframework.amqp.core.Queue;
+import org.springframework.integration.dsl.amqp.Amqp;
+import org.springframework.integration.dsl.amqp.AmqpInboundChannelAdapterSpec;
+import org.springframework.integration.dsl.file.Files;
+import org.springframework.integration.dsl.file.TailAdapterSpec;
+import org.springframework.integration.dsl.jms.Jms;
+import org.springframework.integration.dsl.jms.JmsMessageDrivenChannelAdapterSpec;
+import org.springframework.integration.dsl.mail.ImapIdleChannelAdapterSpec;
+import org.springframework.integration.dsl.mail.Mail;
+import org.springframework.jms.listener.AbstractMessageListenerContainer;
+
+/**
+ * @author Artem Bilan
+ */
+public class MessageProducers {
+
+	public AmqpInboundChannelAdapterSpec amqp(
+			org.springframework.amqp.rabbit.connection.ConnectionFactory connectionFactory, String... queueNames) {
+		return Amqp.inboundAdapter(connectionFactory, queueNames);
+	}
+
+	public AmqpInboundChannelAdapterSpec amqp(
+			org.springframework.amqp.rabbit.connection.ConnectionFactory connectionFactory, Queue... queues) {
+		return Amqp.inboundAdapter(connectionFactory, queues);
+	}
+
+	public TailAdapterSpec tail(File file) {
+		return Files.tailAdapter(file);
+	}
+
+	public ImapIdleChannelAdapterSpec imap(String url) {
+		return Mail.imapIdleAdapter(url);
+	}
+
+	public JmsMessageDrivenChannelAdapterSpec<? extends JmsMessageDrivenChannelAdapterSpec<?>> jms(
+			AbstractMessageListenerContainer listenerContainer) {
+		return Jms.messageDriverChannelAdapter(listenerContainer);
+	}
+
+	public JmsMessageDrivenChannelAdapterSpec<? extends JmsMessageDrivenChannelAdapterSpec<?>> jms(
+			javax.jms.ConnectionFactory connectionFactory) {
+		return Jms.messageDriverChannelAdapter(connectionFactory);
+	}
+
+	public <C extends AbstractMessageListenerContainer>
+	JmsMessageDrivenChannelAdapterSpec<? extends JmsMessageDrivenChannelAdapterSpec<?>> jms(
+			javax.jms.ConnectionFactory connectionFactory,
+			Class<C> containerClass) {
+		return Jms.messageDriverChannelAdapter(connectionFactory, containerClass);
+	}
+
+	MessageProducers() {
+	}
+
+}

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

@@ -0,0 +1,108 @@
+/*
+ * 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.io.File;
+import java.util.Comparator;
+
+import javax.jms.ConnectionFactory;
+
+import org.apache.commons.net.ftp.FTPFile;
+
+import org.springframework.integration.dsl.file.FileInboundChannelAdapterSpec;
+import org.springframework.integration.dsl.file.Files;
+import org.springframework.integration.dsl.ftp.Ftp;
+import org.springframework.integration.dsl.ftp.FtpInboundChannelAdapterSpec;
+import org.springframework.integration.dsl.jms.Jms;
+import org.springframework.integration.dsl.jms.JmsInboundChannelAdapterSpec;
+import org.springframework.integration.dsl.mail.ImapMailInboundChannelAdapterSpec;
+import org.springframework.integration.dsl.mail.Mail;
+import org.springframework.integration.dsl.mail.Pop3MailInboundChannelAdapterSpec;
+import org.springframework.integration.dsl.sftp.Sftp;
+import org.springframework.integration.dsl.sftp.SftpInboundChannelAdapterSpec;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.jms.core.JmsTemplate;
+
+import com.jcraft.jsch.ChannelSftp;
+
+/**
+ * @author Artem Bilan
+ */
+public class MessageSources {
+
+	public FileInboundChannelAdapterSpec file(File directory) {
+		return file(directory, null);
+	}
+
+	public FileInboundChannelAdapterSpec file(File directory, Comparator<File> receptionOrderComparator) {
+		return Files.inboundAdapter(directory, receptionOrderComparator);
+	}
+
+	public FtpInboundChannelAdapterSpec ftp(SessionFactory<FTPFile> sessionFactory) {
+		return ftp(sessionFactory, null);
+	}
+
+	public FtpInboundChannelAdapterSpec ftp(SessionFactory<FTPFile> sessionFactory,
+			Comparator<File> receptionOrderComparator) {
+		return Ftp.inboundAdapter(sessionFactory, receptionOrderComparator);
+	}
+
+	public SftpInboundChannelAdapterSpec sftp(SessionFactory<ChannelSftp.LsEntry> sessionFactory) {
+		return sftp(sessionFactory, null);
+	}
+
+	public SftpInboundChannelAdapterSpec sftp(SessionFactory<ChannelSftp.LsEntry> sessionFactory,
+			Comparator<File> receptionOrderComparator) {
+		return Sftp.inboundAdapter(sessionFactory, receptionOrderComparator);
+	}
+
+	public JmsInboundChannelAdapterSpec<? extends JmsInboundChannelAdapterSpec<?>> jms(JmsTemplate jmsTemplate) {
+		return Jms.inboundAdapter(jmsTemplate);
+	}
+
+	public JmsInboundChannelAdapterSpec.JmsInboundChannelSpecTemplateAware jms(ConnectionFactory connectionFactory) {
+		return Jms.inboundAdapter(connectionFactory);
+	}
+
+	public ImapMailInboundChannelAdapterSpec imap() {
+		return Mail.imapInboundAdapter();
+	}
+
+	public ImapMailInboundChannelAdapterSpec imap(String url) {
+		return Mail.imapInboundAdapter(url);
+	}
+
+	public Pop3MailInboundChannelAdapterSpec pop3() {
+		return Mail.pop3InboundAdapter();
+	}
+
+	public Pop3MailInboundChannelAdapterSpec pop3(String url) {
+		return Mail.pop3InboundAdapter(url);
+	}
+
+	public Pop3MailInboundChannelAdapterSpec pop3(String host, String username, String password) {
+		return pop3(host, -1, username, password);
+	}
+
+	public Pop3MailInboundChannelAdapterSpec pop3(String host, int port, String username, String password) {
+		return Mail.pop3InboundAdapter(host, port, username, password);
+	}
+
+	MessageSources() {
+	}
+
+}

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

@@ -0,0 +1,68 @@
+/*
+ * 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 javax.jms.ConnectionFactory;
+
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.integration.dsl.amqp.Amqp;
+import org.springframework.integration.dsl.amqp.AmqpInboundGatewaySpec;
+import org.springframework.integration.dsl.jms.Jms;
+import org.springframework.integration.dsl.jms.JmsInboundGatewaySpec;
+import org.springframework.jms.listener.AbstractMessageListenerContainer;
+import org.springframework.jms.listener.DefaultMessageListenerContainer;
+
+/**
+ * @author Artem Bilan
+ */
+public class MessagingGateways {
+
+	public AmqpInboundGatewaySpec amqp(org.springframework.amqp.rabbit.connection.ConnectionFactory connectionFactory,
+			String... queueNames) {
+		return Amqp.inboundGateway(connectionFactory, queueNames);
+	}
+
+	public AmqpInboundGatewaySpec amqp(org.springframework.amqp.rabbit.connection.ConnectionFactory connectionFactory,
+			Queue... queues) {
+		return Amqp.inboundGateway(connectionFactory, queues);
+	}
+
+	public AmqpInboundGatewaySpec amqp(SimpleMessageListenerContainer listenerContainer) {
+		return (AmqpInboundGatewaySpec) Amqp.inboundGateway(listenerContainer);
+	}
+
+	public JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec<DefaultMessageListenerContainer> jms(
+			javax.jms.ConnectionFactory connectionFactory) {
+		return Jms.inboundGateway(connectionFactory);
+	}
+
+	public <C extends AbstractMessageListenerContainer>
+	JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec<C> jms(ConnectionFactory connectionFactory,
+			Class<C> containerClass) {
+		return Jms.inboundGateway(connectionFactory, containerClass);
+	}
+
+	public JmsInboundGatewaySpec<? extends JmsInboundGatewaySpec<?>> jms(
+			AbstractMessageListenerContainer listenerContainer) {
+		return Jms.inboundGateway(listenerContainer);
+	}
+
+	MessagingGateways() {
+	}
+
+}

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

@@ -25,8 +25,8 @@ import org.springframework.integration.scheduling.PollerMetadata;
  * @author Artem Bilan
 
  */
-public final class SourcePollingChannelAdapterSpec extends EndpointSpec<SourcePollingChannelAdapterSpec,
-		SourcePollingChannelAdapterFactoryBean, MessageSource<?>> {
+public final class SourcePollingChannelAdapterSpec extends
+		EndpointSpec<SourcePollingChannelAdapterSpec, SourcePollingChannelAdapterFactoryBean, MessageSource<?>> {
 
 	SourcePollingChannelAdapterSpec(MessageSource<?> messageSource) {
 		super(messageSource);

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java

@@ -47,7 +47,8 @@ public abstract class Amqp {
 		return new AmqpInboundGatewaySpec(listenerContainer);
 	}
 
-	public static AmqpInboundChannelAdapterSpec inboundAdapter(ConnectionFactory connectionFactory, String... queueNames) {
+	public static AmqpInboundChannelAdapterSpec inboundAdapter(ConnectionFactory connectionFactory,
+			String... queueNames) {
 		SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
 		listenerContainer.setQueueNames(queueNames);
 		return (AmqpInboundChannelAdapterSpec) inboundAdapter(listenerContainer);
@@ -77,12 +78,13 @@ public abstract class Amqp {
 		return pollableChannel(null, connectionFactory);
 	}
 
-	public static <S extends AmqpPollableMessageChannelSpec<S>> AmqpPollableMessageChannelSpec<S> pollableChannel(String id,
-			ConnectionFactory connectionFactory) {
+	public static <S extends AmqpPollableMessageChannelSpec<S>> AmqpPollableMessageChannelSpec<S> pollableChannel(
+			String id, ConnectionFactory connectionFactory) {
 		return new AmqpPollableMessageChannelSpec<S>(connectionFactory).id(id);
 	}
 
-	public static <S extends AmqpMessageChannelSpec<S>> AmqpMessageChannelSpec<S> channel(ConnectionFactory connectionFactory) {
+	public static <S extends AmqpMessageChannelSpec<S>> AmqpMessageChannelSpec<S> channel(
+			ConnectionFactory connectionFactory) {
 		return channel(null, connectionFactory);
 	}
 

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpInboundChannelAdapterSpec.java

@@ -34,7 +34,8 @@ import org.springframework.util.ErrorHandler;
 /**
  * @author Artem Bilan
  */
-public class AmqpInboundChannelAdapterSpec extends MessageProducerSpec<AmqpInboundChannelAdapterSpec, AmqpInboundChannelAdapter> {
+public class AmqpInboundChannelAdapterSpec
+		extends MessageProducerSpec<AmqpInboundChannelAdapterSpec, AmqpInboundChannelAdapter> {
 
 	private final SimpleMessageListenerContainer listenerContainer;
 

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

@@ -107,7 +107,8 @@ public class QueueChannelSpec extends MessageChannelSpec<QueueChannelSpec, Queue
 				this.queue = new MessageGroupQueue(this.messageGroupStore, this.groupId);
 			}
 
-			((MessageGroupQueue) this.queue).setPriority(this.messageGroupStore instanceof PriorityCapableChannelMessageStore);
+			((MessageGroupQueue) this.queue).setPriority(
+					this.messageGroupStore instanceof PriorityCapableChannelMessageStore);
 
 			return super.doGet();
 		}

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

@@ -18,10 +18,11 @@ package org.springframework.integration.dsl.core;
 
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.core.ResolvableType;
-import org.springframework.integration.dsl.support.PollerSpec;
-import org.springframework.integration.dsl.tuple.Tuple;
-import org.springframework.integration.dsl.tuple.Tuple2;
+import org.springframework.integration.dsl.support.Function;
+import org.springframework.integration.dsl.support.tuple.Tuple;
+import org.springframework.integration.dsl.support.tuple.Tuple2;
 import org.springframework.integration.scheduling.PollerMetadata;
+import org.springframework.util.Assert;
 
 /**
  * @author Artem Bilan
@@ -32,6 +33,7 @@ public abstract class EndpointSpec<S extends EndpointSpec<S, F, H>, F extends Be
 
 	@SuppressWarnings("unchecked")
 	protected EndpointSpec(H handler) {
+		Assert.notNull(handler);
 		try {
 			Class<?> fClass = ResolvableType.forClass(this.getClass()).as(EndpointSpec.class).resolveGenerics()[1];
 			F endpointFactoryBean = (F) fClass.newInstance();
@@ -53,6 +55,10 @@ public abstract class EndpointSpec<S extends EndpointSpec<S, F, H>, F extends Be
 
 	public abstract S poller(PollerMetadata pollerMetadata);
 
+	public S poller(Function<PollerFactory, PollerSpec> pollers) {
+		return poller(pollers.apply(new PollerFactory()));
+	}
+
 	public S poller(PollerSpec pollerMetadataSpec) {
 		return this.poller(pollerMetadataSpec.get());
 	}

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

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2014 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.dsl.core;
+
+import java.util.TimeZone;
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.scheduling.Trigger;
+
+/**
+ * @author Artem Bilan
+ */
+public class PollerFactory {
+
+	public PollerSpec trigger(Trigger trigger) {
+		return Pollers.trigger(trigger);
+	}
+
+	public PollerSpec cron(String cronExpression) {
+		return Pollers.cron(cronExpression);
+	}
+
+	public PollerSpec cron(String cronExpression, TimeZone timeZone) {
+		return Pollers.cron(cronExpression, timeZone);
+	}
+
+	public PollerSpec fixedRate(long period) {
+		return Pollers.fixedRate(period);
+	}
+
+	public PollerSpec fixedRate(long period, TimeUnit timeUnit) {
+		return Pollers.fixedRate(period, timeUnit);
+	}
+
+	public PollerSpec fixedRate(long period, long initialDelay) {
+		return Pollers.fixedRate(period, initialDelay);
+	}
+
+	public PollerSpec fixedDelay(long period, TimeUnit timeUnit, long initialDelay) {
+		return Pollers.fixedDelay(period, timeUnit, initialDelay);
+	}
+
+	public PollerSpec fixedRate(long period, TimeUnit timeUnit, long initialDelay) {
+		return Pollers.fixedRate(period, timeUnit, initialDelay);
+	}
+
+	public PollerSpec fixedDelay(long period, TimeUnit timeUnit) {
+		return Pollers.fixedDelay(period, timeUnit);
+	}
+
+	public PollerSpec fixedDelay(long period, long initialDelay) {
+		return Pollers.fixedDelay(period, initialDelay);
+	}
+
+	public PollerSpec fixedDelay(long period) {
+		return Pollers.fixedDelay(period);
+	}
+
+	PollerFactory() {
+	}
+
+}

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

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.dsl.support;
+package org.springframework.integration.dsl.core;
 
 import java.util.Arrays;
 import java.util.LinkedList;
@@ -23,7 +23,6 @@ import java.util.concurrent.Executor;
 
 import org.aopalliance.aop.Advice;
 
-import org.springframework.integration.dsl.core.IntegrationComponentSpec;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.integration.transaction.TransactionSynchronizationFactory;
 import org.springframework.scheduling.Trigger;
@@ -46,7 +45,8 @@ public final class PollerSpec extends IntegrationComponentSpec<PollerSpec, Polle
 		this.pollerMetadata.setTrigger(trigger);
 	}
 
-	public PollerSpec transactionSynchronizationFactory(TransactionSynchronizationFactory transactionSynchronizationFactory) {
+	public PollerSpec transactionSynchronizationFactory(
+			TransactionSynchronizationFactory transactionSynchronizationFactory) {
 		pollerMetadata.setTransactionSynchronizationFactory(transactionSynchronizationFactory);
 		return this;
 	}

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

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.dsl.support;
+package org.springframework.integration.dsl.core;
 
 import java.util.TimeZone;
 import java.util.concurrent.TimeUnit;

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileTransferringMessageHandlerSpec.java

@@ -53,7 +53,8 @@ public abstract class FileTransferringMessageHandlerSpec<F, S extends FileTransf
 	}
 
 	@SuppressWarnings("unchecked")
-	protected FileTransferringMessageHandlerSpec(RemoteFileTemplate<F> remoteFileTemplate, FileExistsMode fileExistsMode) {
+	protected FileTransferringMessageHandlerSpec(RemoteFileTemplate<F> remoteFileTemplate,
+			FileExistsMode fileExistsMode) {
 		Constructor<?> fileExistsModeConstructor =
 				ClassUtils.getConstructorIfAvailable(FileTransferringMessageHandler.class, RemoteFileTemplate.class,
 						FileExistsMode.class);
@@ -64,7 +65,8 @@ public abstract class FileTransferringMessageHandlerSpec<F, S extends FileTransf
 		}
 		else {
 			try {
-				this.target = (FileTransferringMessageHandler<F>) fileExistsModeConstructor.newInstance(remoteFileTemplate,
+				this.target =
+						(FileTransferringMessageHandler<F>) fileExistsModeConstructor.newInstance(remoteFileTemplate,
 								fileExistsMode);
 			}
 			catch (Exception e) {

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteFileInboundChannelAdapterSpec.java

@@ -30,7 +30,7 @@ import org.springframework.util.Assert;
 /**
  * @author Artem Bilan
  */
-public abstract class RemoteInboundChannelAdapterSpec<F, S extends RemoteInboundChannelAdapterSpec<F, S, MS>,
+public abstract class RemoteFileInboundChannelAdapterSpec<F, S extends RemoteFileInboundChannelAdapterSpec<F, S, MS>,
 		MS extends AbstractInboundFileSynchronizingMessageSource<F>>
 		extends MessageSourceSpec<S, MS> implements ComponentsRegistration {
 
@@ -38,7 +38,7 @@ public abstract class RemoteInboundChannelAdapterSpec<F, S extends RemoteInbound
 
 	private FileListFilter<F> filter;
 
-	protected RemoteInboundChannelAdapterSpec(AbstractInboundFileSynchronizer<F> synchronizer) {
+	protected RemoteFileInboundChannelAdapterSpec(AbstractInboundFileSynchronizer<F> synchronizer) {
 		this.synchronizer = synchronizer;
 	}
 

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/Ftp.java

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.dsl.ftp;
 
+import java.io.File;
 import java.util.Comparator;
 
 import org.apache.commons.net.ftp.FTPFile;
@@ -36,7 +37,7 @@ public abstract class Ftp {
 	}
 
 	public static FtpInboundChannelAdapterSpec inboundAdapter(SessionFactory<FTPFile> sessionFactory,
-			Comparator<java.io.File> receptionOrderComparator) {
+			Comparator<File> receptionOrderComparator) {
 		return new FtpInboundChannelAdapterSpec(sessionFactory, receptionOrderComparator);
 	}
 

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/FtpInboundChannelAdapterSpec.java

@@ -21,7 +21,7 @@ import java.util.Comparator;
 
 import org.apache.commons.net.ftp.FTPFile;
 
-import org.springframework.integration.dsl.file.RemoteInboundChannelAdapterSpec;
+import org.springframework.integration.dsl.file.RemoteFileInboundChannelAdapterSpec;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter;
 import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
@@ -32,8 +32,8 @@ import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizingMe
  * @author Artem Bilan
  */
 public class FtpInboundChannelAdapterSpec
-		extends RemoteInboundChannelAdapterSpec<FTPFile, FtpInboundChannelAdapterSpec,
-		FtpInboundFileSynchronizingMessageSource> {
+		extends RemoteFileInboundChannelAdapterSpec<FTPFile, FtpInboundChannelAdapterSpec,
+				FtpInboundFileSynchronizingMessageSource> {
 
 	FtpInboundChannelAdapterSpec(SessionFactory<FTPFile> sessionFactory, Comparator<File> comparator) {
 		super(new FtpInboundFileSynchronizer(sessionFactory));

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/Jms.java

@@ -111,13 +111,13 @@ public abstract class Jms {
 
 	public static
 	JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec<DefaultMessageListenerContainer>
-	messageDriverAdapter(ConnectionFactory connectionFactory) {
-		return messageDriverAdapter(connectionFactory, DefaultMessageListenerContainer.class);
+	messageDriverChannelAdapter(ConnectionFactory connectionFactory) {
+		return messageDriverChannelAdapter(connectionFactory, DefaultMessageListenerContainer.class);
 	}
 
 	public static <C extends AbstractMessageListenerContainer>
 	JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec<C>
-	messageDriverAdapter(ConnectionFactory connectionFactory, Class<C> containerClass) {
+	messageDriverChannelAdapter(ConnectionFactory connectionFactory, Class<C> containerClass) {
 		try {
 			JmsListenerContainerSpec<C> spec = new JmsListenerContainerSpec<C>(containerClass)
 					.connectionFactory(connectionFactory);

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsDestinationAccessorSpec.java

@@ -25,7 +25,8 @@ import org.springframework.jms.support.destination.JmsDestinationAccessor;
 /**
  * @author Artem Bilan
  */
-public abstract class JmsDestinationAccessorSpec<S extends JmsDestinationAccessorSpec<S, A>, A extends JmsDestinationAccessor>
+public abstract class
+		JmsDestinationAccessorSpec<S extends JmsDestinationAccessorSpec<S, A>, A extends JmsDestinationAccessor>
 		extends IntegrationComponentSpec<S, A> {
 
 	protected JmsDestinationAccessorSpec(A accessor) {

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundChannelAdapterSpec.java

@@ -20,7 +20,7 @@ import javax.jms.ConnectionFactory;
 import javax.jms.Destination;
 
 import org.springframework.integration.dsl.core.MessageSourceSpec;
-import org.springframework.integration.dsl.support.ComponentConfigurer;
+import org.springframework.integration.dsl.support.Consumer;
 import org.springframework.integration.jms.JmsDestinationPollingSource;
 import org.springframework.integration.jms.JmsHeaderMapper;
 import org.springframework.jms.core.JmsTemplate;
@@ -74,9 +74,9 @@ public class JmsInboundChannelAdapterSpec<S extends JmsInboundChannelAdapterSpec
 			super(connectionFactory);
 		}
 
-		public JmsInboundChannelSpecTemplateAware configureJmsTemplate(ComponentConfigurer<JmsTemplateSpec> configurer) {
+		public JmsInboundChannelSpecTemplateAware configureJmsTemplate(Consumer<JmsTemplateSpec> configurer) {
 			Assert.notNull(configurer);
-			configurer.configure(this.jmsTemplateSpec);
+			configurer.accept(this.jmsTemplateSpec);
 			return _this();
 		}
 

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundGatewaySpec.java

@@ -19,7 +19,7 @@ package org.springframework.integration.dsl.jms;
 import javax.jms.Destination;
 
 import org.springframework.integration.dsl.core.MessagingGatewaySpec;
-import org.springframework.integration.dsl.support.ComponentConfigurer;
+import org.springframework.integration.dsl.support.Consumer;
 import org.springframework.integration.jms.ChannelPublishingJmsMessageListener;
 import org.springframework.integration.jms.JmsHeaderMapper;
 import org.springframework.jms.listener.AbstractMessageListenerContainer;
@@ -124,10 +124,10 @@ public class JmsInboundGatewaySpec<S extends JmsInboundGatewaySpec<S>>
 			return _this();
 		}
 
-		public JmsInboundGatewayListenerContainerSpec<C> configureListenerContainer
-				(ComponentConfigurer<JmsListenerContainerSpec<C>> configurer) {
+		public JmsInboundGatewayListenerContainerSpec<C> configureListenerContainer(
+				Consumer<JmsListenerContainerSpec<C>> configurer) {
 			Assert.notNull(configurer);
-			configurer.configure(this.spec);
+			configurer.accept(this.spec);
 			return _this();
 		}
 

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsMessageDrivenChannelAdapterSpec.java

@@ -19,7 +19,7 @@ package org.springframework.integration.dsl.jms;
 import javax.jms.Destination;
 
 import org.springframework.integration.dsl.core.MessageProducerSpec;
-import org.springframework.integration.dsl.support.ComponentConfigurer;
+import org.springframework.integration.dsl.support.Consumer;
 import org.springframework.integration.jms.ChannelPublishingJmsMessageListener;
 import org.springframework.integration.jms.JmsHeaderMapper;
 import org.springframework.jms.listener.AbstractMessageListenerContainer;
@@ -53,7 +53,8 @@ public class JmsMessageDrivenChannelAdapterSpec<S extends JmsMessageDrivenChanne
 	}
 
 
-	public static class JmsMessageDrivenChannelAdapterListenerContainerSpec<C extends AbstractMessageListenerContainer> extends
+	public static class
+			JmsMessageDrivenChannelAdapterListenerContainerSpec<C extends AbstractMessageListenerContainer> extends
 			JmsMessageDrivenChannelAdapterSpec<JmsMessageDrivenChannelAdapterListenerContainerSpec<C>> {
 
 		private final JmsListenerContainerSpec<C> spec;
@@ -74,10 +75,10 @@ public class JmsMessageDrivenChannelAdapterSpec<S extends JmsMessageDrivenChanne
 			return _this();
 		}
 
-		public JmsMessageDrivenChannelAdapterListenerContainerSpec<C> configureListenerContainer
-				(ComponentConfigurer<JmsListenerContainerSpec<C>> configurer) {
+		public JmsMessageDrivenChannelAdapterListenerContainerSpec<C> configureListenerContainer(
+				Consumer<JmsListenerContainerSpec<C>> configurer) {
 			Assert.notNull(configurer);
-			configurer.configure(this.spec);
+			configurer.accept(this.spec);
 			return _this();
 		}
 

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsOutboundChannelAdapterSpec.java

@@ -20,7 +20,7 @@ import javax.jms.ConnectionFactory;
 import javax.jms.Destination;
 
 import org.springframework.integration.dsl.core.MessageHandlerSpec;
-import org.springframework.integration.dsl.support.ComponentConfigurer;
+import org.springframework.integration.dsl.support.Consumer;
 import org.springframework.integration.jms.JmsHeaderMapper;
 import org.springframework.integration.jms.JmsSendingMessageHandler;
 import org.springframework.jms.core.JmsTemplate;
@@ -79,9 +79,9 @@ public class JmsOutboundChannelAdapterSpec<S extends JmsOutboundChannelAdapterSp
 			super(connectionFactory);
 		}
 
-		public JmsOutboundChannelSpecTemplateAware configureJmsTemplate(ComponentConfigurer<JmsTemplateSpec> configurer) {
+		public JmsOutboundChannelSpecTemplateAware configureJmsTemplate(Consumer<JmsTemplateSpec> configurer) {
 			Assert.notNull(configurer);
-			configurer.configure(this.jmsTemplateSpec);
+			configurer.accept(this.jmsTemplateSpec);
 			return _this();
 		}
 

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsOutboundGatewaySpec.java

@@ -23,7 +23,7 @@ import javax.jms.Destination;
 
 import org.springframework.integration.dsl.core.IntegrationComponentSpec;
 import org.springframework.integration.dsl.core.MessageHandlerSpec;
-import org.springframework.integration.dsl.support.ComponentConfigurer;
+import org.springframework.integration.dsl.support.Consumer;
 import org.springframework.integration.jms.JmsHeaderMapper;
 import org.springframework.integration.jms.JmsOutboundGateway;
 import org.springframework.jms.support.converter.MessageConverter;
@@ -140,10 +140,10 @@ public class JmsOutboundGatewaySpec extends MessageHandlerSpec<JmsOutboundGatewa
 		return _this();
 	}
 
-	public JmsOutboundGatewaySpec replyContainer(ComponentConfigurer<ReplyContainerSpec> configurer) {
+	public JmsOutboundGatewaySpec replyContainer(Consumer<ReplyContainerSpec> configurer) {
 		Assert.notNull(configurer);
 		ReplyContainerSpec spec = new ReplyContainerSpec();
-		configurer.configure(spec);
+		configurer.accept(spec);
 		this.target.setReplyContainerProperties(spec.get());
 		return _this();
 	}

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/Mail.java

@@ -35,6 +35,10 @@ public class Mail {
 		return new Pop3MailInboundChannelAdapterSpec(url);
 	}
 
+	public static Pop3MailInboundChannelAdapterSpec pop3InboundAdapter(String host, String username, String password) {
+		return pop3InboundAdapter(host, -1, username, password);
+	}
+
 	public static Pop3MailInboundChannelAdapterSpec pop3InboundAdapter(String host, int port, String username,
 			String password) {
 		return new Pop3MailInboundChannelAdapterSpec(host, port, username, password);

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/Sftp.java

@@ -16,16 +16,17 @@
 
 package org.springframework.integration.dsl.sftp;
 
+import java.io.File;
 import java.util.Comparator;
 
-import com.jcraft.jsch.ChannelSftp;
-
 import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.file.support.FileExistsMode;
 import org.springframework.integration.sftp.gateway.SftpOutboundGateway;
 
+import com.jcraft.jsch.ChannelSftp;
+
 /**
  * @author Artem Bilan
  */
@@ -36,7 +37,7 @@ public abstract class Sftp {
 	}
 
 	public static SftpInboundChannelAdapterSpec inboundAdapter(SessionFactory<ChannelSftp.LsEntry> sessionFactory,
-			Comparator<java.io.File> receptionOrderComparator) {
+			Comparator<File> receptionOrderComparator) {
 		return new SftpInboundChannelAdapterSpec(sessionFactory, receptionOrderComparator);
 	}
 

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpInboundChannelAdapterSpec.java

@@ -21,7 +21,7 @@ import java.util.Comparator;
 
 import com.jcraft.jsch.ChannelSftp;
 
-import org.springframework.integration.dsl.file.RemoteInboundChannelAdapterSpec;
+import org.springframework.integration.dsl.file.RemoteFileInboundChannelAdapterSpec;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter;
 import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter;
@@ -32,8 +32,8 @@ import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizing
  * @author Artem Bilan
  */
 public class SftpInboundChannelAdapterSpec
-		extends RemoteInboundChannelAdapterSpec<ChannelSftp.LsEntry, SftpInboundChannelAdapterSpec,
-		SftpInboundFileSynchronizingMessageSource> {
+		extends RemoteFileInboundChannelAdapterSpec<ChannelSftp.LsEntry, SftpInboundChannelAdapterSpec,
+				SftpInboundFileSynchronizingMessageSource> {
 
 	SftpInboundChannelAdapterSpec(SessionFactory<ChannelSftp.LsEntry> sessionFactory, Comparator<File> comparator) {
 		super(new SftpInboundFileSynchronizer(sessionFactory));

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpMessageHandlerSpec.java

@@ -16,17 +16,18 @@
 
 package org.springframework.integration.dsl.sftp;
 
-import com.jcraft.jsch.ChannelSftp;
-
 import org.springframework.integration.dsl.file.FileTransferringMessageHandlerSpec;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.file.support.FileExistsMode;
 
+import com.jcraft.jsch.ChannelSftp;
+
 /**
  * @author Artem Bilan
  */
-public class SftpMessageHandlerSpec extends FileTransferringMessageHandlerSpec<ChannelSftp.LsEntry, SftpMessageHandlerSpec> {
+public class SftpMessageHandlerSpec
+		extends FileTransferringMessageHandlerSpec<ChannelSftp.LsEntry, SftpMessageHandlerSpec> {
 
 	SftpMessageHandlerSpec(SessionFactory<ChannelSftp.LsEntry> sessionFactory) {
 		super(sessionFactory);

spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpOutboundGatewaySpec.java

@@ -16,17 +16,18 @@
 
 package org.springframework.integration.dsl.sftp;
 
-import com.jcraft.jsch.ChannelSftp;
-
 import org.springframework.integration.dsl.file.RemoteFileOutboundGatewaySpec;
 import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
 import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter;
 import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter;
 
+import com.jcraft.jsch.ChannelSftp;
+
 /**
  * @author Artem Bilan
  */
-public class SftpOutboundGatewaySpec extends RemoteFileOutboundGatewaySpec<ChannelSftp.LsEntry, SftpOutboundGatewaySpec> {
+public class SftpOutboundGatewaySpec
+		extends RemoteFileOutboundGatewaySpec<ChannelSftp.LsEntry, SftpOutboundGatewaySpec> {
 
 
 	SftpOutboundGatewaySpec(AbstractRemoteFileOutboundGateway<ChannelSftp.LsEntry> outboundGateway) {

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

@@ -16,14 +16,20 @@
 
 package org.springframework.integration.dsl.support;
 
-import org.springframework.integration.dsl.core.IntegrationComponentSpec;
-
 /**
- * @author Artem Bilan
-
+ * Implementations accept a given value and perform work on the argument.
+ *
+ * @param <T> the type of values to accept
+ *
+ * @author Jon Brisbin
+ * @author Stephane Maldini
  */
-public interface ComponentConfigurer<S extends IntegrationComponentSpec<?, ?>> {
+public interface Consumer<T> {
 
-	void configure(S spec);
+	/**
+	 * Execute the logic of the action, accepting the given parameter.
+	 * @param t The parameter to pass to the consumer.
+	 */
+	void accept(T t);
 
 }

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

@@ -5,7 +5,7 @@
  * 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
+ *      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,
@@ -16,14 +16,24 @@
 
 package org.springframework.integration.dsl.support;
 
-import org.springframework.integration.dsl.core.EndpointSpec;
-
 /**
- * @author Artem Bilan
-
+ * Implementations of this class perform work on the given parameter
+ * and return a result of an optionally different type.
+ *
+ * @param <T> The type of the input to the apply operation
+ * @param <R> The type of the result of the apply operation
+ *
+ * @author Jon Brisbin
+ * @author Stephane Maldini
  */
-public interface EndpointConfigurer<S extends EndpointSpec<?, ?, ?>> {
+public interface Function<T, R> {
 
-	void configure(S spec);
+	/**
+	 * Execute the logic of the action, accepting the given parameter.
+	 * @param t The parameter to pass to the action.
+	 * @return result
+	 */
+	R apply(T t);
 
 }
+

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

@@ -27,7 +27,7 @@ import org.springframework.core.serializer.Deserializer;
 import org.springframework.core.serializer.Serializer;
 import org.springframework.expression.Expression;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
-import org.springframework.integration.dsl.tuple.Tuple2;
+import org.springframework.integration.dsl.support.tuple.Tuple2;
 import org.springframework.integration.file.transformer.FileToByteArrayTransformer;
 import org.springframework.integration.file.transformer.FileToStringTransformer;
 import org.springframework.integration.json.JsonToObjectTransformer;

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

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.dsl.tuple;
+package org.springframework.integration.dsl.support.tuple;
 
 import java.io.Serializable;
 import java.util.Arrays;
@@ -24,7 +24,8 @@ import java.util.Iterator;
 import org.springframework.util.Assert;
 
 /**
- * A {@literal Tuple} is an immutable {@link java.util.Collection} of objects, each of which can be of an arbitrary type.
+ * A {@literal Tuple} is an immutable {@link java.util.Collection} of objects,
+ * each of which can be of an arbitrary type.
  *
  * @author Jon Brisbin
  * @author Stephane Maldini

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

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.dsl.tuple;
+package org.springframework.integration.dsl.support.tuple;
 
 /**
  * A tuple that holds a single value

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

@@ -14,13 +14,13 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.dsl.tuple;
+package org.springframework.integration.dsl.support.tuple;
 
 /**
  * A tuple that holds two values
  *
  * @param <T1> The type of the first value held by this tuple
- * @param <T2> The type of the second balue held by this tuple
+ * @param <T2> The type of the second value held by this tuple
  *
  * @author Jon Brisbin
  */

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

@@ -1,4 +1,4 @@
 /**
  * Tuples provide a type-safe way to specify multiple parameters.
  */
-package org.springframework.integration.dsl.tuple;
+package org.springframework.integration.dsl.support.tuple;

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

@@ -101,8 +101,12 @@ import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.config.GlobalChannelInterceptor;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.dsl.Channels;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.MessageProducers;
+import org.springframework.integration.dsl.MessageSources;
+import org.springframework.integration.dsl.MessagingGateways;
 import org.springframework.integration.dsl.amqp.Amqp;
 import org.springframework.integration.dsl.channel.DirectChannelSpec;
 import org.springframework.integration.dsl.channel.MessageChannels;
@@ -110,7 +114,7 @@ import org.springframework.integration.dsl.file.Files;
 import org.springframework.integration.dsl.ftp.Ftp;
 import org.springframework.integration.dsl.jms.Jms;
 import org.springframework.integration.dsl.sftp.Sftp;
-import org.springframework.integration.dsl.support.Pollers;
+import org.springframework.integration.dsl.core.Pollers;
 import org.springframework.integration.dsl.support.Transformers;
 import org.springframework.integration.dsl.test.TestFtpServer;
 import org.springframework.integration.dsl.test.TestSftpServer;
@@ -119,6 +123,7 @@ import org.springframework.integration.event.core.MessagingEvent;
 import org.springframework.integration.event.outbound.ApplicationEventPublishingMessageHandler;
 import org.springframework.integration.file.DefaultFileNameGenerator;
 import org.springframework.integration.file.FileHeaders;
+import org.springframework.integration.file.FileWritingMessageHandler;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
 import org.springframework.integration.file.tail.ApacheCommonsFileTailingMessageProducer;
@@ -163,7 +168,6 @@ import org.springframework.util.StreamUtils;
 
 import com.jcraft.jsch.ChannelSftp;
 import com.mongodb.MongoClient;
-
 import de.flapdoodle.embed.mongo.MongodExecutable;
 import de.flapdoodle.embed.mongo.MongodStarter;
 import de.flapdoodle.embed.mongo.config.MongodConfigBuilder;
@@ -442,8 +446,7 @@ public class IntegrationFlowTests {
 		assertEquals("test", reply.getPayload());
 
 		assertTrue(this.beanFactory.containsBean("bridgeFlow2.channel#0"));
-		assertThat(this.beanFactory.getBean("bridgeFlow2.channel#0"), instanceOf(FixedSubscriberChannel
-				.class));
+		assertThat(this.beanFactory.getBean("bridgeFlow2.channel#0"), instanceOf(FixedSubscriberChannel.class));
 
 		try {
 			this.bridgeFlow2Input.send(message);
@@ -1357,7 +1360,7 @@ public class IntegrationFlowTests {
 
 		@Bean
 		public IntegrationFlow jmsOutboundFlow() {
-			return f -> f.handle(Jms.outboundAdapter(this.jmsConnectionFactory)
+			return f -> f.handleWithAdapter(h -> h.jms(this.jmsConnectionFactory)
 					.destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER));
 		}
 
@@ -1369,8 +1372,7 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow jmsInboundFlow() {
 			return IntegrationFlows
-					.from(Jms.inboundAdapter(this.jmsConnectionFactory)
-							.destination("jmsInbound"))
+					.from((MessageSources s) -> s.jms(this.jmsConnectionFactory).destination("jmsInbound"))
 					.<String, String>transform(String::toUpperCase)
 					.channel(this.jmsOutboundInboundReplyChannel())
 					.get();
@@ -1379,7 +1381,7 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow jmsMessageDriverFlow() {
 			return IntegrationFlows
-					.from(Jms.messageDriverAdapter(this.jmsConnectionFactory)
+					.from(Jms.messageDriverChannelAdapter(this.jmsConnectionFactory)
 							.destination("jmsMessageDriver"))
 					.<String, String>transform(String::toLowerCase)
 					.channel(this.jmsOutboundInboundReplyChannel())
@@ -1388,14 +1390,14 @@ public class IntegrationFlowTests {
 
 		@Bean
 		public IntegrationFlow jmsOutboundGatewayFlow() {
-			return f -> f.handle(Jms.outboundGateway(this.jmsConnectionFactory)
+			return f -> f.handleWithAdapter(a -> a.jmsGateway(this.jmsConnectionFactory)
 					.replyContainer()
 					.requestDestination("jmsPipelineTest"));
 		}
 
 		@Bean
 		public IntegrationFlow jmsInboundGatewayFlow() {
-			return IntegrationFlows.from(Jms.inboundGateway(this.jmsConnectionFactory)
+			return IntegrationFlows.from((MessagingGateways g) -> g.jms(this.jmsConnectionFactory)
 					.destination("jmsPipelineTest"))
 					.<String, String>transform(String::toUpperCase)
 					.get();
@@ -1416,7 +1418,7 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow ftpInboundFlow() {
 			return IntegrationFlows
-					.from(Ftp.inboundAdapter(this.ftpSessionFactory)
+					.from(s -> s.ftp(this.ftpSessionFactory)
 									.preserveTimestamp(true)
 									.remoteDirectory("ftpSource")
 									.regexFilter(".*\\.txt$")
@@ -1430,7 +1432,7 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow sftpInboundFlow() {
 			return IntegrationFlows
-					.from(Sftp.inboundAdapter(this.sftpSessionFactory)
+					.from(s -> s.sftp(this.sftpSessionFactory)
 									.preserveTimestamp(true)
 									.remoteDirectory("sftpSource")
 									.regexFilter(".*\\.txt$")
@@ -1482,15 +1484,18 @@ public class IntegrationFlowTests {
 					.channel(remoteFileOutputChannel())
 					.get();
 		}
+
 		@Bean
 		public IntegrationFlow sftpMGetFlow() {
 			return IntegrationFlows.from("sftpMgetInputChannel")
-					.handle(Sftp.outboundGateway(this.sftpSessionFactory, AbstractRemoteFileOutboundGateway.Command.MGET,
-							"payload")
-							.options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE)
-							.regexFileNameFilter("(subSftpSource|.*1.txt)")
-							.localDirectoryExpression("@sftpServer.targetLocalDirectoryName + #remoteDirectory")
-							.localFilenameGeneratorExpression("#remoteFileName.replaceFirst('sftpSource', 'localTarget')"))
+					.handleWithAdapter(h ->
+							h.sftpGateway(this.sftpSessionFactory, AbstractRemoteFileOutboundGateway.Command.MGET,
+									"payload")
+									.options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE)
+									.regexFileNameFilter("(subSftpSource|.*1.txt)")
+									.localDirectoryExpression("@sftpServer.targetLocalDirectoryName + #remoteDirectory")
+									.localFilenameGeneratorExpression(
+											"#remoteFileName.replaceFirst('sftpSource', 'localTarget')"))
 					.channel(remoteFileOutputChannel())
 					.get();
 		}
@@ -1559,7 +1564,8 @@ public class IntegrationFlowTests {
 
 		@Bean
 		public IntegrationFlow priorityFlow(PriorityCapableChannelMessageStore mongoDbChannelMessageStore) {
-			return IntegrationFlows.from(MessageChannels.priority("priorityChannel", mongoDbChannelMessageStore, "priorityGroup"))
+			return IntegrationFlows.from((Channels c) ->
+					c.priority("priorityChannel", mongoDbChannelMessageStore, "priorityGroup"))
 					.bridge(s -> s.poller(Pollers.fixedDelay(100))
 							.autoStartup(false)
 							.id("priorityChannelBridge"))
@@ -1627,7 +1633,7 @@ public class IntegrationFlowTests {
 					.bridge(c -> c.autoStartup(false).id("bridge"))
 					.fixedSubscriberChannel()
 					.delay("delayer", "200", c -> c.advice(this.delayedAdvice).messageStore(this.messageStore()))
-					.channel(MessageChannels.queue("bridgeFlow2Output"))
+					.channel(c -> c.queue("bridgeFlow2Output"))
 					.get();
 		}
 
@@ -1695,8 +1701,8 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow fileFlow1() {
 			return IntegrationFlows.from("fileFlow1Input")
-					.handle(Files.outboundAdapter(tmpDir).fileNameGenerator(message -> null),
-							c -> c.id("fileWriting"))
+					.<FileWritingMessageHandler>handleWithAdapter(h -> h.file(tmpDir).fileNameGenerator(message -> null)
+							, c -> c.id("fileWriting"))
 					.get();
 		}
 
@@ -1722,7 +1728,7 @@ public class IntegrationFlowTests {
 		@Bean
 		@DependsOn("enrichFlow")
 		public IntegrationFlow enricherFlow() {
-			return IntegrationFlows.fromFixedMessageChannel("enricherInput")
+			return IntegrationFlows.from("enricherInput", true)
 					.enrich(e -> e.requestChannel("enrichChannel")
 									.requestPayloadExpression("payload")
 									.shouldClonePayload(false)
@@ -1736,7 +1742,7 @@ public class IntegrationFlowTests {
 		@Bean
 		@DependsOn("enrichFlow")
 		public IntegrationFlow enricherFlow2() {
-			return IntegrationFlows.fromFixedMessageChannel("enricherInput2")
+			return IntegrationFlows.from("enricherInput2", true)
 					.enrich(e -> e.requestChannel("enrichChannel")
 									.requestPayloadExpression("payload")
 									.shouldClonePayload(false)
@@ -1749,7 +1755,7 @@ public class IntegrationFlowTests {
 		@Bean
 		@DependsOn("enrichFlow")
 		public IntegrationFlow enricherFlow3() {
-			return IntegrationFlows.fromFixedMessageChannel("enricherInput3")
+			return IntegrationFlows.from("enricherInput3", true)
 					.enrich(e -> e.requestChannel("enrichChannel")
 									.requestPayloadExpression("payload")
 									.shouldClonePayload(false)
@@ -1787,11 +1793,11 @@ public class IntegrationFlowTests {
 		public IntegrationFlow splitResequenceFlow() {
 			return f -> f.enrichHeaders(s -> s.header("FOO", "BAR"))
 					.split("testSplitterData", "buildList", c -> c.applySequence(false))
-					.channel(MessageChannels.executor(this.taskExecutor()))
+					.channel(c -> c.executor(this.taskExecutor()))
 					.split(Message.class, target -> (List<?>) target.getPayload(), c -> c.applySequence(false))
 					.channel(MessageChannels.executor(this.taskExecutor()))
 					.split(s -> s.applySequence(false).get().getT2().setDelimiters(","))
-					.channel(MessageChannels.executor(this.taskExecutor()))
+					.channel(c -> c.executor(this.taskExecutor()))
 					.<String, Integer>transform(Integer::parseInt)
 					.enrichHeaders(s -> s.headerExpression(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, "payload"))
 					.resequence(r -> r.releasePartialSequences(true).correlationExpression("'foo'"), null)
@@ -1800,7 +1806,7 @@ public class IntegrationFlowTests {
 
 		@Bean
 		public IntegrationFlow splitAggregateFlow() {
-			return IntegrationFlows.fromFixedMessageChannel("splitAggregateInput")
+			return IntegrationFlows.from("splitAggregateInput", true)
 					.split(null)
 					.channel(MessageChannels.executor(this.taskExecutor()))
 					.resequence()
@@ -1812,8 +1818,10 @@ public class IntegrationFlowTests {
 		public IntegrationFlow xpathHeaderEnricherFlow() {
 			return IntegrationFlows.from("xpathHeaderEnricherInput")
 					.enrichHeaders(
-							s -> s.header("one", new XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementOne"))
-									.header("two", new XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementTwo"))
+							s -> s.header("one",
+									new XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementOne"))
+									.header("two",
+											new XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementTwo"))
 									.headerChannelsToString(),
 							c -> c.autoStartup(false).id("xpathHeaderEnricher")
 					)
@@ -1874,11 +1882,11 @@ public class IntegrationFlowTests {
 
 		@Bean
 		public IntegrationFlow tailFlow() {
-			return IntegrationFlows.from(Files.tailAdapter(new File(tmpDir, "TailTest"))
-						.delay(500)
-						.end(false)
-						.id("tailer")
-						.autoStartup(false))
+			return IntegrationFlows.from((MessageProducers p) -> p.tail(new File(tmpDir, "TailTest"))
+					.delay(500)
+					.end(false)
+					.id("tailer")
+					.autoStartup(false))
 					.transform("hello "::concat)
 					.channel(MessageChannels.queue("tailChannel"))
 					.get();
@@ -1922,7 +1930,7 @@ public class IntegrationFlowTests {
 
 		@Bean
 		public IntegrationFlow amqpInboundFlow() {
-			return IntegrationFlows.from(Amqp.inboundAdapter(this.rabbitConnectionFactory, fooQueue()))
+			return IntegrationFlows.from((MessageProducers p) -> p.amqp(this.rabbitConnectionFactory, fooQueue()))
 					.transform(String.class, String::toUpperCase)
 					.channel(Amqp.pollableChannel(this.rabbitConnectionFactory)
 							.queueName("amqpReplyChannel")
@@ -1955,7 +1963,7 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow fileReadingFlow() {
 			return IntegrationFlows
-					.from(Files.inboundAdapter(tmpDir).patternFilter("*.sitest"),
+					.from(s -> s.file(tmpDir).patternFilter("*.sitest"),
 							e -> e.poller(Pollers.fixedDelay(100)))
 					.transform(Transformers.fileToString())
 					.aggregate(a -> a.correlationExpression("1")

spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/MailTests.java

@@ -48,9 +48,10 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.MessageProducers;
 import org.springframework.integration.dsl.channel.MessageChannels;
 import org.springframework.integration.dsl.mail.Mail;
-import org.springframework.integration.dsl.support.Pollers;
+import org.springframework.integration.dsl.core.Pollers;
 import org.springframework.integration.dsl.test.mail.PoorMansMailServer.ImapServer;
 import org.springframework.integration.dsl.test.mail.PoorMansMailServer.Pop3Server;
 import org.springframework.integration.dsl.test.mail.PoorMansMailServer.SmtpServer;
@@ -197,7 +198,7 @@ public class MailTests {
 		public IntegrationFlow sendMailFlow() {
 			return IntegrationFlows.from("sendMailChannel")
 					.enrichHeaders(Mail.headers().subject("foo").from("foo@bar").to("bar@baz"))
-					.handle(Mail.outboundAdapter("localhost")
+					.handleWithAdapter(h -> h.mail("localhost")
 									.port(smtpPort)
 									.credentials("user", "pw")
 									.protocol("smtp")
@@ -209,10 +210,9 @@ public class MailTests {
 		@Bean
 		public IntegrationFlow pop3MailFlow() {
 			return IntegrationFlows
-					.from(Mail.pop3InboundAdapter("localhost", pop3Port, "user", "pw")
+					.from(s -> s.pop3("localhost", pop3Port, "user", "pw")
 									.javaMailProperties(p -> p.put("mail.debug", "true")),
-							e -> e.autoStartup(true)
-									.poller(Pollers.fixedDelay(1000)))
+							e -> e.autoStartup(true).poller(p -> p.fixedDelay(1000)))
 					.enrichHeaders(s -> s.headerExpressions(c -> c.put(MailHeaders.SUBJECT, "payload.subject")
 									.put(MailHeaders.FROM, "payload.from[0].toString()")))
 					.channel(MessageChannels.queue("pop3Channel"))
@@ -222,11 +222,11 @@ public class MailTests {
 		@Bean
 		public IntegrationFlow imapMailFlow() {
 			return IntegrationFlows
-					.from(Mail.imapInboundAdapter("imap://user:pw@localhost:" + imapPort + "/INBOX")
+					.from(s -> s.imap("imap://user:pw@localhost:" + imapPort + "/INBOX")
 									.searchTermStrategy(this::fromAndNotSeenTerm)
 									.javaMailProperties(p -> p.put("mail.debug", "true")),
 							e -> e.autoStartup(true)
-									.poller(Pollers.fixedDelay(1000)))
+									.poller(p -> p.fixedDelay(1000)))
 					.channel(MessageChannels.queue("imapChannel"))
 					.get();
 		}
@@ -234,7 +234,7 @@ public class MailTests {
 		@Bean
 		public IntegrationFlow imapIdleFlow() {
 			return IntegrationFlows
-					.from(Mail.imapIdleAdapter("imap://user:pw@localhost:" + imapIdlePort + "/INBOX")
+					.from((MessageProducers mp) -> mp.imap("imap://user:pw@localhost:" + imapIdlePort + "/INBOX")
 							.searchTermStrategy(this::fromAndNotSeenTerm)
 							.javaMailProperties(p -> p.put("mail.debug", "true")
 									.put("mail.imap.connectionpoolsize", "5"))

spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/PoorMansMailServer.java

@@ -276,7 +276,8 @@ public class PoorMansMailServer {
 									+ "((\"Bar\" NIL \"bar\" \"baz.net\")) NIL NIL "
 									+ "\"<4DA0A7E4.3010506@baz.net>\" "
 									+ "\"<CACVnpJkAUUfa3d_-4GNZW2WpxbB39tBCHC=T0gc7hty6dOEHcA@foo.bar.com>\") "
-									+ "BODYSTRUCTURE (\"TEXT\" \"PLAIN\" (\"CHARSET\" \"ISO-8859-1\") NIL NIL \"7BIT\" 1176 43)))");
+									+ "BODYSTRUCTURE (\"TEXT\" \"PLAIN\" (\"CHARSET\" \"ISO-8859-1\") NIL NIL "
+									+ "\"7BIT\" 1176 43)))");
 							write(tag + "OK FETCH completed");
 						}
 						else if (line.contains("STORE 1 +FLAGS (\\Flagged)")) {