INT-2855 File Tailing Inbound Channel Adapter

Two implementations:

  - native 'tail' command (e.g. tail -F -n 0 /foo/bar)
  - Apache commons-io Tailer

Further changes:

 - Add documentation to section "what's new"
 - Add a JUnit @Rule so the OSDFTMP test doesn't fail on Windows

build.gradle

@@ -35,6 +35,7 @@ subprojects { subproject ->
 		aspectjVersion = '1.6.8'
 		cglibVersion = '2.2'
 		commonsNetVersion = '3.0.1'
+		commonsIoVersion = '2.4'
 		derbyVersion = '10.9.1.0'
 		easymockVersion = '2.3'
 		groovyVersion = '2.1.0'
@@ -203,6 +204,7 @@ project('spring-integration-file') {
 	dependencies {
 		compile project(":spring-integration-core")
 		compile "org.springframework:spring-context:$springVersion"
+		compile("commons-io:commons-io:$commonsIoVersion", optional)
 		testCompile project(":spring-integration-test")
 	}
 }

spring-integration-file/src/main/java/org/springframework/integration/file/config/FileNamespaceHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2013 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.
@@ -20,7 +20,7 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
 
 /**
  * Namespace handler for Spring Integration's 'file' namespace.
- * 
+ *
  * @author Iwein Fuld
  * @author Mark Fisher
  */
@@ -32,6 +32,7 @@ public class FileNamespaceHandler extends AbstractIntegrationNamespaceHandler {
 		registerBeanDefinitionParser("outbound-gateway", new FileOutboundGatewayParser());
 		registerBeanDefinitionParser("file-to-string-transformer", new FileToStringTransformerParser());
 		registerBeanDefinitionParser("file-to-bytes-transformer", new FileToByteArrayTransformerParser());
+		registerBeanDefinitionParser("tail-inbound-channel-adapter", new FileTailInboundChannelAdapterParser());
 	}
 
 }

spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java

@@ -0,0 +1,163 @@
+/*
+ * Copyright 2002-2013 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.file.config;
+
+import java.io.File;
+
+import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.beans.factory.config.AbstractFactoryBean;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.integration.MessageChannel;
+import org.springframework.integration.file.tail.ApacheCommonsFileTailingMessageProducer;
+import org.springframework.integration.file.tail.FileTailingMessageProducerSupport;
+import org.springframework.integration.file.tail.OSDelegatingFileTailingMessageProducer;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.util.StringUtils;
+
+/**
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBean<FileTailingMessageProducerSupport>
+		implements BeanNameAware {
+
+	private volatile String nativeOptions;
+
+	private volatile File file;
+
+	private volatile TaskExecutor taskExecutor;
+
+	private volatile TaskScheduler taskScheduler;
+
+	private volatile Long delay;
+
+	private volatile Long fileDelay;
+
+	private volatile Boolean end;
+
+	private volatile Boolean reopen;
+
+	private volatile FileTailingMessageProducerSupport adapter;
+
+	private volatile String beanName;
+
+	private volatile MessageChannel outputChannel;
+
+	private volatile Boolean autoStartup;
+
+	private volatile Integer phase;
+
+	public void setNativeOptions(String nativeOptions) {
+		this.nativeOptions = nativeOptions;
+	}
+
+	public void setFile(File file) {
+		this.file = file;
+	}
+
+	public void setTaskExecutor(TaskExecutor taskExecutor) {
+		this.taskExecutor = taskExecutor;
+	}
+
+	public void setTaskScheduler(TaskScheduler taskScheduler) {
+		this.taskScheduler = taskScheduler;
+	}
+
+	public void setDelay(long delay) {
+		this.delay = delay;
+	}
+
+	public void setFileDelay(long fileDelay) {
+		this.fileDelay = fileDelay;
+	}
+
+	public void setEnd(Boolean end) {
+		this.end = end;
+	}
+
+	public void setReopen(Boolean reopen) {
+		this.reopen = reopen;
+	}
+
+	@Override
+	public void setBeanName(String name) {
+		this.beanName = name;
+	}
+
+	public void setOutputChannel(MessageChannel outputChannel) {
+		this.outputChannel = outputChannel;
+	}
+
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	public void setPhase(int phase) {
+		this.phase = phase;
+	}
+
+	@Override
+	public Class<?> getObjectType() {
+		return this.adapter == null ? FileTailingMessageProducerSupport.class : this.adapter.getClass();
+	}
+
+	@Override
+	protected FileTailingMessageProducerSupport createInstance() throws Exception {
+		FileTailingMessageProducerSupport adapter;
+		if (this.delay == null && this.end == null && this.reopen == null) {
+			adapter = new OSDelegatingFileTailingMessageProducer();
+			if (this.nativeOptions != null) {
+				((OSDelegatingFileTailingMessageProducer) adapter).setOptions(this.nativeOptions);
+			}
+		}
+		else {
+			if (this.nativeOptions != null && StringUtils.hasText(this.nativeOptions) && logger.isWarnEnabled()) {
+				logger.warn("'native-options' are ignored with an Apache commons-io 'Tailer' adapter");
+			}
+			adapter = new ApacheCommonsFileTailingMessageProducer();
+			if (this.delay != null) {
+				((ApacheCommonsFileTailingMessageProducer) adapter).setPollingDelay(this.delay);
+			}
+			if (this.end != null) {
+				((ApacheCommonsFileTailingMessageProducer) adapter).setEnd(this.end);
+			}
+			if (this.reopen != null) {
+				((ApacheCommonsFileTailingMessageProducer) adapter).setReopen(this.reopen);
+			}
+		}
+		adapter.setFile(this.file);
+		adapter.setTaskExecutor(this.taskExecutor);
+		if (this.taskScheduler != null) {
+			adapter.setTaskScheduler(this.taskScheduler);
+		}
+		if (this.fileDelay != null) {
+			adapter.setTailAttemptsDelay(this.fileDelay);
+		}
+		adapter.setOutputChannel(outputChannel);
+		adapter.setBeanName(this.beanName);
+		if (this.autoStartup != null) {
+			adapter.setAutoStartup(this.autoStartup);
+		}
+		if (this.phase != null) {
+			adapter.setPhase(this.phase);
+		}
+		adapter.afterPropertiesSet();
+		this.adapter = adapter;
+		return adapter;
+	}
+
+}

spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParser.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2002-2013 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.file.config;
+
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.xml.AbstractChannelAdapterParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.util.StringUtils;
+
+/**
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+public class FileTailInboundChannelAdapterParser extends AbstractChannelAdapterParser {
+
+	@Override
+	protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.rootBeanDefinition(FileTailInboundChannelAdapterFactoryBean.class);
+
+		if (StringUtils.hasText(channelName)) {
+			builder.addPropertyReference("outputChannel", channelName);
+		}
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "native-options");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "file");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-scheduler");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delay");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "file-delay");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "end");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reopen");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase");
+
+		return builder.getBeanDefinition();
+	}
+
+}

spring-integration-file/src/main/java/org/springframework/integration/file/tail/ApacheCommonsFileTailingMessageProducer.java

@@ -0,0 +1,114 @@
+/*
+ * Copyright 2002-2013 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.file.tail;
+
+import org.apache.commons.io.input.Tailer;
+import org.apache.commons.io.input.TailerListener;
+
+/**
+ * File tailer that delegates to the Apache Commons Tailer.
+ *
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+public class ApacheCommonsFileTailingMessageProducer extends FileTailingMessageProducerSupport
+		implements TailerListener {
+
+	private volatile Tailer tailer;
+
+	private volatile long pollingDelay = 1000;
+
+	private volatile boolean end = true;
+
+	private volatile boolean reopen = false;
+
+	/**
+	 * The delay between checks of the file for new content in milliseconds.
+	 * @param pollingDelay The delay.
+	 */
+	public void setPollingDelay(long pollingDelay) {
+		this.pollingDelay = pollingDelay;
+	}
+
+	/**
+	 * If true, tail from the end of the file, otherwise
+	 * include all lines from the beginning. Default true.
+	 * @param end true or false
+	 */
+	public void setEnd(boolean end) {
+		this.end = end;
+	}
+
+	/**
+	 * If true, close and reopen the file between reading chunks;
+	 * default false.
+	 * @param reopen true or false.
+	 */
+	public void setReopen(boolean reopen) {
+		this.reopen = reopen;
+	}
+
+	@Override
+	public String getComponentType() {
+		return super.getComponentType() + " (Apache)";
+	}
+
+	@Override
+	protected void doStart() {
+		super.doStart();
+		Tailer tailer = new Tailer(this.getFile(), this, this.pollingDelay, this.end, this.reopen);
+		this.getTaskExecutor().execute(tailer);
+		this.tailer = tailer;
+	}
+
+	@Override
+	protected void doStop() {
+		super.doStop();
+		this.tailer.stop();
+	}
+
+	@Override
+	public void init(Tailer tailer) {
+	}
+
+	@Override
+	public void fileNotFound() {
+		this.publish("File not found:" + this.getFile().getAbsolutePath());
+		try {
+			Thread.sleep(this.getMissingFileDelay());
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+		}
+	}
+
+	@Override
+	public void fileRotated() {
+		this.publish("File rotated:" + this.getFile().getAbsolutePath());
+	}
+
+	@Override
+	public void handle(String line) {
+		send(line);
+	}
+
+	@Override
+	public void handle(Exception ex) {
+		this.publish(ex.getMessage());
+	}
+
+}

spring-integration-file/src/main/java/org/springframework/integration/file/tail/FileTailingMessageProducerSupport.java

@@ -0,0 +1,149 @@
+/*
+ * Copyright 2002-2013 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.file.tail;
+
+import java.io.File;
+
+import org.springframework.context.ApplicationEvent;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.integration.Message;
+import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.file.FileHeaders;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.util.Assert;
+
+/**
+ * Base class for file tailing inbound adapters.
+ *
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+public abstract class FileTailingMessageProducerSupport extends MessageProducerSupport
+		implements ApplicationEventPublisherAware {
+
+	private volatile File file;
+
+	private volatile ApplicationEventPublisher eventPublisher;
+
+	private volatile TaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
+
+	private volatile long tailAttemptsDelay = 5000;
+
+	@Override
+	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+		this.eventPublisher = applicationEventPublisher;
+	}
+
+	/**
+	 * The name of the file you wish to tail.
+	 * @param file The absolute path of the file.
+	 */
+	public void setFile(File file) {
+		Assert.notNull("'file' cannot be null");
+		this.file = file;
+	}
+
+	protected File getFile() {
+		if (this.file == null) {
+			throw new IllegalStateException("No 'file' has been provided");
+		}
+		return this.file;
+	}
+
+	/**
+	 * A task executor; default is a {@link SimpleAsyncTaskExecutor}.
+	 * @param taskExecutor
+	 */
+	public void setTaskExecutor(TaskExecutor taskExecutor) {
+		Assert.notNull("'taskExecutor' cannot be null");
+		this.taskExecutor = taskExecutor;
+	}
+
+	/**
+	 * The delay in milliseconds between attempts to tail a non-existent file,
+	 * or between attempts to execute a process if it fails for any reason.
+	 * @param missingFileDelay the delay.
+	 */
+	public void setTailAttemptsDelay(long tailAttemptsDelay) {
+		Assert.isTrue(tailAttemptsDelay > 0, "'tailAttemptsDelay' must be > 0");
+		this.tailAttemptsDelay = tailAttemptsDelay;
+	}
+
+	protected long getMissingFileDelay() {
+		return tailAttemptsDelay;
+	}
+
+	protected TaskExecutor getTaskExecutor() {
+		return this.taskExecutor;
+	}
+
+	@Override
+	public String getComponentType() {
+		return "file:tail-inbound-channel-adapter";
+	}
+
+	protected void send(String line) {
+		Message<?> message = MessageBuilder.withPayload(line)
+				.setHeader(FileHeaders.FILENAME, this.file.getAbsolutePath())
+				.build();
+		super.sendMessage(message);
+	}
+
+	protected void publish(String message) {
+		if (this.eventPublisher != null) {
+			FileTailingEvent event = new FileTailingEvent(this, message, this.file);
+			this.eventPublisher.publishEvent(event);
+		}
+		else {
+			logger.info("No publisher for event:" + message);
+		}
+	}
+
+	public static class FileTailingEvent extends ApplicationEvent {
+
+		private static final long serialVersionUID = -3382255736225946206L;
+
+		private final String message;
+
+		private final File file;
+
+		public FileTailingEvent(Object source, String message, File file) {
+			super(source);
+			this.message = message;
+			this.file = file;
+		}
+
+		protected String getMessage() {
+			return message;
+		}
+
+		public File getFile() {
+			return file;
+		}
+
+		@Override
+		public String toString() {
+			return "FileTailingEvent " + super.toString() +
+					" [message=" + this.message +
+					", file=" + this.file.getAbsolutePath() + "]";
+		}
+
+	}
+}

spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java

@@ -0,0 +1,253 @@
+/*
+ * Copyright 2002-2013 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.file.tail;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.Date;
+
+import org.springframework.integration.MessagingException;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.util.Assert;
+
+/**
+ * A file tailing message producer that delegates to the OS tail program.
+ * This is likely the most efficient mechanism on platforms that support it.
+ * Default options are "-F -n 0" (follow file name, no existing records).
+ *
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+public class OSDelegatingFileTailingMessageProducer extends FileTailingMessageProducerSupport
+		implements Runnable {
+
+	private volatile Process process;
+
+	private volatile String options = "-F -n 0";
+
+	private volatile String command = "ADAPTER_NOT_INITIALIZED";
+
+	private volatile BufferedReader reader;
+
+	private volatile TaskScheduler scheduler;
+
+	public void setOptions(String options) {
+		if (options == null) {
+			this.options = "";
+		}
+		else {
+			this.options = options;
+		}
+	}
+
+	@Override
+	public String getComponentType() {
+		return super.getComponentType() + " (native)";
+	}
+
+	@Override
+	protected void onInit() {
+		Assert.notNull(getFile(), "File cannot be null");
+		super.onInit();
+		this.command = "tail " + this.options + " " + this.getFile().getAbsolutePath();
+	}
+
+	@Override
+	protected void doStart() {
+		super.doStart();
+		destroyProcess();
+		this.getTaskExecutor().execute(new Runnable() {
+
+			@Override
+			public void run() {
+				runExec();
+			}
+		});
+	}
+
+	@Override
+	protected void doStop() {
+		super.doStop();
+		destroyProcess();
+	}
+
+	private void destroyProcess() {
+		Process process = this.process;
+		if (process != null) {
+			process.destroy();
+			this.process = null;
+		}
+	}
+
+	/**
+	 * Exec the native tail process.
+	 */
+	private void runExec() {
+		this.destroyProcess();
+		if (logger.isInfoEnabled()) {
+			logger.info("Starting tail process");
+		}
+		try {
+			Process process = Runtime.getRuntime().exec(this.command);
+			BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()));
+			this.process = process;
+			this.startProcessMonitor();
+			this.startStatusReader();
+			this.reader = reader;
+			this.getTaskExecutor().execute(this);
+		}
+		catch (IOException e) {
+			throw new MessagingException("Failed to exec tail command: '" + this.command + "'", e);
+		}
+	}
+
+	private TaskScheduler getRequiredTaskScheduler() {
+		if (this.scheduler == null) {
+			TaskScheduler taskScheduler = super.getTaskScheduler();
+			if (taskScheduler == null) {
+				ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
+				scheduler.initialize();
+				taskScheduler = scheduler;
+			}
+			this.scheduler = taskScheduler;
+		}
+		return this.scheduler;
+	}
+	/**
+	 * Runs a thread that waits for the Process result.
+	 */
+	private void startProcessMonitor() {
+		this.getTaskExecutor().execute(new Runnable() {
+
+			@Override
+			public void run() {
+				Process process = OSDelegatingFileTailingMessageProducer.this.process;
+				if (process == null) {
+					if (logger.isDebugEnabled()) {
+						logger.debug("Process destroyed before starting process monitor");
+					}
+					return;
+				}
+
+				int result = Integer.MIN_VALUE;
+				try {
+					if (logger.isDebugEnabled()) {
+						logger.debug("Monitoring process " + process);
+					}
+					result = process.waitFor();
+					if (logger.isInfoEnabled()) {
+						logger.info("tail process terminated with value " + result);
+					}
+				}
+				catch (InterruptedException e) {
+					Thread.currentThread().interrupt();
+					logger.error("Interrupted - stopping adapter", e);
+					stop();
+				}
+				finally {
+					destroyProcess();
+				}
+				if (isRunning()) {
+					if (logger.isInfoEnabled()) {
+						logger.info("Restarting tail process in " + getMissingFileDelay() + " milliseconds");
+					}
+					getRequiredTaskScheduler().schedule(new Runnable() {
+
+						@Override
+						public void run() {
+							runExec();
+						}
+					}, new Date(System.currentTimeMillis() + getMissingFileDelay()));
+				}
+			}
+		});
+	}
+
+	/**
+	 * Runs a thread that reads stderr - on some platforms status messages
+	 * (file not available, rotations etc) are sent to stderr.
+	 */
+	private void startStatusReader() {
+		Process process = this.process;
+		if (process == null) {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Process destroyed before starting stderr reader");
+			}
+			return;
+		}
+		final BufferedReader errorReader = new BufferedReader(new InputStreamReader(process.getErrorStream()));
+		this.getTaskExecutor().execute(new Runnable() {
+
+			@Override
+			public void run() {
+				String statusMessage;
+				if (logger.isDebugEnabled()) {
+					logger.debug("Reading stderr");
+				}
+				try {
+					while ((statusMessage = errorReader.readLine()) != null) {
+						publish(statusMessage);
+						if (logger.isTraceEnabled()) {
+							logger.trace(statusMessage);
+						}
+					}
+				}
+				catch (IOException e) {
+					logger.error("Exception on tail error reader", e);
+				}
+				finally {
+					try {
+						errorReader.close();
+					}
+					catch (IOException e) {
+						logger.error("Exception while closing stderr", e);
+					}
+				}
+			}
+		});
+	}
+
+	/**
+	 * Reads lines from stdout and sends in a message to the output channel.
+	 */
+	@Override
+	public void run() {
+		String line;
+		try {
+			if (logger.isDebugEnabled()) {
+				logger.debug("Reading stdout");
+			}
+			while ((line = this.reader.readLine()) != null) {
+				this.send(line);
+			}
+		}
+		catch (IOException e) {
+			logger.error("Exception on tail reader", e);
+			try {
+				this.reader.close();
+			}
+			catch (IOException e1) {
+
+			}
+			this.destroyProcess();
+		}
+	}
+
+
+}

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

@@ -0,0 +1,4 @@
+/**
+ * Classes used for tailing file system files.
+ */
+package org.springframework.integration.file.tail;

spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-3.0.xsd

@@ -193,6 +193,151 @@ Only files matching this regular expression will be picked up by this adapter.
         </xsd:complexType>
     </xsd:element>
 
+    <xsd:element name="tail-inbound-channel-adapter">
+        <xsd:annotation>
+            <xsd:documentation>
+                Configures an inbound channel adapter that 'tails' a file on the filesystem.
+            </xsd:documentation>
+        </xsd:annotation>
+        <xsd:complexType>
+            <xsd:attribute name="id" type="xsd:string">
+                <xsd:annotation>
+                    <xsd:documentation><![CDATA[Identifies the underlying Spring bean definition (xxxFileTailingMessageProducer)
+
+                        If no "channel" attribute is defined, then the "id" attribute is required.
+                        In that case the "id" attribute's value will be used as the channel name and ".adapter" will be
+                        appended to the "id" value of the underlying bean definition.
+                        ]]></xsd:documentation>
+                </xsd:annotation>
+            </xsd:attribute>
+            <xsd:attribute name="channel" type="xsd:string">
+                <xsd:annotation>
+                    <xsd:documentation><![CDATA[Defines the message channel to which the payload shall be forwarded to.
+
+                        Any Channel Adapter can be created without a "channel" reference in which case
+                        it will implicitly create an instance of DirectChannel. The created channel's name will match
+                        the "id" attribute of the <inbound-channel-adapter/> element. Therefore, if the "channel" attribute
+                        is not provided, then the "id" attribute is required.
+                        ]]></xsd:documentation>
+                    <xsd:appinfo>
+                        <tool:annotation kind="ref">
+                            <tool:expected-type type="org.springframework.integration.MessageChannel"/>
+                        </tool:annotation>
+                    </xsd:appinfo>
+                </xsd:annotation>
+            </xsd:attribute>
+            <xsd:attribute name="native-options" type="xsd:string">
+                <xsd:annotation>
+                    <xsd:documentation>
+                        Configures the adapter to exec 'tail' with these options (appended by the file name).
+                        Default: "-F -n 0" (follow the filename and emit no
+                        existing lines). This attribute is not allowed if 'delay' 'end' or 'reopen' is
+                        specified, which cause the Apache commons-io 'Tailer' class to be used instead of
+                        using a native 'tail' command.
+                    </xsd:documentation>
+                </xsd:annotation>
+            </xsd:attribute>
+            <xsd:attribute name="file" type="xsd:string">
+                <xsd:annotation>
+                    <xsd:documentation>
+                        The fully qualified name of the file to be tailed.
+                    </xsd:documentation>
+                </xsd:annotation>
+            </xsd:attribute>
+            <xsd:attribute name="task-executor" type="xsd:string">
+                <xsd:annotation>
+                    <xsd:documentation>
+                        A reference to a TaskExecutor; the default is a SimpleAsyncTaskExecutor; the
+                        native adapter uses three threads - one for reading stdout, one for
+                        reading stderr and one for monitoring the process.
+                    </xsd:documentation>
+                    <xsd:appinfo>
+                        <tool:annotation kind="ref">
+                            <tool:expected-type type="org.springframework.core.task.TaskExecutor"/>
+                        </tool:annotation>
+                    </xsd:appinfo>
+                </xsd:annotation>
+            </xsd:attribute>
+            <xsd:attribute name="task-scheduler" type="xsd:string">
+                <xsd:annotation>
+                    <xsd:documentation>
+                        A reference to a TaskScheduler; the default is the 'taskScheduler' bean
+                        which is automatically configured for all Spring Integration applications.
+                        The scheduler is used by the native adapter to reschedule
+                        the 'tail' process after a failure according to the 'file-delay'.
+                        This attribute is not allowed when using the Apache adapter.
+                    </xsd:documentation>
+                    <xsd:appinfo>
+                        <tool:annotation kind="ref">
+                            <tool:expected-type type="org.springframework.core.task.TaskExecutor"/>
+                        </tool:annotation>
+                    </xsd:appinfo>
+                </xsd:annotation>
+            </xsd:attribute>
+            <xsd:attribute name="file-delay" type="xsd:string">
+                <xsd:annotation>
+                    <xsd:documentation>
+                        The delay in milliseconds between attempts to open the file when no file was found (Apache adapter).
+                        For the native adapter, this is used as a delay before starting a new process after process failures.
+                        On some platforms, when the file doesn't exist, the 'tail' process is suspended until the file
+                        appears; on other platforms, the 'tail' process exits immediately if the file doesn't exist.
+                        Default 5000.
+                    </xsd:documentation>
+                </xsd:annotation>
+            </xsd:attribute>
+            <xsd:attribute name="delay" type="xsd:string">
+                <xsd:annotation>
+                    <xsd:documentation>
+                        Does not apply to the native adapter - the delay in milliseconds between polls when no new
+                        data was detected in the file. Default 1000. Note: Setting this option forces the use of the Apache
+                        Tailer implementation instead of the native 'tail' command.
+                    </xsd:documentation>
+                </xsd:annotation>
+            </xsd:attribute>
+            <xsd:attribute name="end">
+                <xsd:annotation>
+                    <xsd:documentation>
+                        Does not apply to the native adapter.
+                        Set to 'true' to tail from the end of the file, 'false' to tail from the beginning of the file.
+                        Default 'true'. Note: Setting this option forces the use of the Apache
+                        Tailer implementation instead of the native 'tail' command.
+                    </xsd:documentation>
+                </xsd:annotation>
+				<xsd:simpleType>
+					<xsd:union memberTypes="xsd:boolean xsd:string"/>
+				</xsd:simpleType>
+            </xsd:attribute>
+            <xsd:attribute name="reopen">
+                <xsd:annotation>
+                    <xsd:documentation>
+                        Does not apply to the native adapter.
+                        If 'true', close and reopen the file between reading chunks.
+                        Default 'false'. Note: Setting this option forces the use of the Apache
+                        Tailer implementation instead of the native 'tail' command.
+                    </xsd:documentation>
+                </xsd:annotation>
+				<xsd:simpleType>
+					<xsd:union memberTypes="xsd:boolean xsd:string"/>
+				</xsd:simpleType>
+            </xsd:attribute>
+            <xsd:attribute name="auto-startup" type="xsd:string" default="true">
+                <xsd:annotation>
+				    <xsd:documentation>
+				        Lifecycle attribute signaling if this component should be started during Application Context startup.
+				    </xsd:documentation>
+                </xsd:annotation>
+            </xsd:attribute>
+            <xsd:attribute name="phase" type="xsd:string">
+                <xsd:annotation>
+				    <xsd:documentation>
+				        Lifecycle attribute signaling the phase in which this component should be started during
+				        Application Context startup when 'auto-startup' is true.
+				    </xsd:documentation>
+                </xsd:annotation>
+            </xsd:attribute>
+        </xsd:complexType>
+    </xsd:element>
+
     <xsd:element name="outbound-gateway">
         <xsd:annotation>
             <xsd:documentation>

spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests-context.xml

@@ -0,0 +1,54 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int-file="http://www.springframework.org/schema/integration/file"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/file http://www.springframework.org/schema/integration/file/spring-integration-file.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+	<int-file:tail-inbound-channel-adapter id="default"
+		channel="input"
+		task-executor="exec"
+		file="/tmp/baz"
+		auto-startup="false"
+		phase="123" />
+
+	<int-file:tail-inbound-channel-adapter id="native"
+		channel="input"
+		native-options="-F -n 6"
+		task-executor="exec"
+		task-scheduler="sched"
+		file-delay="456"
+		file="/tmp/foo"
+		auto-startup="false"
+		phase="123" />
+
+	<int-file:tail-inbound-channel-adapter id="apacheDefault"
+		channel="input"
+		task-executor="exec"
+		file="/tmp/bar"
+		delay="2000"
+		file-delay="10000"
+		auto-startup="false"
+		phase="123" />
+	
+	<int-file:tail-inbound-channel-adapter id="apacheEndReopen"
+		channel="input"
+		task-executor="exec"
+		file="/tmp/qux"
+		delay="2000"
+		file-delay="10000"
+		end="false"
+		reopen="true"
+		auto-startup="false"
+		phase="123" />
+
+	<int:channel id="input" />
+	
+	<task:executor id="exec" />
+
+	<task:scheduler id="sched" />
+</beans>

spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests.java

@@ -0,0 +1,130 @@
+/*
+ * Copyright 2002-2013 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.file.config;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertSame;
+
+import java.io.File;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.integration.file.tail.ApacheCommonsFileTailingMessageProducer;
+import org.springframework.integration.file.tail.OSDelegatingFileTailingMessageProducer;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class FileTailInboundChannelAdapterParserTests {
+
+	@Autowired @Qualifier("default")
+	private OSDelegatingFileTailingMessageProducer defaultAdapter;
+
+	@Autowired @Qualifier("native")
+	private OSDelegatingFileTailingMessageProducer nativeAdapter;
+
+	@Autowired
+	private ApacheCommonsFileTailingMessageProducer apacheDefault;
+
+	@Autowired
+	private ApacheCommonsFileTailingMessageProducer apacheEndReopen;
+
+	@Autowired
+	private TaskExecutor exec;
+
+	@Autowired
+	private TaskScheduler sched;
+
+	@Autowired
+	private TaskScheduler taskScheduler;
+
+	@Test
+	public void testDefault() {
+		String fileName = TestUtils.getPropertyValue(defaultAdapter, "file", File.class).getAbsolutePath();
+		String normalizedName = getNormalizedPath(fileName);
+		assertEquals("/tmp/baz", normalizedName);
+		assertEquals("tail -F -n 0 " + fileName, TestUtils.getPropertyValue(defaultAdapter, "command"));
+		assertSame(exec, TestUtils.getPropertyValue(defaultAdapter, "taskExecutor"));
+		assertFalse(TestUtils.getPropertyValue(defaultAdapter, "autoStartup", Boolean.class));
+		assertEquals(123, TestUtils.getPropertyValue(defaultAdapter, "phase"));
+	}
+
+	@Test
+	public void testNative() {
+		String fileName = TestUtils.getPropertyValue(nativeAdapter, "file", File.class).getAbsolutePath();
+		String normalizedName = getNormalizedPath(fileName);
+		assertEquals("/tmp/foo", normalizedName);
+		assertEquals("tail -F -n 6 " + fileName, TestUtils.getPropertyValue(nativeAdapter, "command"));
+		assertSame(exec, TestUtils.getPropertyValue(nativeAdapter, "taskExecutor"));
+		assertSame(sched, TestUtils.getPropertyValue(nativeAdapter, "taskScheduler"));
+		assertFalse(TestUtils.getPropertyValue(nativeAdapter, "autoStartup", Boolean.class));
+		assertEquals(123, TestUtils.getPropertyValue(nativeAdapter, "phase"));
+		assertEquals(456L, TestUtils.getPropertyValue(nativeAdapter, "tailAttemptsDelay"));
+	}
+
+	@Test
+	public void testApacheDefault() {
+		String fileName = TestUtils.getPropertyValue(apacheDefault, "file", File.class).getAbsolutePath();
+		String normalizedName = getNormalizedPath(fileName);
+		assertEquals("/tmp/bar", normalizedName);
+		assertSame(exec, TestUtils.getPropertyValue(apacheDefault, "taskExecutor"));
+		assertEquals(2000L, TestUtils.getPropertyValue(apacheDefault, "pollingDelay"));
+		assertEquals(10000L, TestUtils.getPropertyValue(apacheDefault, "tailAttemptsDelay"));
+		assertFalse(TestUtils.getPropertyValue(apacheDefault, "autoStartup", Boolean.class));
+		assertEquals(123, TestUtils.getPropertyValue(apacheDefault, "phase"));
+		assertEquals(Boolean.TRUE, TestUtils.getPropertyValue(apacheDefault, "end"));
+		assertEquals(Boolean.FALSE, TestUtils.getPropertyValue(apacheDefault, "reopen"));
+	}
+
+	@Test
+	public void testApacheEndReopen() {
+		String fileName = TestUtils.getPropertyValue(apacheEndReopen, "file", File.class).getAbsolutePath();
+		String normalizedName = getNormalizedPath(fileName);
+		assertEquals("/tmp/qux", normalizedName);
+		assertSame(exec, TestUtils.getPropertyValue(apacheEndReopen, "taskExecutor"));
+		assertEquals(2000L, TestUtils.getPropertyValue(apacheEndReopen, "pollingDelay"));
+		assertEquals(10000L, TestUtils.getPropertyValue(apacheEndReopen, "tailAttemptsDelay"));
+		assertFalse(TestUtils.getPropertyValue(apacheEndReopen, "autoStartup", Boolean.class));
+		assertEquals(123, TestUtils.getPropertyValue(apacheEndReopen, "phase"));
+		assertEquals(Boolean.FALSE, TestUtils.getPropertyValue(apacheEndReopen, "end"));
+		assertEquals(Boolean.TRUE, TestUtils.getPropertyValue(apacheEndReopen, "reopen"));
+	}
+
+	/**
+	 * Fix up windows paths.
+	 */
+	private String getNormalizedPath(String fileName) {
+		String absolutePath = fileName.replaceAll("\\\\", "/");
+		int index = absolutePath.indexOf(":");
+		if (index >= 0) {
+			absolutePath = absolutePath.substring(index + 1);
+		}
+		return absolutePath;
+	}
+}

spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java

@@ -0,0 +1,156 @@
+/*
+ * Copyright 2002-2013 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.file.tail;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.integration.Message;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.file.tail.FileTailingMessageProducerSupport.FileTailingEvent;
+
+/**
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+public class FileTailingMessageProducerTests {
+
+	private static final String TAIL_OPTIONS_FOLLOW_NAME_MANY_LINES = "-F -n 99999999";
+
+	@Rule
+	public TailRule tailRule = new TailRule(TAIL_OPTIONS_FOLLOW_NAME_MANY_LINES);
+
+	private final Log logger = LogFactory.getLog(this.getClass());
+
+	private final String tmpDir = System.getProperty("java.io.tmpdir");
+
+	private File testDir;
+
+	private FileTailingMessageProducerSupport adapter;
+
+	@Before
+	public void setup() {
+		File f = new File(tmpDir, "FileTailingMessageProducerTests");
+		f.mkdir();
+		this.testDir = f;
+	}
+
+	@After
+	public void tearDown() {
+		if (this.adapter != null) {
+			adapter.stop();
+		}
+	}
+
+	@Test
+	@TailAvailable
+	public void testOS() throws Exception {
+		OSDelegatingFileTailingMessageProducer adapter = new OSDelegatingFileTailingMessageProducer();
+		adapter.setOptions(TAIL_OPTIONS_FOLLOW_NAME_MANY_LINES);
+		testGuts(adapter, "reader");
+	}
+
+	@Test
+	public void testApache() throws Exception {
+		ApacheCommonsFileTailingMessageProducer adapter = new ApacheCommonsFileTailingMessageProducer();
+		adapter.setPollingDelay(100);
+		adapter.setEnd(false);
+		testGuts(adapter, "tailer");
+	}
+
+	private void testGuts(FileTailingMessageProducerSupport adapter, String field)
+			throws Exception {
+		this.adapter = adapter;
+		final List<FileTailingEvent> events = new ArrayList<FileTailingEvent>();
+		adapter.setApplicationEventPublisher(new ApplicationEventPublisher() {
+			@Override
+			public void publishEvent(ApplicationEvent event) {
+				FileTailingEvent tailEvent = (FileTailingEvent) event;
+				logger.warn(event);
+				events.add(tailEvent);
+			}
+		});
+		adapter.setFile(new File(testDir, "foo"));
+		QueueChannel outputChannel = new QueueChannel();
+		adapter.setOutputChannel(outputChannel);
+		adapter.setTailAttemptsDelay(500);
+		adapter.afterPropertiesSet();
+		File file = new File(testDir, "foo");
+		File renamed = new File(testDir, "bar");
+		file.delete();
+		renamed.delete();
+		adapter.start();
+		waitForField(adapter, field);
+		FileOutputStream foo = new FileOutputStream(file);
+		for (int i = 0; i < 50; i++) {
+			foo.write(("hello" + i + "\n").getBytes());
+		}
+		foo.flush();
+		for (int i = 0; i < 50; i++) {
+			Message<?> message = outputChannel.receive(5000);
+			assertNotNull("expected a non-null message", message);
+			assertEquals("hello" + i, message.getPayload());
+		}
+		file.renameTo(renamed);
+		foo.close();
+		foo = new FileOutputStream(file);
+		if (adapter instanceof ApacheCommonsFileTailingMessageProducer) {
+			Thread.sleep(1000);
+		}
+		for (int i = 50; i < 100; i++) {
+			foo.write(("hello" + i + "\n").getBytes());
+		}
+		foo.flush();
+		for (int i = 50; i < 100; i++) {
+			Message<?> message = outputChannel.receive(3000);
+			assertNotNull("expected a non-null message", message);
+			assertEquals("hello" + i, message.getPayload());
+		}
+		foo.close();
+	}
+
+	private void waitForField(FileTailingMessageProducerSupport adapter, String field) throws Exception {
+		int n = 0;
+		DirectFieldAccessor accessor = new DirectFieldAccessor(adapter);
+		while (n < 100) {
+			if (accessor.getPropertyValue(field) == null) {
+				Thread.sleep(100);
+			}
+			else {
+				return;
+			}
+		}
+		fail("adapter failed to start");
+	}
+
+}

spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailAvailable.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2002-2013 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.file.tail;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.METHOD})
+public @interface TailAvailable {
+
+}

spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailRule.java

@@ -0,0 +1,140 @@
+/*
+ * Copyright 2002-2013 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.file.tail;
+
+import static org.junit.Assert.assertFalse;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.Test;
+import org.junit.rules.TestWatcher;
+import org.junit.runner.Description;
+import org.junit.runners.model.Statement;
+
+/**
+ * Ignores tests annotated with {@link TailAvailable} if 'tail' with the requested options
+ * does not work on this platform.
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+public class TailRule extends TestWatcher {
+
+	private static final Log logger = LogFactory.getLog(TailRule.class);
+
+	private static final String tmpDir = System.getProperty("java.io.tmpdir");
+
+	private final String commandToTest;
+
+	public TailRule(String optionsToTest) {
+		this.commandToTest = "tail " + optionsToTest + " ";
+	}
+
+	@Override
+	public Statement apply(Statement base, Description description) {
+		if (description.getAnnotation(TailAvailable.class) != null) {
+			if (!tailWorksOnThisMachine()) {
+				return new Statement() {
+
+					@Override
+					public void evaluate() throws Throwable {
+						// skip
+					}};
+			}
+		}
+		return super.apply(base, description);
+	}
+
+	private boolean tailWorksOnThisMachine() {
+		if (tmpDir.contains(":")) {
+			return false;
+		}
+		File testDir = new File(tmpDir, "FileTailingMessageProducerTests");
+		testDir.mkdir();
+		final File file = new File(testDir, "foo");
+		int result = -99;
+		try {
+			OutputStream fos = new FileOutputStream(file);
+			fos.write("foo".getBytes());
+			fos.close();
+			final AtomicReference<Integer> c = new AtomicReference<Integer>();
+			final CountDownLatch latch = new CountDownLatch(1);
+			Future<Process> future = Executors.newSingleThreadExecutor().submit(new Callable<Process>() {
+
+				@Override
+				public Process call() throws Exception {
+					final Process process = Runtime.getRuntime().exec(commandToTest + " " + file.getAbsolutePath());
+					Executors.newSingleThreadExecutor().execute(new Runnable() {
+
+						@Override
+						public void run() {
+							try {
+								c.set(process.getInputStream().read());
+								latch.countDown();
+							}
+							catch (IOException e) {
+								logger.error("Error reading test stream", e);
+							}
+						}
+					});
+					return process;
+				}
+			});
+			try {
+				Process process = future.get(10, TimeUnit.SECONDS);
+				if (latch.await(10, TimeUnit.SECONDS)) {
+					Integer read = c.get();
+					if (read != null && read == 'f') {
+						result = 0;
+					}
+				}
+				process.destroy();
+			}
+			catch (ExecutionException e) {
+				result = -999;
+			}
+			file.delete();
+		}
+		catch (Exception e) {
+			logger.error("failed to test tail", e);
+		}
+		if (result != 0) {
+			logger.warn("tail command is not available on this platform; result:" + result);
+		}
+		return result == 0;
+	}
+
+	public static class TestRule {
+
+		@Test
+		public void test1() {
+			TailRule rule = new TailRule("-BLAH");
+			assertFalse(rule.tailWorksOnThisMachine());
+		}
+	}
+}

src/reference/docbook/file.xml

@@ -149,6 +149,78 @@
         <para>
             This gives you full freedom to choose the ordering, listing and locking strategies.
         </para>
+        <section id="file-tailing">
+          <title>'Tail'ing Files</title>
+          <para>
+            Another popular use case is to get 'lines' from the end (or tail) of a file. Two implementations are provided;
+            the first, <classname>OSDelegatingFileTailingMessageProducer</classname>, uses the native <code>tail</code>
+            command (on operating systems that have one). This is likely the most efficient implementation on those
+            platforms. For operating systems that do not have a <code>tail</code> command, the second implementation
+            <classname>ApacheCommonsFileTailingMessageProducer</classname> which uses the Apache <code>commons-io
+            Tailer</code> class.
+          </para>
+          <para>
+            In both cases, file system events, such as files being unavailable etc, are published as
+            <interfacename>ApplicationEvent</interfacename>s using the normal Spring event publishing mechanism.
+            Examples of such events are:
+          </para>
+          <para><code>
+          [message=tail: cannot open `/tmp/foo' for reading:
+               No such file or directory, file=/tmp/foo]
+          </code></para>
+          <para><code>
+          [message=tail: `/tmp/foo' has become accessible, file=/tmp/foo]
+          </code></para>
+          <para><code>
+          [message=tail: `/tmp/foo' has become inaccessible:
+               No such file or directory, file=/tmp/foo]
+           </code></para>
+           <para><code>
+          [message=tail: `/tmp/foo' has appeared;
+               following end of new file, file=/tmp/foo]
+          </code></para>
+          <para>
+            This sequence of events might occur, for example, when a file is rotated.
+          </para>
+          <note>
+            Not all platforms supporting a <code>tail</code> command provide these status messages.
+          </note>
+          <para>
+            Example configurations:
+          </para>
+          <programlisting language="xml"><![CDATA[<int-file:tail-inbound-channel-adapter id="native"
+	channel="input"
+	task-executor="exec"
+	file="/tmp/foo"/>]]></programlisting>
+          <para>
+            This creates a native adapter with default '-F -n 0' options (follow the file name from the current end).
+          </para>
+          <programlisting language="xml"><![CDATA[<int-file:tail-inbound-channel-adapter id="native"
+	channel="input"
+	native-options="-F -n 6"
+	task-executor="exec"
+	file-delay=10000
+	file="/tmp/foo"/>]]></programlisting>
+          <para>
+            This creates a native adapter with '-F -n 6' options (follow the file name, emit up to 6 lines before the current end).
+            If the tail command fails (on some platforms, a missing file causes the <code>tail</code> to fail, even with
+            <code>-F</code> specified), the command will be retried every 10 seconds.
+          </para>
+          <programlisting language="xml"><![CDATA[<int-file:tail-inbound-channel-adapter id="apache"
+	channel="input"
+	task-executor="exec"
+	file="/tmp/bar"
+	delay="2000"
+	end="false"
+	reopen="true"
+	file-delay="10000"/>]]></programlisting>
+          <para>
+            This creates a commons-io <classname>Tailer</classname> adapter that examines the file for new lines every
+            2 seconds, and checks for existence of a missing file every 10 seconds. The file will be tailed from the
+            beginning (<code>end="false"</code>) instead of the end (which is the default). The file will be
+            reopened for each chunk (the default is to keep the file open).
+          </para>
+        </section>
   </section>
 	<section id="file-writing">
 		<title>Writing files</title>

src/reference/docbook/whats-new.xml

@@ -60,6 +60,14 @@
 				<xref linkend="syslog"/>.
 			</para>
 		</section>
+		<section id="3.0-tail">
+			<title>'Tail' Support</title>
+			<para>
+				File 'tail'ing inbound channel adapters are now provided to generate messages when
+				lines are added to the end of text files.
+				<xref linkend="file-tailing"/>.
+			</para>
+		</section>
 	</section>
 
 	<section id="3.0-general">