INT-4188: Add Idle Event Interval Support

JIRA: https://jira.springsource.org/browse/INT-4188

* add `<idle-event-interval>` XSD element
* add `FileTailingIdleEvent`
* move `TaskScheduler` and `getRequiredTaskScheduler` to `FileTailingMessageProducerSupport`
* add `setIdleEventInterval`

use taskExecutor instead of creating one
stop the scheduled task in case of `doStop()`
other required changes

update Test and Reference

implement changes required

What do you think about this checking if file exist before sending event.

Mock is used for `file.exist()` plus other minor updates

Polishing
* Some typos in the code as well as in the docs
* Fix `FileTailingMessageProducerTests.testIdleEvent()` to verify that `FileTailingIdleEvent` isn't emitted when file does not exist

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

@@ -46,6 +46,8 @@ public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBea
 
 	private volatile boolean enableStatusReader = true;
 
+	private volatile Long idleEventInterval;
+
 	private volatile File file;
 
 	private volatile TaskExecutor taskExecutor;
@@ -90,6 +92,15 @@ public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBea
 		this.enableStatusReader = enableStatusReader;
 	}
 
+	/**
+	 * How often to emit {@link FileTailingMessageProducerSupport.FileTailingIdleEvent}s in milliseconds.
+	 * @param idleEventInterval the interval.
+	 * @since 5.0
+	 */
+	public void setIdleEventInterval(long idleEventInterval) {
+		this.idleEventInterval = idleEventInterval;
+	}
+
 	public void setFile(File file) {
 		this.file = file;
 	}
@@ -222,6 +233,9 @@ public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBea
 		if (this.fileDelay != null) {
 			adapter.setTailAttemptsDelay(this.fileDelay);
 		}
+		if (this.idleEventInterval != null) {
+			adapter.setIdleEventInterval(this.idleEventInterval);
+		}
 		adapter.setOutputChannel(this.outputChannel);
 		adapter.setErrorChannel(this.errorChannel);
 		adapter.setBeanName(this.beanName);

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

@@ -43,6 +43,7 @@ public class FileTailInboundChannelAdapterParser extends AbstractChannelAdapterP
 		}
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "native-options");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "enable-status-reader");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "idle-event-interval");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "file");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-scheduler");

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

@@ -65,6 +65,27 @@ public class TailAdapterSpec extends MessageProducerSpec<TailAdapterSpec, FileTa
 		return _this();
 	}
 
+
+	/**
+	 * This field control the stderr events.
+	 * @param enableStatusReader boolean to enable or disable events from stderr.
+	 * @return the spec
+	 */
+	public TailAdapterSpec enableStatusReader(boolean enableStatusReader) {
+		this.factoryBean.setEnableStatusReader(enableStatusReader);
+		return _this();
+	}
+
+	/**
+	 * Specify the idle interval before start sending idle events.
+	 * @param idleEventInterval interval in ms for the event idle time.
+	 * @return the spec.
+	 */
+	public TailAdapterSpec idleEventInterval(long idleEventInterval)  {
+		this.factoryBean.setIdleEventInterval(idleEventInterval);
+		return _this();
+	}
+
 	/**
 	 * Configure a task executor. Defaults to a
 	 * {@link org.springframework.core.task.SimpleAsyncTaskExecutor}.

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

@@ -88,7 +88,7 @@ public class ApacheCommonsFileTailingMessageProducer extends FileTailingMessageP
 
 	@Override
 	public void fileNotFound() {
-		this.publish("File not found:" + this.getFile().getAbsolutePath());
+		this.publish("File not found: " + this.getFile().getAbsolutePath());
 		try {
 			Thread.sleep(this.getMissingFileDelay());
 		}
@@ -99,7 +99,7 @@ public class ApacheCommonsFileTailingMessageProducer extends FileTailingMessageP
 
 	@Override
 	public void fileRotated() {
-		this.publish("File rotated:" + this.getFile().getAbsolutePath());
+		this.publish("File rotated: " + this.getFile().getAbsolutePath());
 	}
 
 	@Override

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

@@ -17,6 +17,8 @@
 package org.springframework.integration.file.tail;
 
 import java.io.File;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.atomic.AtomicLong;
 
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
@@ -33,6 +35,7 @@ import org.springframework.util.Assert;
  *
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Ali Shahbour
  * @since 3.0
  *
  */
@@ -47,6 +50,14 @@ public abstract class FileTailingMessageProducerSupport extends MessageProducerS
 
 	private volatile long tailAttemptsDelay = 5000;
 
+	private final AtomicLong lastNoMessageAlert = new AtomicLong();
+
+	private long idleEventInterval = 0;
+
+	private volatile long lastProduce = System.currentTimeMillis();
+
+	private ScheduledFuture<?> idleEventScheduledFuture;
+
 	@Override
 	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
 		this.eventPublisher = applicationEventPublisher;
@@ -87,6 +98,16 @@ public abstract class FileTailingMessageProducerSupport extends MessageProducerS
 		this.tailAttemptsDelay = tailAttemptsDelay;
 	}
 
+	/**
+	 * How often to emit {@link FileTailingIdleEvent}s in milliseconds.
+	 * @param idleEventInterval the interval.
+	 * @since 5.0
+	 */
+	public void setIdleEventInterval(long idleEventInterval) {
+		Assert.isTrue(idleEventInterval > 0, "'idleEventInterval' must be > 0");
+		this.idleEventInterval = idleEventInterval;
+	}
+
 	protected long getMissingFileDelay() {
 		return this.tailAttemptsDelay;
 	}
@@ -106,6 +127,7 @@ public abstract class FileTailingMessageProducerSupport extends MessageProducerS
 				.setHeader(FileHeaders.ORIGINAL_FILE, this.file)
 				.build();
 		super.sendMessage(message);
+		updateLastProduce();
 	}
 
 	protected void publish(String message) {
@@ -114,10 +136,74 @@ public abstract class FileTailingMessageProducerSupport extends MessageProducerS
 			this.eventPublisher.publishEvent(event);
 		}
 		else {
-			logger.info("No publisher for event:" + message);
+			logger.info("No publisher for event: " + message);
 		}
 	}
 
+
+
+	@Override
+	protected void doStart() {
+		super.doStart();
+		if (this.idleEventInterval > 0) {
+			this.idleEventScheduledFuture = getTaskScheduler().scheduleWithFixedDelay(() -> {
+				long now = System.currentTimeMillis();
+				long lastAlertAt = this.lastNoMessageAlert.get();
+				long lastProduce = this.lastProduce;
+				if (now > lastProduce + this.idleEventInterval
+						&& now > lastAlertAt + this.idleEventInterval
+						&& this.lastNoMessageAlert.compareAndSet(lastAlertAt, now)) {
+					publishIdleEvent(now - lastProduce);
+				}
+			}, this.idleEventInterval);
+		}
+	}
+
+	@Override
+	protected void doStop() {
+		super.doStop();
+		if (this.idleEventScheduledFuture != null) {
+			this.idleEventScheduledFuture.cancel(true);
+		}
+	}
+
+	private void publishIdleEvent(long idleTime) {
+		if (this.eventPublisher != null) {
+			if (getFile().exists()) {
+				FileTailingIdleEvent event = new FileTailingIdleEvent(this, this.file, idleTime);
+				this.eventPublisher.publishEvent(event);
+			}
+		}
+		else {
+			logger.info("No publisher for idle event");
+		}
+	}
+
+	private void updateLastProduce() {
+		if (this.idleEventInterval > 0) {
+			this.lastProduce = System.currentTimeMillis();
+		}
+	}
+
+	public static class FileTailingIdleEvent extends FileTailingEvent {
+
+		private static final long serialVersionUID = -967118535347976767L;
+
+		private final long idleTime;
+
+		public FileTailingIdleEvent(Object source, File file, long idleTime) {
+			super(source, "Idle timeout", file);
+			this.idleTime = idleTime;
+		}
+
+		@Override
+		public String toString() {
+			return super.toString() +
+					" [idle time=" + this.idleTime + "]";
+		}
+
+	}
+
 	public static class FileTailingEvent extends FileIntegrationEvent {
 
 		private static final long serialVersionUID = -3382255736225946206L;
@@ -148,4 +234,5 @@ public abstract class FileTailingMessageProducerSupport extends MessageProducerS
 		}
 
 	}
+
 }

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

@@ -23,8 +23,6 @@ import java.util.Date;
 
 import org.springframework.messaging.MessagingException;
 import org.springframework.scheduling.SchedulingAwareRunnable;
-import org.springframework.scheduling.TaskScheduler;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.util.Assert;
 
 /**
@@ -51,8 +49,6 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr
 
 	private volatile BufferedReader reader;
 
-	private volatile TaskScheduler scheduler;
-
 	public void setOptions(String options) {
 		if (options == null) {
 			this.options = "";
@@ -138,18 +134,7 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr
 		}
 	}
 
-	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.
 	 */
@@ -185,7 +170,7 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr
 				if (logger.isInfoEnabled()) {
 					logger.info("Restarting tail process in " + getMissingFileDelay() + " milliseconds");
 				}
-				getRequiredTaskScheduler()
+				getTaskScheduler()
 						.schedule(this::runExec, new Date(System.currentTimeMillis() + getMissingFileDelay()));
 			}
 		});

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

@@ -272,8 +272,8 @@ Only files matching this regular expression will be picked up by this adapter.
                         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.
+                        the 'tail' process after a failure according to the 'file-delay', and also
+                        it is used to emit idle event.
                     </xsd:documentation>
                     <xsd:appinfo>
                         <tool:annotation kind="ref">
@@ -302,6 +302,13 @@ Only files matching this regular expression will be picked up by this adapter.
                     </xsd:documentation>
                 </xsd:annotation>
             </xsd:attribute>
+            <xsd:attribute name="idle-event-interval" type="xsd:string">
+                <xsd:annotation>
+                    <xsd:documentation>
+                        The delay in milliseconds between idle events when no new lines are being tailed.
+                    </xsd:documentation>
+                </xsd:annotation>
+            </xsd:attribute>
             <xsd:attribute name="end">
                 <xsd:annotation>
                     <xsd:documentation>

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

@@ -44,6 +44,7 @@
 		file="/tmp/bar"
 		delay="${foo}"
 		file-delay="10000"
+		idle-event-interval="10000"
 		auto-startup="false"
 		phase="123" />
 

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

@@ -112,6 +112,7 @@ public class FileTailInboundChannelAdapterParserTests {
 		assertSame(exec, TestUtils.getPropertyValue(apacheDefault, "taskExecutor"));
 		assertEquals(2000L, TestUtils.getPropertyValue(apacheDefault, "pollingDelay"));
 		assertEquals(10000L, TestUtils.getPropertyValue(apacheDefault, "tailAttemptsDelay"));
+		assertEquals(10000L, TestUtils.getPropertyValue(apacheDefault, "idleEventInterval"));
 		assertFalse(TestUtils.getPropertyValue(apacheDefault, "autoStartup", Boolean.class));
 		assertEquals(123, TestUtils.getPropertyValue(apacheDefault, "phase"));
 		assertEquals(Boolean.TRUE, TestUtils.getPropertyValue(apacheDefault, "end"));

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

@@ -16,16 +16,25 @@
 
 package org.springframework.integration.file.tail;
 
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
 
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -36,17 +45,21 @@ import org.junit.Test;
 
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.tail.FileTailingMessageProducerSupport.FileTailingEvent;
+import org.springframework.integration.file.tail.FileTailingMessageProducerSupport.FileTailingIdleEvent;
 import org.springframework.messaging.Message;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 
 /**
  * @author Gary Russell
  * @author Gavin Gray
  * @author Artem Bilan
- * @since 3.0
+ * @author Ali Shahbour
  *
+ * @since 3.0
  */
 public class FileTailingMessageProducerTests {
 
@@ -124,13 +137,61 @@ public class FileTailingMessageProducerTests {
 		adapter.stop();
 	}
 
+	@Test
+	public void testIdleEvent() throws Exception {
+		ApacheCommonsFileTailingMessageProducer adapter = new ApacheCommonsFileTailingMessageProducer();
+
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.afterPropertiesSet();
+		adapter.setTaskScheduler(taskScheduler);
+
+		CountDownLatch idleCountDownLatch = new CountDownLatch(1);
+		CountDownLatch fileExistCountDownLatch = new CountDownLatch(1);
+
+		adapter.setApplicationEventPublisher(event -> {
+			if (event instanceof FileTailingIdleEvent) {
+				idleCountDownLatch.countDown();
+			}
+			if (event instanceof FileTailingEvent) {
+				FileTailingEvent fileTailingEvent = (FileTailingEvent) event;
+				if (fileTailingEvent.getMessage().contains("File not found")) {
+					fileExistCountDownLatch.countDown();
+				}
+			}
+		});
+
+		File file = spy(new File(this.testDir, "foo"));
+		file.delete();
+		adapter.setFile(file);
+
+		adapter.setOutputChannel(new NullChannel());
+		adapter.setIdleEventInterval(10);
+		adapter.afterPropertiesSet();
+		adapter.start();
+
+		boolean noFile = fileExistCountDownLatch.await(10, TimeUnit.SECONDS);
+		assertTrue("file does not exist event did not emit ", noFile);
+		boolean noEvent = idleCountDownLatch.await(100, TimeUnit.MILLISECONDS);
+		assertFalse("event should not emit when no file exit", noEvent);
+		verify(file, atLeastOnce()).exists();
+
+		file.createNewFile();
+		boolean eventRaised = idleCountDownLatch.await(10, TimeUnit.SECONDS);
+		assertTrue("idle event did not emit", eventRaised);
+		adapter.stop();
+		file.delete();
+	}
+
 	private void testGuts(FileTailingMessageProducerSupport adapter, String field)
 			throws Exception {
 		this.adapter = adapter;
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.afterPropertiesSet();
+		adapter.setTaskScheduler(taskScheduler);
 		final List<FileTailingEvent> events = new ArrayList<FileTailingEvent>();
 		adapter.setApplicationEventPublisher(event -> {
 			FileTailingEvent tailEvent = (FileTailingEvent) event;
-			logger.warn(event);
+			logger.debug(event);
 			events.add(tailEvent);
 		});
 		adapter.setFile(new File(testDir, "foo"));
@@ -174,6 +235,8 @@ public class FileTailingMessageProducerTests {
 			assertEquals(file, message.getHeaders().get(FileHeaders.ORIGINAL_FILE));
 			assertEquals(file.getName(), message.getHeaders().get(FileHeaders.FILENAME));
 		}
+
+		assertThat(events.size(), greaterThanOrEqualTo(1));
 	}
 
 	private void waitForField(FileTailingMessageProducerSupport adapter, String field) throws Exception {

src/reference/asciidoc/file.adoc

@@ -411,6 +411,10 @@ Examples of such events are:
 
 This sequence of events might occur, for example, when a file is rotated.
 
+Starting with _version 5.0_, a `FileTailingIdleEvent` is emitted when there is no data in the file during `idleEventInterval`.
+
+`[message=Idle timeout, file=/tmp/foo] [idle time=5438]`
+
 NOTE: Not all platforms supporting a `tail` command provide these status messages.
 
 Messages emitted from these endpoints have the following headers:
@@ -418,7 +422,7 @@ Messages emitted from these endpoints have the following headers:
 - `FileHeaders.ORIGINAL_FILE` - the `File` object
 - `FileHeaders.FILENAME` - the file name (`File.getName()`)
 
-NOTE: In versions prior to _5.0_, the `FileHeaders.FILENAME` header contained a string representation of the file's absolute path.
+NOTE: In versions prior to _version 5.0_, the `FileHeaders.FILENAME` header contained a string representation of the file's absolute path.
 You can now obtain that by calling `getAbsolutePath()` on the original file header.
 
 Example configurations:
@@ -458,6 +462,18 @@ If the tail command fails (on some platforms, a missing file causes the `tail` t
 By default native adapter capture from standard output and send them as messages and from standard error to raise events.
 Starting with _version 4.3.6_, you can discard the standard error events by setting the `enable-status-reader` to `false`.
 
+[source,xml]
+----
+<int-file:tail-inbound-channel-adapter id="native"
+	channel="input"
+	idle-event-interval="5000"
+	task-executor="exec"
+	file="/tmp/foo"/>
+----
+
+`IdleEventInterval` is set to 5000 then, if no lines are written for 5 second, `FileTailingIdleEvent` will be triggered every 5 second.
+This can be useful if we need to stop the adapter.
+
 [source,xml]
 ----
 <int-file:tail-inbound-channel-adapter id="apache"

src/reference/asciidoc/whats-new.adoc

@@ -55,6 +55,9 @@ See <<feed>> for more information.
 The new `FileHeaders.RELATIVE_PATH` Message header has been introduced to represent relative path in the `FileReadingMessageSource`.
 See <<file-reading>> for more information.
 
+The tail adapter now supports `idleEventInterval` to emit events when there is no data in the file during that period.
+See <<file-tailing>> for more information.
+
 ==== (S)FTP Changes
 
 The inbound channel adapters now have a property `max-fetch-size` which is used to limit the number of files fetched during a poll when there are no files currently in the local directory.