DSL: Add File. Namespace Builder Factory

* Introduce `ComponentsRegistration` *marker* to extract internal components from the `IntegrationComponentSpec` to be registered as bean in the application context
* Add `ConsumerEndpointSpec#order(int order)` for the `order` of target `AbstractMessageHandler`
* Fix `AggregatorSpec` to use `DefaultAggregatingMessageGroupProcessor` by default
* Fix `IntegrationFlowBeanPostProcessor#generateBeanName` to check if `instance instanceof NamedComponent` and its `beanName` has been configured
using `IntegrationComponentSpec#id`

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

@@ -17,6 +17,7 @@
 package org.springframework.integration.dsl;
 
 import org.springframework.integration.aggregator.AggregatingMessageHandler;
+import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor;
 import org.springframework.integration.aggregator.ExpressionEvaluatingMessageGroupProcessor;
 import org.springframework.integration.aggregator.MessageGroupProcessor;
 import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor;
@@ -26,7 +27,7 @@ import org.springframework.integration.aggregator.MethodInvokingMessageGroupProc
  */
 public class AggregatorSpec extends CorrelationHandlerSpec<AggregatorSpec, AggregatingMessageHandler> {
 
-	private MessageGroupProcessor outputProcessor;
+	private MessageGroupProcessor outputProcessor = new DefaultAggregatingMessageGroupProcessor();
 
 	private boolean expireGroupsUponCompletion;
 

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

@@ -22,7 +22,6 @@ import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler;
 import org.springframework.integration.aggregator.AggregatingMessageHandler;
-import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor;
 import org.springframework.integration.aggregator.ResequencingMessageHandler;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.FixedSubscriberChannel;
@@ -94,6 +93,13 @@ public final class IntegrationFlowBuilder {
 		return this;
 	}
 
+	IntegrationFlowBuilder addComponents(Collection<Object> components) {
+		for (Object component : components) {
+			this.flow.addComponent(component);
+		}
+		return this;
+	}
+
 	IntegrationFlowBuilder currentComponent(Object component) {
 		this.currentComponent = component;
 		return this;
@@ -406,8 +412,7 @@ public final class IntegrationFlowBuilder {
 
 	public IntegrationFlowBuilder
 	aggregate(EndpointConfigurer<GenericEndpointSpec<AggregatingMessageHandler>> endpointConfigurer) {
-		return handle(new AggregatorSpec().outputProcessor(new DefaultAggregatingMessageGroupProcessor()).get(),
-				endpointConfigurer);
+		return handle(new AggregatorSpec().get(), endpointConfigurer);
 	}
 
 	public IntegrationFlowBuilder aggregate(ComponentConfigurer<AggregatorSpec> aggregatorConfigurer,

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

@@ -20,6 +20,7 @@ import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.dsl.channel.MessageChannelSpec;
+import org.springframework.integration.dsl.core.ComponentsRegistration;
 import org.springframework.integration.dsl.core.MessageSourceSpec;
 import org.springframework.integration.dsl.core.MessagingGatewaySpec;
 import org.springframework.integration.dsl.core.MessagingProducerSpec;
@@ -56,22 +57,22 @@ public final class IntegrationFlows {
 		return from(new FixedSubscriberChannelPrototype(messageChannelName));
 	}
 
-	public static IntegrationFlowBuilder from(MessageChannel messageChannel) {
-		return new IntegrationFlowBuilder().channel(messageChannel);
-	}
-
 	public static IntegrationFlowBuilder from(MessageChannelSpec<?, ?> messageChannelSpec) {
 		return from(messageChannelSpec.get());
 	}
 
-	public static <S extends MessageSourceSpec<S, ? extends MessageSource<?>>> IntegrationFlowBuilder from(S
-			messageSourceSpec) {
-		return from(messageSourceSpec.get());
+	public static IntegrationFlowBuilder from(MessageChannel messageChannel) {
+		return new IntegrationFlowBuilder().channel(messageChannel);
 	}
 
-	public static <S extends MessageSourceSpec<S, ? extends MessageSource<?>>> IntegrationFlowBuilder from(S messageSource,
-			EndpointConfigurer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
-		return from(messageSource.get(), endpointConfigurer);
+	public static <S extends MessageSourceSpec<S, ? extends MessageSource<?>>> IntegrationFlowBuilder
+	from(S messageSourceSpec) {
+		return from(messageSourceSpec, null);
+	}
+
+	public static <S extends MessageSourceSpec<S, ? extends MessageSource<?>>> IntegrationFlowBuilder
+	from(S messageSourceSpec, EndpointConfigurer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
+		return from(messageSourceSpec.get(), endpointConfigurer, registerComponents(messageSourceSpec));
 	}
 
 	public static IntegrationFlowBuilder from(MessageSource<?> messageSource) {
@@ -80,31 +81,50 @@ public final class IntegrationFlows {
 
 	public static IntegrationFlowBuilder from(MessageSource<?> messageSource,
 			EndpointConfigurer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
+		return from(messageSource, endpointConfigurer, null);
+	}
+
+	private static IntegrationFlowBuilder from(MessageSource<?> messageSource,
+			EndpointConfigurer<SourcePollingChannelAdapterSpec> endpointConfigurer,
+			IntegrationFlowBuilder integrationFlowBuilder) {
 		SourcePollingChannelAdapterSpec spec = new SourcePollingChannelAdapterSpec(messageSource);
 		if (endpointConfigurer != null) {
 			endpointConfigurer.configure(spec);
 		}
-		return new IntegrationFlowBuilder()
-				.addComponent(spec)
+		if (integrationFlowBuilder == null) {
+			integrationFlowBuilder = new IntegrationFlowBuilder();
+		}
+		return integrationFlowBuilder.addComponent(spec)
 				.currentComponent(spec);
 	}
 
 	public static IntegrationFlowBuilder from(MessagingProducerSpec<?, ?> messagingProducerSpec) {
-		return from(messagingProducerSpec.get());
+		return from(messagingProducerSpec.get(), registerComponents(messagingProducerSpec));
 	}
 
 	public static IntegrationFlowBuilder from(MessageProducerSupport messageProducer) {
+		return from(messageProducer, null);
+	}
+
+	private static IntegrationFlowBuilder from(MessageProducerSupport messageProducer,
+			IntegrationFlowBuilder integrationFlowBuilder) {
 		DirectFieldAccessor dfa = new DirectFieldAccessor(messageProducer);
 		MessageChannel outputChannel = (MessageChannel) dfa.getPropertyValue("outputChannel");
 		if (outputChannel == null) {
 			outputChannel = new DirectChannel();
 			messageProducer.setOutputChannel(outputChannel);
 		}
-		return from(outputChannel).addComponent(messageProducer);
+		if (integrationFlowBuilder == null) {
+			integrationFlowBuilder = from(outputChannel);
+		}
+		else {
+			integrationFlowBuilder.channel(outputChannel);
+		}
+		return integrationFlowBuilder.addComponent(messageProducer);
 	}
 
 	public static IntegrationFlowBuilder from(MessagingGatewaySpec<?, ?> inboundGatewaySpec) {
-		return from(inboundGatewaySpec.get());
+		return from(inboundGatewaySpec.get(), registerComponents(inboundGatewaySpec));
 	}
 
 	public static IntegrationFlowBuilder from(MessagingGatewaySupport inboundGateway) {
@@ -117,6 +137,31 @@ public final class IntegrationFlows {
 		return from(outputChannel).addComponent(inboundGateway);
 	}
 
+	private static IntegrationFlowBuilder from(MessagingGatewaySupport inboundGateway,
+			IntegrationFlowBuilder integrationFlowBuilder) {
+		DirectFieldAccessor dfa = new DirectFieldAccessor(inboundGateway);
+		MessageChannel outputChannel = (MessageChannel) dfa.getPropertyValue("requestChannel");
+		if (outputChannel == null) {
+			outputChannel = new DirectChannel();
+			inboundGateway.setRequestChannel(outputChannel);
+		}
+		if (integrationFlowBuilder == null) {
+			integrationFlowBuilder = from(outputChannel);
+		}
+		else {
+			integrationFlowBuilder.channel(outputChannel);
+		}
+		return integrationFlowBuilder.addComponent(inboundGateway);
+	}
+
+	private static IntegrationFlowBuilder registerComponents(Object spec) {
+		if (spec instanceof ComponentsRegistration) {
+			return new IntegrationFlowBuilder()
+					.addComponents(((ComponentsRegistration) spec).getComponentsToRegister());
+		}
+		return null;
+	}
+
 	private IntegrationFlows() {
 	}
 

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

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2014 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.dsl.core;
+
+import java.util.Collection;
+
+/**
+ * @author Artem Bilan
+ */
+public interface ComponentsRegistration {
+
+	Collection<Object> getComponentsToRegister();
+
+}

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

@@ -23,13 +23,13 @@ import java.util.List;
 import org.aopalliance.aop.Advice;
 
 import org.springframework.integration.config.ConsumerEndpointFactoryBean;
+import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.messaging.MessageHandler;
 
 /**
  * @author Artem Bilan
-
  */
 public abstract class ConsumerEndpointSpec<S extends ConsumerEndpointSpec<S, H>, H extends MessageHandler>
 		extends EndpointSpec<S, ConsumerEndpointFactoryBean, H> {
@@ -89,4 +89,15 @@ public abstract class ConsumerEndpointSpec<S extends ConsumerEndpointSpec<S, H>,
 		return _this();
 	}
 
+	public S order(int order) {
+		H handler = this.target.getT2();
+		if (handler instanceof AbstractMessageHandler) {
+			((AbstractMessageHandler) handler).setOrder(order);
+		}
+		else {
+			logger.warn("'order' can be applied only for AbstractMessageHandler");
+		}
+		return _this();
+	}
+
 }

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

@@ -34,6 +34,7 @@ import org.springframework.integration.config.SourcePollingChannelAdapterFactory
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.SourcePollingChannelAdapterSpec;
 import org.springframework.integration.dsl.support.MessageChannelReference;
+import org.springframework.integration.support.context.NamedComponent;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.util.Assert;
@@ -156,6 +157,9 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean
 	}
 
 	private String generateBeanName(Object instance) {
+		if (instance instanceof NamedComponent && ((NamedComponent) instance).getComponentName() != null) {
+			return ((NamedComponent) instance).getComponentName();
+		}
 		String generatedBeanName = instance.getClass().getName();
 		String id = instance.getClass().getName();
 		int counter = -1;

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

@@ -55,7 +55,7 @@ public abstract class MessagingProducerSpec<S extends MessagingProducerSpec<S, P
 	}
 
 	@Override
-	protected final P doGet() {
+	protected P doGet() {
 		throw new UnsupportedOperationException();
 	}
 

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

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2014 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.dsl.file;
+
+import java.util.Comparator;
+
+/**
+ * @author Artem Bilan
+ */
+public abstract class File {
+
+	public static FileInboundChannelAdapterSpec inboundAdapter(java.io.File directory) {
+		return inboundAdapter(directory, null);
+	}
+
+	public static FileInboundChannelAdapterSpec inboundAdapter(java.io.File directory,
+			Comparator<java.io.File> receptionOrderComparator) {
+		return new FileInboundChannelAdapterSpec(receptionOrderComparator).directory(directory);
+	}
+
+	public static FileWritingMessageHandlerSpec outboundAdapter(java.io.File destinationDirectory) {
+		return new FileWritingMessageHandlerSpec(destinationDirectory).expectReply(false);
+	}
+
+	public static FileWritingMessageHandlerSpec outboundAdapter(String directoryExpression) {
+		return new FileWritingMessageHandlerSpec(directoryExpression).expectReply(false);
+	}
+
+	public static FileWritingMessageHandlerSpec outboundGateway(java.io.File destinationDirectory) {
+		return new FileWritingMessageHandlerSpec(destinationDirectory).expectReply(true);
+	}
+
+	public static FileWritingMessageHandlerSpec outboundGateway(String directoryExpression) {
+		return new FileWritingMessageHandlerSpec(directoryExpression).expectReply(true);
+	}
+
+	public static TailAdapterSpec tailAdapter(java.io.File file) {
+		return new TailAdapterSpec().file(file);
+	}
+
+}

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

@@ -0,0 +1,138 @@
+/*
+ * 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.file;
+
+import java.io.File;
+import java.util.Comparator;
+
+import org.springframework.integration.dsl.core.MessageSourceSpec;
+import org.springframework.integration.file.DirectoryScanner;
+import org.springframework.integration.file.FileLocker;
+import org.springframework.integration.file.FileReadingMessageSource;
+import org.springframework.integration.file.filters.AcceptAllFileListFilter;
+import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
+import org.springframework.integration.file.filters.CompositeFileListFilter;
+import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.file.filters.RegexPatternFileListFilter;
+import org.springframework.integration.file.filters.SimplePatternFileListFilter;
+import org.springframework.integration.file.locking.NioFileLocker;
+import org.springframework.util.Assert;
+
+/**
+ * @author Artem Bilan
+ */
+public class FileInboundChannelAdapterSpec
+		extends MessageSourceSpec<FileInboundChannelAdapterSpec, FileReadingMessageSource> {
+
+	private FileListFilter<File> filter;
+
+	private FileLocker locker;
+
+	FileInboundChannelAdapterSpec() {
+		this.target = new FileReadingMessageSource();
+	}
+
+	FileInboundChannelAdapterSpec(Comparator<File> receptionOrderComparator) {
+		this.target = new FileReadingMessageSource(receptionOrderComparator);
+	}
+
+	FileInboundChannelAdapterSpec directory(File directory) {
+		this.target.setDirectory(directory);
+		return _this();
+	}
+
+	public FileInboundChannelAdapterSpec scanner(DirectoryScanner scanner) {
+		this.target.setScanner(scanner);
+		return _this();
+	}
+
+	public FileInboundChannelAdapterSpec autoCreateDirectory(boolean autoCreateDirectory) {
+		this.target.setAutoCreateDirectory(autoCreateDirectory);
+		return _this();
+	}
+
+	public FileInboundChannelAdapterSpec filter(FileListFilter<File> filter) {
+		return filter(filter, false);
+	}
+
+	public FileInboundChannelAdapterSpec filter(FileListFilter<File> filter, boolean preventDuplicates) {
+		Assert.isNull(this.filter,
+				"The 'filter' (" + this.filter + ") is already configured for the FileReadingMessageSource");
+		FileListFilter<File> targetFilter = filter;
+		if (preventDuplicates) {
+			targetFilter = createCompositeWithAcceptOnceFilter(filter);
+		}
+		this.filter = targetFilter;
+		this.target.setFilter(targetFilter);
+		return _this();
+	}
+
+	public FileInboundChannelAdapterSpec preventDuplicatesFilter(boolean preventDuplicates) {
+		if (preventDuplicates) {
+			return filter(new AcceptOnceFileListFilter<File>(), false);
+		}
+		else {
+			return filter(new AcceptAllFileListFilter<File>(), false);
+		}
+	}
+
+	public FileInboundChannelAdapterSpec patternFilter(String pattern) {
+		return patternFilter(pattern, true);
+	}
+
+	public FileInboundChannelAdapterSpec patternFilter(String pattern, boolean preventDuplicates) {
+		return filter(new SimplePatternFileListFilter(pattern), preventDuplicates);
+	}
+
+	public FileInboundChannelAdapterSpec regexFilter(String regex) {
+		return regexFilter(regex, true);
+	}
+
+	public FileInboundChannelAdapterSpec regexFilter(String regex, boolean preventDuplicates) {
+		return filter(new RegexPatternFileListFilter(regex), preventDuplicates);
+	}
+
+	private CompositeFileListFilter<File> createCompositeWithAcceptOnceFilter(FileListFilter<File> otherFilter) {
+		CompositeFileListFilter<File> compositeFilter = new CompositeFileListFilter<File>();
+		compositeFilter.addFilter(new AcceptOnceFileListFilter<File>());
+		compositeFilter.addFilter(otherFilter);
+		return compositeFilter;
+	}
+
+	public FileInboundChannelAdapterSpec locker(FileLocker locker) {
+		Assert.isNull(this.locker,
+				"The 'locker' (" + this.locker + ") is already configured for the FileReadingMessageSource");
+		this.locker = locker;
+		this.target.setLocker(locker);
+		return _this();
+	}
+
+	public FileInboundChannelAdapterSpec nioLocker() {
+		return locker(new NioFileLocker());
+	}
+
+	public FileInboundChannelAdapterSpec scanEachPoll(boolean scanEachPoll) {
+		this.target.setScanEachPoll(scanEachPoll);
+		return _this();
+	}
+
+	@Override
+	protected FileReadingMessageSource doGet() {
+		throw new UnsupportedOperationException();
+	}
+
+}

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

@@ -0,0 +1,107 @@
+/*
+ * 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.file;
+
+import java.util.Collection;
+import java.util.Collections;
+
+import org.springframework.integration.dsl.core.ComponentsRegistration;
+import org.springframework.integration.dsl.core.MessageHandlerSpec;
+import org.springframework.integration.file.DefaultFileNameGenerator;
+import org.springframework.integration.file.FileNameGenerator;
+import org.springframework.integration.file.FileWritingMessageHandler;
+import org.springframework.integration.file.support.FileExistsMode;
+import org.springframework.util.Assert;
+
+/**
+ * @author Artem Bilan
+ */
+public class FileWritingMessageHandlerSpec
+		extends MessageHandlerSpec<FileWritingMessageHandlerSpec, FileWritingMessageHandler>
+		implements ComponentsRegistration {
+
+	private FileNameGenerator fileNameGenerator;
+
+	private DefaultFileNameGenerator defaultFileNameGenerator;
+
+	FileWritingMessageHandlerSpec(java.io.File destinationDirectory) {
+		this.target = new FileWritingMessageHandler(destinationDirectory);
+	}
+
+	FileWritingMessageHandlerSpec(String directoryExpression) {
+		this.target = new FileWritingMessageHandler(PARSER.parseExpression(directoryExpression));
+	}
+
+	FileWritingMessageHandlerSpec expectReply(boolean expectReply) {
+		target.setExpectReply(expectReply);
+		return _this();
+	}
+
+	public FileWritingMessageHandlerSpec autoCreateDirectory(boolean autoCreateDirectory) {
+		target.setAutoCreateDirectory(autoCreateDirectory);
+		return _this();
+	}
+
+	public FileWritingMessageHandlerSpec temporaryFileSuffix(String temporaryFileSuffix) {
+		target.setTemporaryFileSuffix(temporaryFileSuffix);
+		return _this();
+	}
+
+	public FileWritingMessageHandlerSpec fileExistsMode(FileExistsMode fileExistsMode) {
+		target.setFileExistsMode(fileExistsMode);
+		return _this();
+	}
+
+	public FileWritingMessageHandlerSpec fileNameGenerator(FileNameGenerator fileNameGenerator) {
+		this.fileNameGenerator = fileNameGenerator;
+		target.setFileNameGenerator(fileNameGenerator);
+		return _this();
+	}
+
+	public FileWritingMessageHandlerSpec fileNameGeneratorExpression(String fileNameGeneratorExpression) {
+		Assert.isNull(this.fileNameGenerator,
+				"'fileNameGenerator' and 'fileNameGeneratorExpression' are mutually exclusive.");
+		this.defaultFileNameGenerator = new DefaultFileNameGenerator();
+		this.defaultFileNameGenerator.setExpression(fileNameGeneratorExpression);
+		return _this();
+	}
+
+
+	public FileWritingMessageHandlerSpec deleteSourceFiles(boolean deleteSourceFiles) {
+		target.setDeleteSourceFiles(deleteSourceFiles);
+		return _this();
+	}
+
+	public FileWritingMessageHandlerSpec charset(String charset) {
+		target.setCharset(charset);
+		return _this();
+	}
+
+	@Override
+	public Collection<Object> getComponentsToRegister() {
+		if (this.defaultFileNameGenerator != null) {
+			return Collections.<Object>singletonList(this.defaultFileNameGenerator);
+		}
+		return Collections.<Object>emptyList();
+	}
+
+	@Override
+	protected FileWritingMessageHandler doGet() {
+		throw new UnsupportedOperationException();
+	}
+
+}

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

@@ -0,0 +1,136 @@
+/*
+ * 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.file;
+
+import org.springframework.beans.factory.support.DefaultListableBeanFactory;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.integration.channel.NullChannel;
+import org.springframework.integration.dsl.core.MessagingProducerSpec;
+import org.springframework.integration.file.config.FileTailInboundChannelAdapterFactoryBean;
+import org.springframework.integration.file.tail.FileTailingMessageProducerSupport;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.util.Assert;
+
+/**
+ * @author Artem Bilan
+ */
+public class TailAdapterSpec extends MessagingProducerSpec<TailAdapterSpec, FileTailingMessageProducerSupport> {
+
+	private final FileTailInboundChannelAdapterFactoryBean factoryBean = new FileTailInboundChannelAdapterFactoryBean();
+
+	private MessageChannel outputChannel;
+
+	private MessageChannel errorChannel;
+
+	TailAdapterSpec() {
+		super(null);
+		this.factoryBean.setBeanFactory(new DefaultListableBeanFactory());
+	}
+
+	TailAdapterSpec file(java.io.File file) {
+		Assert.notNull(file);
+		this.factoryBean.setFile(file);
+		return _this();
+	}
+
+	public TailAdapterSpec nativeOptions(String nativeOptions) {
+		this.factoryBean.setNativeOptions(nativeOptions);
+		return _this();
+	}
+
+	public TailAdapterSpec taskExecutor(TaskExecutor taskExecutor) {
+		this.factoryBean.setTaskExecutor(taskExecutor);
+		return _this();
+	}
+
+	public TailAdapterSpec taskScheduler(TaskScheduler taskScheduler) {
+		this.factoryBean.setTaskScheduler(taskScheduler);
+		return _this();
+	}
+
+	public TailAdapterSpec delay(long delay) {
+		this.factoryBean.setDelay(delay);
+		return _this();
+	}
+
+	public TailAdapterSpec fileDelay(long fileDelay) {
+		this.factoryBean.setFileDelay(fileDelay);
+		return _this();
+	}
+
+	public TailAdapterSpec end(boolean end) {
+		this.factoryBean.setEnd(end);
+		return _this();
+	}
+
+	public TailAdapterSpec reopen(boolean reopen) {
+		this.factoryBean.setReopen(reopen);
+		return _this();
+	}
+
+	@Override
+	public TailAdapterSpec id(String id) {
+		this.factoryBean.setBeanName(id);
+		return _this();
+	}
+
+	@Override
+	public TailAdapterSpec phase(int phase) {
+		this.factoryBean.setPhase(phase);
+		return _this();
+	}
+
+	@Override
+	public TailAdapterSpec autoStartup(boolean autoStartup) {
+		this.factoryBean.setAutoStartup(autoStartup);
+		return _this();
+	}
+
+	@Override
+	public TailAdapterSpec outputChannel(MessageChannel outputChannel) {
+		this.outputChannel = outputChannel;
+		return _this();
+	}
+
+	@Override
+	public TailAdapterSpec errorChannel(MessageChannel errorChannel) {
+		this.errorChannel = errorChannel;
+		return _this();
+	}
+
+	@Override
+	protected FileTailingMessageProducerSupport doGet() {
+		if (this.outputChannel == null) {
+			this.factoryBean.setOutputChannel(new NullChannel());
+		}
+		FileTailingMessageProducerSupport tailingMessageProducerSupport = null;
+		try {
+			this.factoryBean.afterPropertiesSet();
+			tailingMessageProducerSupport = this.factoryBean.getObject();
+		}
+		catch (Exception e) {
+			throw new IllegalStateException(e);
+		}
+		if (this.errorChannel != null) {
+			tailingMessageProducerSupport.setErrorChannel(this.errorChannel);
+		}
+		tailingMessageProducerSupport.setOutputChannel(this.outputChannel);
+		return tailingMessageProducerSupport;
+	}
+
+}

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

@@ -0,0 +1,4 @@
+/**
+ * Provides File Components support for Spring Integration Java DSL.
+ */
+package org.springframework.integration.dsl.file;

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

@@ -17,11 +17,20 @@
 package org.springframework.integration.dsl.test;
 
 import static org.hamcrest.Matchers.*;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
-import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -34,13 +43,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
-import com.mongodb.MongoClient;
-import de.flapdoodle.embed.mongo.MongodExecutable;
-import de.flapdoodle.embed.mongo.MongodStarter;
-import de.flapdoodle.embed.mongo.config.MongodConfigBuilder;
-import de.flapdoodle.embed.mongo.config.Net;
-import de.flapdoodle.embed.mongo.distribution.Version;
-import de.flapdoodle.embed.process.runtime.Network;
 import org.aopalliance.aop.Advice;
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
@@ -92,15 +94,15 @@ import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.amqp.Amqp;
 import org.springframework.integration.dsl.channel.DirectChannelSpec;
 import org.springframework.integration.dsl.channel.MessageChannels;
+import org.springframework.integration.dsl.file.File;
 import org.springframework.integration.dsl.jms.Jms;
 import org.springframework.integration.dsl.support.Pollers;
+import org.springframework.integration.dsl.support.Transformers;
 import org.springframework.integration.endpoint.MethodInvokingMessageSource;
 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.tail.ApacheCommonsFileTailingMessageProducer;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice;
 import org.springframework.integration.mongodb.store.MongoDbChannelMessageStore;
@@ -111,6 +113,7 @@ import org.springframework.integration.store.PriorityCapableChannelMessageStore;
 import org.springframework.integration.store.SimpleMessageStore;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.support.MutableMessageBuilder;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.integration.transformer.PayloadDeserializingTransformer;
 import org.springframework.integration.transformer.PayloadSerializingTransformer;
 import org.springframework.integration.xml.transformer.support.XPathExpressionEvaluatingHeaderValueMessageProcessor;
@@ -135,6 +138,15 @@ import org.springframework.stereotype.Service;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.util.StreamUtils;
+
+import com.mongodb.MongoClient;
+import de.flapdoodle.embed.mongo.MongodExecutable;
+import de.flapdoodle.embed.mongo.MongodStarter;
+import de.flapdoodle.embed.mongo.config.MongodConfigBuilder;
+import de.flapdoodle.embed.mongo.config.Net;
+import de.flapdoodle.embed.mongo.distribution.Version;
+import de.flapdoodle.embed.process.runtime.Network;
 
 /**
  * @author Artem Bilan
@@ -144,7 +156,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 @DirtiesContext
 public class IntegrationFlowTests {
 
-	private static final File tmpDir = new File(System.getProperty("java.io.tmpdir"));
+	private static final java.io.File tmpDir = new java.io.File(System.getProperty("java.io.tmpdir"));
 
 	private static int mongoPort;
 
@@ -203,7 +215,7 @@ public class IntegrationFlowTests {
 	private MessageChannel fileFlow1Input;
 
 	@Autowired
-	@Qualifier("fileWritingMessageHandler")
+	@Qualifier("fileWriting.handler")
 	private MessageHandler fileWritingMessageHandler;
 
 	@Autowired
@@ -337,11 +349,12 @@ public class IntegrationFlowTests {
 			assertNotNull(message);
 			assertEquals("" + i, message.getPayload());
 		}
+		this.controlBus.send("@integerEndpoint.stop()");
 
 		assertTrue(((ChannelInterceptorAware) this.outputChannel).getChannelInterceptors()
 				.contains(this.testChannelInterceptor));
 		assertThat(this.testChannelInterceptor.getInvoked(), Matchers.greaterThanOrEqualTo(5));
-		this.controlBus.send("@integerEndpoint.stop()");
+
 	}
 
 	@Test
@@ -476,7 +489,7 @@ public class IntegrationFlowTests {
 		dfa.setPropertyValue("fileNameGenerator", fileNameGenerator);
 		this.fileFlow1Input.send(message);
 
-		assertTrue(new File(tmpDir, "foo").exists());
+		assertTrue(new java.io.File(tmpDir, "foo").exists());
 	}
 
 	@Test
@@ -492,7 +505,7 @@ public class IntegrationFlowTests {
 	}
 
 	@Test
-	public void testLamdas() {
+	public void testLambdas() {
 		QueueChannel replyChannel = new QueueChannel();
 		Message<?> message = MessageBuilder.withPayload("World")
 				.setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel)
@@ -867,12 +880,10 @@ public class IntegrationFlowTests {
 
 	@Test
 	public void testMessageProducerFlow() throws Exception {
-		FileOutputStream file = new FileOutputStream(new File(tmpDir, "TailTest"));
+		FileOutputStream file = new FileOutputStream(new java.io.File(tmpDir, "TailTest"));
 		for (int i = 0; i < 50; i++) {
 			file.write((i + "\n").getBytes());
 		}
-		file.flush();
-		file.close();
 
 		for (int i = 0; i < 50; i++) {
 			Message<?> message = this.tailChannel.receive(5000);
@@ -880,6 +891,9 @@ public class IntegrationFlowTests {
 			assertEquals("hello " + i, message.getPayload());
 		}
 		assertNull(this.tailChannel.receive(1));
+
+		this.controlBus.send("@tailer.stop()");
+		file.close();
 	}
 
 
@@ -991,6 +1005,59 @@ public class IntegrationFlowTests {
 		assertEquals("HELLO THROUGH THE JMS PIPELINE", receive.getPayload());
 	}
 
+	@Autowired
+	@Qualifier("fileReadingResultChannel")
+	private PollableChannel fileReadingResultChannel;
+
+	@Test
+	public void testFileReadingFlow() throws Exception {
+		List<Integer> evens = new ArrayList<>(25);
+		for (int i = 0; i < 50; i++) {
+			boolean even = i % 2 == 0;
+			String extension = even ? ".sitest" : ".foofile";
+			if (even) {
+				evens.add(i);
+			}
+			FileOutputStream file = new FileOutputStream(new java.io.File(tmpDir, i + extension));
+			file.write(("" + i).getBytes());
+			file.flush();
+			file.close();
+		}
+
+		Message<?> message = fileReadingResultChannel.receive(10000);
+		assertNotNull(message);
+		Object payload = message.getPayload();
+		assertThat(payload, instanceOf(List.class));
+		@SuppressWarnings("unchecked")
+		List<String> result = (List<String>) payload;
+		assertEquals(25, result.size());
+		result.forEach(s -> assertTrue(evens.contains(Integer.parseInt(s))));
+	}
+
+
+	@Autowired
+	@Qualifier("fileWritingInput")
+	private MessageChannel fileWritingInput;
+
+	@Autowired
+	@Qualifier("fileWritingResultChannel")
+	private PollableChannel fileWritingResultChannel;
+
+	@Test
+	public void testFileWritingFlow() throws Exception {
+		String payload = "Spring Integration";
+		this.fileWritingInput.send(new GenericMessage<>(payload));
+		Message<?> receive = this.fileWritingResultChannel.receive(1000);
+		assertNotNull(receive);
+		assertThat(receive.getPayload(), instanceOf(java.io.File.class));
+		java.io.File resultFile = (java.io.File) receive.getPayload();
+		assertThat(resultFile.getAbsolutePath(),
+				endsWith(TestUtils.applySystemFileSeparator("fileWritingFlow/foo.sitest")));
+		String fileContent = StreamUtils.copyToString(new FileInputStream(resultFile), Charset.defaultCharset());
+		assertEquals(payload, fileContent);
+	}
+
+
 	@MessagingGateway(defaultRequestChannel = "controlBus")
 	private static interface ControlBusGateway {
 
@@ -1021,7 +1088,7 @@ public class IntegrationFlowTests {
 		@Bean
 		public IntegrationFlow flow1() {
 			return IntegrationFlows.from(this.integerMessageSource(),
-					c -> c.poller(Pollers.fixedRate(1000, 2000))
+					c -> c.poller(Pollers.fixedRate(100))
 							.id("integerEndpoint")
 							.autoStartup(false))
 					.fixedSubscriberChannel("integerChannel")
@@ -1296,18 +1363,11 @@ public class IntegrationFlowTests {
 	@Configuration
 	public static class ContextConfiguration4 {
 
-		@Bean
-		public MessageHandler fileWritingMessageHandler() {
-			FileWritingMessageHandler fileWritingMessageHandler = new FileWritingMessageHandler(tmpDir);
-			fileWritingMessageHandler.setFileNameGenerator(message -> null);
-			fileWritingMessageHandler.setExpectReply(false);
-			return fileWritingMessageHandler;
-		}
-
 		@Bean
 		public IntegrationFlow fileFlow1() {
 			return IntegrationFlows.from("fileFlow1Input")
-					.handle(this.fileWritingMessageHandler())
+					.handle(File.outboundAdapter(tmpDir).fileNameGenerator(message -> null),
+							c -> c.id("fileWriting"))
 					.get();
 		}
 
@@ -1464,10 +1524,9 @@ public class IntegrationFlowTests {
 
 		@Bean
 		public IntegrationFlow tailFlow() {
-			ApacheCommonsFileTailingMessageProducer adapter = new ApacheCommonsFileTailingMessageProducer();
-			adapter.setFile(new File(tmpDir, "TailTest"));
-
-			return IntegrationFlows.from(adapter)
+			return IntegrationFlows.from(File.tailAdapter(new java.io.File(tmpDir, "TailTest"))
+					.delay(500)
+					.id("tailer"))
 					.transform("hello "::concat)
 					.channel(MessageChannels.queue("tailChannel"))
 					.get();
@@ -1540,6 +1599,29 @@ public class IntegrationFlowTests {
 			return MessageChannels.queue().get();
 		}
 
+
+		@Bean
+		public IntegrationFlow fileReadingFlow() {
+			return IntegrationFlows
+					.from(File.inboundAdapter(tmpDir).patternFilter("*.sitest"),
+							e -> e.poller(Pollers.fixedDelay(100)))
+					.transform(Transformers.fileToString())
+					.aggregate(a -> a.correlationExpression("1")
+							.releaseExpression("size() == 25"), null)
+					.channel(MessageChannels.queue("fileReadingResultChannel"))
+					.get();
+		}
+
+		@Bean
+		public IntegrationFlow fileWritingFlow() {
+			return IntegrationFlows.from("fileWritingInput")
+					.enrichHeaders(h -> h.header(FileHeaders.FILENAME, "foo.sitest")
+							.header("directory", new java.io.File(tmpDir, "fileWritingFlow")))
+					.handle(File.outboundGateway("headers[directory]"))
+					.channel(MessageChannels.queue("fileWritingResultChannel"))
+					.get();
+		}
+
 	}
 
 	private static class RoutingTestBean {