From f7b1ec88851219d666d4c6046c29661bc5717682 Mon Sep 17 00:00:00 2001 From: Ali Shahbour Date: Thu, 22 Dec 2016 13:23:52 +0200 Subject: [PATCH] INT-4188: Add Idle Event Interval Support JIRA: https://jira.springsource.org/browse/INT-4188 * add `` 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 --- ...eTailInboundChannelAdapterFactoryBean.java | 14 +++ .../FileTailInboundChannelAdapterParser.java | 1 + .../integration/file/dsl/TailAdapterSpec.java | 21 +++++ ...acheCommonsFileTailingMessageProducer.java | 4 +- .../FileTailingMessageProducerSupport.java | 89 ++++++++++++++++++- ...SDelegatingFileTailingMessageProducer.java | 19 +--- .../config/spring-integration-file-5.0.xsd | 11 ++- ...boundChannelAdapterParserTests-context.xml | 1 + ...eTailInboundChannelAdapterParserTests.java | 1 + .../tail/FileTailingMessageProducerTests.java | 67 +++++++++++++- src/reference/asciidoc/file.adoc | 18 +++- src/reference/asciidoc/whats-new.adoc | 3 + 12 files changed, 224 insertions(+), 25 deletions(-) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java index a46873b996..106bbb5325 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java +++ b/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); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParser.java index 4d0b93f332..c561055538 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParser.java +++ b/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"); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/TailAdapterSpec.java b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/TailAdapterSpec.java index bcc893e11d..f5926abf65 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/TailAdapterSpec.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/TailAdapterSpec.java @@ -65,6 +65,27 @@ public class TailAdapterSpec extends MessageProducerSpec 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 } } + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java index 1c26e18c7c..6ca5eadfb9 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java +++ b/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())); } }); diff --git a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.0.xsd b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.0.xsd index 15033ab4b4..777f183660 100644 --- a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-5.0.xsd +++ b/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. @@ -302,6 +302,13 @@ Only files matching this regular expression will be picked up by this adapter. + + + + The delay in milliseconds between idle events when no new lines are being tailed. + + + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests-context.xml index acce911261..7222f04e4c 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests-context.xml +++ b/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" /> diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests.java index 9d044b8982..7d3e42d1e1 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterParserTests.java +++ b/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")); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java index bee29e7846..c6c4f27470 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java +++ b/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 events = new ArrayList(); 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 { diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index 2202b66bc2..402a5bd9da 100644 --- a/src/reference/asciidoc/file.adoc +++ b/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] +---- + +---- + +`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] ---- > for more information. The new `FileHeaders.RELATIVE_PATH` Message header has been introduced to represent relative path in the `FileReadingMessageSource`. See <> for more information. +The tail adapter now supports `idleEventInterval` to emit events when there is no data in the file during that period. +See <> 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.