diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java index b09359067d..0f17728f22 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java @@ -203,7 +203,7 @@ public class AsyncAmqpGatewayTests { Message returned = returnChannel.receive(10000); assertNotNull(returned); assertEquals("fiz", returned.getPayload()); - + ackChannel.receive(10000); ackChannel.purge(null); // Simulate a nack - it's hard to get Rabbit to generate one diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java index d8f7736319..a16ee29a9c 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java @@ -17,21 +17,39 @@ package org.springframework.integration.file; import java.io.File; +import java.io.IOException; +import java.nio.file.FileSystems; +import java.nio.file.FileVisitResult; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.SimpleFileVisitor; +import java.nio.file.StandardWatchEventKinds; +import java.nio.file.WatchEvent; +import java.nio.file.WatchKey; +import java.nio.file.WatchService; +import java.nio.file.attribute.BasicFileAttributes; +import java.util.Arrays; +import java.util.Collection; import java.util.Comparator; import java.util.LinkedHashSet; import java.util.List; import java.util.Queue; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.PriorityBlockingQueue; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.context.Lifecycle; import org.springframework.integration.aggregator.ResequencingMessageGroupProcessor; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.core.MessageSource; import org.springframework.integration.file.filters.AcceptOnceFileListFilter; import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.filters.ResettableFileListFilter; +import org.springframework.lang.UsesJava7; import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; @@ -64,7 +82,7 @@ import org.springframework.util.Assert; * @author Gary Russell * @author Artem Bilan */ -public class FileReadingMessageSource extends IntegrationObjectSupport implements MessageSource { +public class FileReadingMessageSource extends IntegrationObjectSupport implements MessageSource, Lifecycle { private static final int DEFAULT_INTERNAL_QUEUE_CAPACITY = 5; @@ -87,10 +105,16 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement private volatile boolean scanEachPoll = false; + private volatile boolean running; + private FileListFilter filter; private FileLocker locker; + private boolean useWatchService; + + private WatchEventType[] watchEvents = new WatchEventType[] { WatchEventType.CREATE }; + /** * Creates a FileReadingMessageSource with a naturally ordered queue of unbounded capacity. */ @@ -236,11 +260,59 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement this.scanEachPoll = scanEachPoll; } + /** + * Switch this {@link FileReadingMessageSource} to use its internal + * {@link FileReadingMessageSource.WatchServiceDirectoryScanner}. + * @param useWatchService the {@code boolean} flag to switch to + * {@link FileReadingMessageSource.WatchServiceDirectoryScanner} on {@code true}. + * @since 4.3 + * @see #setWatchEvents + */ + public void setUseWatchService(boolean useWatchService) { + this.useWatchService = useWatchService; + } + + /** + * The {@link WatchService} event types. + * If {@link #setUseWatchService} isn't {@code true}, this option is ignored. + * @param watchEvents the set of {@link WatchEventType}. + * @since 4.3 + * @see #setUseWatchService + */ + public void setWatchEvents(WatchEventType... watchEvents) { + Assert.notEmpty(watchEvents, "'watchEvents' must not be empty."); + Assert.noNullElements(watchEvents, "'watchEvents' must not contain null elements."); + Assert.state(!this.running, "Cannot change watch events while running."); + + this.watchEvents = Arrays.copyOf(watchEvents, watchEvents.length); + } + @Override public String getComponentType() { return "file:inbound-channel-adapter"; } + @Override + public void start() { + if (this.scanner instanceof Lifecycle) { + ((Lifecycle) this.scanner).start(); + } + this.running = true; + } + + @Override + public void stop() { + if (this.scanner instanceof Lifecycle) { + ((Lifecycle) this.scanner).start(); + } + this.running = false; + } + + @Override + public boolean isRunning() { + return this.running; + } + @Override protected void onInit() { Assert.notNull(this.directory, "'directory' must not be null"); @@ -253,6 +325,14 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement "Source path [" + this.directory + "] does not point to a directory."); Assert.isTrue(this.directory.canRead(), "Source directory [" + this.directory + "] is not readable."); + + Assert.state(!(this.scannerExplicitlySet && this.useWatchService), + "The 'scanner' and 'useWatchService' options are mutually exclusive: " + this.scanner); + + if (this.useWatchService) { + this.scanner = new WatchServiceDirectoryScanner(); + } + Assert.state(!(this.scannerExplicitlySet && (this.filter != null || this.locker != null)), "The 'filter' and 'locker' options must be present on the provided external 'scanner': " + this.scanner); @@ -262,6 +342,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement if (this.locker != null) { this.scanner.setLocker(this.locker); } + } public Message receive() throws MessagingException { @@ -302,9 +383,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement /** * Adds the failed message back to the 'toBeReceived' queue if there is room. - * - * @param failedMessage - * the {@link org.springframework.messaging.Message} that failed + * @param failedMessage the {@link Message} that failed */ public void onFailure(Message failedMessage) { if (logger.isWarnEnabled()) { @@ -316,14 +395,200 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement /** * The message is just logged. It was already removed from the queue during * the call to receive() - * - * @param sentMessage - * the message that was successfully delivered + * @param sentMessage the message that was successfully delivered + * @deprecated with no replacement. Redundant method. */ + @Deprecated public void onSend(Message sentMessage) { if (logger.isDebugEnabled()) { logger.debug("Sent: " + sentMessage); } } + @UsesJava7 + public enum WatchEventType { + + CREATE(StandardWatchEventKinds.ENTRY_CREATE), + + MODIFY(StandardWatchEventKinds.ENTRY_MODIFY), + + DELETE(StandardWatchEventKinds.ENTRY_DELETE); + + private final WatchEvent.Kind kind; + + WatchEventType(WatchEvent.Kind kind) { + this.kind = kind; + } + + } + + @UsesJava7 + private class WatchServiceDirectoryScanner extends DefaultDirectoryScanner implements Lifecycle { + + private final ConcurrentMap pathKeys = new ConcurrentHashMap(); + + private WatchService watcher; + + private Collection initialFiles; + + private WatchEvent.Kind[] kinds; + + @Override + public void start() { + try { + this.watcher = FileSystems.getDefault().newWatchService(); + } + catch (IOException e) { + logger.error("Failed to create watcher for " + FileReadingMessageSource.this.directory, e); + } + + this.kinds = new WatchEvent.Kind[FileReadingMessageSource.this.watchEvents.length]; + + for (int i = 0; i < FileReadingMessageSource.this.watchEvents.length; i++) { + this.kinds[i] = FileReadingMessageSource.this.watchEvents[i].kind; + } + + final Set initialFiles = walkDirectory(FileReadingMessageSource.this.directory.toPath(), null); + initialFiles.addAll(filesFromEvents()); + this.initialFiles = initialFiles; + } + + @Override + public void stop() { + try { + this.watcher.close(); + this.watcher = null; + } + catch (IOException e) { + logger.error("Failed to close watcher for " + FileReadingMessageSource.this.directory, e); + } + } + + @Override + public boolean isRunning() { + return true; + } + + @Override + protected File[] listEligibleFiles(File directory) { + Assert.state(this.watcher != null, "The WatchService has'nt been started"); + if (this.initialFiles != null) { + File[] initial = this.initialFiles.toArray(new File[this.initialFiles.size()]); + this.initialFiles = null; + return initial; + } + Collection files = filesFromEvents(); + return files.toArray(new File[files.size()]); + } + + private Set filesFromEvents() { + WatchKey key = this.watcher.poll(); + Set files = new LinkedHashSet(); + while (key != null) { + File parentDir = ((Path) key.watchable()).toAbsolutePath().toFile(); + for (WatchEvent event : key.pollEvents()) { + if (event.kind() == StandardWatchEventKinds.ENTRY_CREATE || + event.kind() == StandardWatchEventKinds.ENTRY_MODIFY || + event.kind() == StandardWatchEventKinds.ENTRY_DELETE) { + Path item = (Path) event.context(); + File file = new File(parentDir, item.toFile().getName()); + if (logger.isDebugEnabled()) { + logger.debug("Watch event [" + event.kind() + "] for file [" + file + "]"); + } + + if (event.kind() == StandardWatchEventKinds.ENTRY_DELETE) { + if (FileReadingMessageSource.this.filter instanceof ResettableFileListFilter) { + ((ResettableFileListFilter) FileReadingMessageSource.this.filter).remove(file); + } + boolean fileRemoved = files.remove(file); + if (fileRemoved && logger.isDebugEnabled()) { + logger.debug("The file [" + file + + "] has been removed from the queue because of DELETE event."); + } + } + else { + if (file.exists()) { + if (file.isDirectory()) { + files.addAll(walkDirectory(file.toPath(), event.kind())); + } + else { + files.remove(file); + files.add(file); + } + } + else { + if (logger.isDebugEnabled()) { + logger.debug("A file [" + file + "] for the event [" + event.kind() + + "] doesn't exist. Ignored."); + } + } + } + } + else if (event.kind() == StandardWatchEventKinds.OVERFLOW) { + if (logger.isDebugEnabled()) { + logger.debug("Watch event [" + StandardWatchEventKinds.OVERFLOW + + "] with context [" + event.context() + "]"); + } + + for (WatchKey watchKey : pathKeys.values()) { + watchKey.cancel(); + } + this.pathKeys.clear(); + + if (event.context() != null && event.context() instanceof Path) { + files.addAll(walkDirectory((Path) event.context(), event.kind())); + } + else { + files.addAll(walkDirectory(FileReadingMessageSource.this.directory.toPath(), event.kind())); + } + } + } + key.reset(); + key = this.watcher.poll(); + } + return files; + } + + private Set walkDirectory(Path directory, final WatchEvent.Kind kind) { + final Set walkedFiles = new LinkedHashSet(); + try { + registerWatch(directory); + Files.walkFileTree(directory, new SimpleFileVisitor() { + + @Override + public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException { + FileVisitResult fileVisitResult = super.preVisitDirectory(dir, attrs); + registerWatch(dir); + return fileVisitResult; + } + + @Override + public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException { + FileVisitResult fileVisitResult = super.visitFile(file, attrs); + if (!StandardWatchEventKinds.ENTRY_MODIFY.equals(kind)) { + walkedFiles.add(file.toFile()); + } + return fileVisitResult; + } + + }); + } + catch (IOException e) { + logger.error("Failed to walk directory: " + directory.toString(), e); + } + return walkedFiles; + } + + private void registerWatch(Path dir) throws IOException { + if (!this.pathKeys.containsKey(dir)) { + if (logger.isDebugEnabled()) { + logger.debug("registering: " + dir + " for file events"); + } + WatchKey watchKey = dir.register(this.watcher, this.kinds); + this.pathKeys.putIfAbsent(dir, watchKey); + } + } + + } + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/RecursiveLeafOnlyDirectoryScanner.java b/spring-integration-file/src/main/java/org/springframework/integration/file/RecursiveLeafOnlyDirectoryScanner.java index 5b59eb6fa7..4269e7f47c 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/RecursiveLeafOnlyDirectoryScanner.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/RecursiveLeafOnlyDirectoryScanner.java @@ -30,7 +30,7 @@ import java.util.List; * @author Iwein Fuld * @author Gary Russell * - * @deprecated in favor of {@link WatchServiceDirectoryScanner} (when using Java 7 or later) + * @deprecated in favor of {@link FileReadingMessageSource#setUseWatchService(boolean)} (when using Java 7 or later) */ @Deprecated public class RecursiveLeafOnlyDirectoryScanner extends DefaultDirectoryScanner { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/WatchServiceDirectoryScanner.java b/spring-integration-file/src/main/java/org/springframework/integration/file/WatchServiceDirectoryScanner.java index 18bb01e7a6..50376c8c58 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/WatchServiceDirectoryScanner.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/WatchServiceDirectoryScanner.java @@ -60,9 +60,13 @@ import org.springframework.util.Assert; * @author Gary Russell * @author Artem Bilan * @since 4.2 + * @deprecated since 4.3 in favor of internal {@link WatchService} logic in the {@link FileReadingMessageSource}. + * Will be removed in Spring Integration 5.0. * */ +@Deprecated @UsesJava7 +@SuppressWarnings("deprecation") public class WatchServiceDirectoryScanner extends DefaultDirectoryScanner implements SmartLifecycle { private final static Log logger = LogFactory.getLog(WatchServiceDirectoryScanner.class); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java index f4965cb4b4..7c5ffd6053 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 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. @@ -46,6 +46,8 @@ public class FileInboundChannelAdapterParser extends AbstractPollingInboundChann BeanDefinitionBuilder.genericBeanDefinition(FileReadingMessageSourceFactoryBean.class); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "comparator"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "scanner"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "use-watch-service"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "watch-events"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "directory"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-create-directory"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "queue-size"); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java index 08d4a67136..20c48387d1 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java @@ -35,6 +35,8 @@ import org.springframework.integration.file.locking.AbstractFileLockerFilter; * @author Mark Fisher * @author Iwein Fuld * @author Gary Russell + * @author Artem Bilan + * * @since 1.0.3 */ public class FileReadingMessageSourceFactoryBean implements FactoryBean, @@ -54,6 +56,10 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean filter) { if (filter instanceof AbstractFileLockerFilter && (this.locker == null)) { this.setLocker((AbstractFileLockerFilter) filter); @@ -146,6 +160,12 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean The type that will be filtered. */ -public class CompositeFileListFilter implements ReversibleFileListFilter, Closeable { +public class CompositeFileListFilter implements ReversibleFileListFilter, ResettableFileListFilter, Closeable { private final Set> fileFilters; @@ -120,4 +121,16 @@ public class CompositeFileListFilter implements ReversibleFileListFilter, } } + @Override + public boolean remove(F f) { + boolean removed = false; + for (FileListFilter fileFilter : this.fileFilters) { + if (fileFilter instanceof ResettableFileListFilter) { + ((ResettableFileListFilter) fileFilter).remove(f); + removed = true; + } + } + return removed; + } + } diff --git a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.3.xsd b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.3.xsd index 3254cd0583..59c5d38733 100644 --- a/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.3.xsd +++ b/spring-integration-file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-4.3.xsd @@ -94,7 +94,10 @@ Only files matching this regular expression will be picked up by this adapter. - + + Reference to a custom DirectoryScanner implementation. + Mutually exclusive with 'use-watch-service' attribute. + + + + + Indicates if the 'FileReadingMessageSource' should use an internal 'DirectoryScanner' + for the Java 7 'WatchService'. + Mutually exclusive with 'scanner' attribute. + + + + + + + + + + Comma-separated value for the 'FileReadingMessageSource.WatchEventType's + to specify which kinds of files system events the 'WatchService' will listen to. + Used only if 'use-watch-service == true'. + + + + + + - + + + + + + + + + + Configures a Consumer Endpoint for the diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests-context.xml index d1d2dbf365..83959e9dfd 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests-context.xml +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests-context.xml @@ -12,7 +12,8 @@ + channel="input" auto-startup="false" directory="${java.io.tmpdir}/si-test1" + use-watch-service="true"> @@ -29,7 +30,7 @@ - + @@ -38,9 +39,9 @@ - + - + @@ -51,7 +52,7 @@ - + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests.java index f8ed98e93b..adf1b6307c 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests.java @@ -16,9 +16,11 @@ package org.springframework.integration.file; +import static org.hamcrest.Matchers.containsString; 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 java.io.File; @@ -31,6 +33,7 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; @@ -45,6 +48,7 @@ import org.springframework.transaction.support.DefaultTransactionStatus; /** * @author Gary Russell + * @author Artem Bilan * @since 2.2 * */ @@ -108,6 +112,9 @@ public class FileInboundTransactionTests { pseudoTx.stop(); assertFalse(transactionManager.getCommitted()); assertFalse(transactionManager.getRolledBack()); + + Object scanner = TestUtils.getPropertyValue(pseudoTx.getMessageSource(), "scanner"); + assertThat(scanner.getClass().getName(), containsString("FileReadingMessageSource$WatchServiceDirectoryScanner")); } @Test diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java index 29d85b0190..7e02d3db12 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java @@ -111,13 +111,10 @@ public class FileReadingMessageSourceIntegrationTests { public void getFiles() throws Exception { Message received1 = pollableFileSource.receive(); assertNotNull("This should return the first message", received1); - pollableFileSource.onSend(received1); Message received2 = pollableFileSource.receive(); assertNotNull(received2); - pollableFileSource.onSend(received2); Message received3 = pollableFileSource.receive(); assertNotNull(received3); - pollableFileSource.onSend(received3); assertNotSame(received1 + " == " + received2, received1.getPayload(), received2.getPayload()); assertNotSame(received1 + " == " + received3, received1.getPayload(), received3.getPayload()); assertNotSame(received2 + " == " + received3, received2.getPayload(), received3.getPayload()); @@ -154,7 +151,6 @@ public class FileReadingMessageSourceIntegrationTests { Thread.yield(); received = pollableFileSource.receive(); } - pollableFileSource.onSend(received); } }; @@ -181,9 +177,6 @@ public class FileReadingMessageSourceIntegrationTests { } // make sure three different files were taken Message received = pollableFileSource.receive(); - if (received != null) { - pollableFileSource.onSend(received); - } assertNull(received); } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourcePersistentFilterIntegrationTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourcePersistentFilterIntegrationTests.java index 299df932ee..16526160f0 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourcePersistentFilterIntegrationTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourcePersistentFilterIntegrationTests.java @@ -104,13 +104,10 @@ public class FileReadingMessageSourcePersistentFilterIntegrationTests { public void getFiles() throws Exception { Message received1 = pollableFileSource.receive(); assertNotNull("This should return the first message", received1); - pollableFileSource.onSend(received1); Message received2 = pollableFileSource.receive(); assertNotNull(received2); - pollableFileSource.onSend(received2); Message received3 = pollableFileSource.receive(); assertNotNull(received3); - pollableFileSource.onSend(received3); assertNotSame(received1 + " == " + received2, received1.getPayload(), received2.getPayload()); assertNotSame(received1 + " == " + received3, received1.getPayload(), received3.getPayload()); assertNotSame(received2 + " == " + received3, received2.getPayload(), received3.getPayload()); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java index 372fee6664..b37900e5bb 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/WatchServiceDirectoryScannerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-2016 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. @@ -16,8 +16,11 @@ package org.springframework.integration.file; +import static org.hamcrest.Matchers.containsString; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; import java.io.File; import java.io.IOException; @@ -25,12 +28,18 @@ import java.util.ArrayList; import java.util.HashSet; import java.util.List; import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.junit.Before; import org.junit.Rule; import org.junit.Test; import org.junit.rules.TemporaryFolder; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.file.filters.FileSystemPersistentAcceptOnceFileListFilter; +import org.springframework.integration.metadata.SimpleMetadataStore; + /** * @author Gary Russell * @author Artem Bilan @@ -62,9 +71,35 @@ public class WatchServiceDirectoryScannerTests { } @Test - public void testInitialAndAddMore() throws Exception { - WatchServiceDirectoryScanner scanner = new WatchServiceDirectoryScanner(folder.getRoot().getAbsolutePath()); - scanner.start(); + public void testInitialAndAddMoreThanRemove() throws Exception { + FileReadingMessageSource fileReadingMessageSource = new FileReadingMessageSource(); + fileReadingMessageSource.setDirectory(folder.getRoot()); + fileReadingMessageSource.setUseWatchService(true); + fileReadingMessageSource.setWatchEvents(FileReadingMessageSource.WatchEventType.CREATE, + FileReadingMessageSource.WatchEventType.MODIFY, + FileReadingMessageSource.WatchEventType.DELETE); + fileReadingMessageSource.setBeanFactory(mock(BeanFactory.class)); + + final CountDownLatch removeFileLatch = new CountDownLatch(1); + + FileSystemPersistentAcceptOnceFileListFilter filter = + new FileSystemPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "test") { + + @Override + public boolean remove(File fileToRemove) { + removeFileLatch.countDown(); + return super.remove(fileToRemove); + } + + }; + + fileReadingMessageSource.setFilter(filter); + fileReadingMessageSource.afterPropertiesSet(); + fileReadingMessageSource.start(); + DirectoryScanner scanner = fileReadingMessageSource.getScanner(); + assertThat(scanner.getClass().getName(), + containsString("FileReadingMessageSource$WatchServiceDirectoryScanner")); + List files = scanner.listFiles(folder.getRoot()); assertEquals(3, files.size()); assertTrue(files.contains(top1)); @@ -125,7 +160,26 @@ public class WatchServiceDirectoryScannerTests { assertTrue(accum.contains(baz2)); - scanner.stop(); + File baz2Copy = new File(baz2.getAbsolutePath()); + + baz2Copy.setLastModified(baz2.lastModified() + 100000); + + Thread.sleep(100); + + files = scanner.listFiles(folder.getRoot()); + + assertEquals(1, files.size()); + assertTrue(files.contains(baz2)); + + baz2.delete(); + + Thread.sleep(100); + + scanner.listFiles(folder.getRoot()); + + assertTrue(removeFileLatch.await(10, TimeUnit.SECONDS)); + + fileReadingMessageSource.stop(); } } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml index 625c973903..abd65444f1 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml @@ -15,9 +15,11 @@ filter="filter" comparator="testComparator" ignore-hidden="false" - auto-startup="false"> + auto-startup="false" + use-watch-service="true" + watch-events="MODIFY, DELETE"> - + @@ -25,10 +27,10 @@ directory="${java.io.tmpdir}" filter="filter" auto-startup="false"> - + - + @@ -47,12 +49,17 @@ - + - - + + - + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java index 7d5547dd3c..b89e3ce8af 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 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. @@ -16,8 +16,11 @@ package org.springframework.integration.file.config; +import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.instanceOf; +import static org.hamcrest.Matchers.isOneOf; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; @@ -52,13 +55,14 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * @author Mark Fisher * @author Gary Russell * @author Gunnar Hillert + * @author Artem Bilan */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) @DirtiesContext public class FileInboundChannelAdapterParserTests { - @Autowired(required = true) + @Autowired private ApplicationContext context; @Autowired @@ -107,6 +111,18 @@ public class FileInboundChannelAdapterParserTests { Object filter = scannerAccessor.getPropertyValue("filter"); assertTrue("'filter' should be set and be of instance AcceptOnceFileListFilter but got " + filter.getClass().getSimpleName(), filter instanceof AcceptOnceFileListFilter); + + assertThat(scanner.getClass().getName(), + containsString("FileReadingMessageSource$WatchServiceDirectoryScanner")); + + FileReadingMessageSource.WatchEventType[] watchEvents = + (FileReadingMessageSource.WatchEventType[]) this.accessor.getPropertyValue("watchEvents"); + assertEquals(2, watchEvents.length); + for (FileReadingMessageSource.WatchEventType watchEvent : watchEvents) { + assertNotEquals(FileReadingMessageSource.WatchEventType.CREATE, watchEvent); + assertThat(watchEvent, isOneOf(FileReadingMessageSource.WatchEventType.MODIFY, + FileReadingMessageSource.WatchEventType.DELETE)); + } } @Test diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index 0bb9a54e37..e647ad3137 100644 --- a/src/reference/asciidoc/file.adoc +++ b/src/reference/asciidoc/file.adoc @@ -231,11 +231,12 @@ on that `scanner` not on the `FileReadingMessageSource`. NOTE: The `DefaultDirectoryScanner` uses a `IgnoreHiddenFileListFilter` and `AcceptOnceFileListFilter` by default. To prevent their use, you should configure your own filter (e.g. `AcceptAllFileListFilter`) or even set it to `null`. - +[[watch-service-directory-scanner]] ==== WatchServiceDirectoryScanner This scanner was added in _version 4.2_. It replaces the existing `RecursiveLeafOnlyDirectoryScanner` which is -inefficient for large directory trees. The `WatchServiceDirectoryScanner` requires Java 7 or above. +inefficient for large directory trees. +The `FileReadingMessageSource.WatchServiceDirectoryScanner` requires Java 7 or above. This scanner relies on file system events when new files are added to the directory. During initialization, the directory is registered to generate events; the initial file list is also built. @@ -253,19 +254,38 @@ In this case, the root directory is re-scanned completely. To avoid duplicates consider using an appropriate `FileListFilter` such as the `AcceptOnceFileListFilter` and/or remove files when processing is completed. -[source, xml] ----- - - - ----- +Since _version 4.3_, the top level `WatchServiceDirectoryScanner` has been deprecated in favor of +`FileReadingMessageSource` internal logic for the `WatchService`. +Now this can be enable via `use-watch-service` option, which is mutually exclusive with the `scanner` option. +An internal `FileReadingMessageSource.WatchServiceDirectoryScanner` instance is populated for the provided `directory`. -[source, java] +In addition, now the `WatchService` polling logic can track the `StandardWatchEventKinds.ENTRY_MODIFY` and +`StandardWatchEventKinds.ENTRY_DELETE`, too. + +The `ENTRY_MODIFY` events logic should be implemented properly in the `FileListFilter` to track not only new files but +also the modification, if that is requirement. +Otherwise the files from those events are treated the same way. + +The `ENTRY_DELETE` events have effect for the `ResettableFileListFilter` implementations and, therefore, their files + are provided for the `remove()` operation. + +For this purpose the `watch-events` +(`FileReadingMessageSource.setWatchEvents(FileReadingMessageSource.WatchEventType... watchEvents)`) has been introduced. +With such an option we can implement some scenarios, when we would like to do one downstream flow logic for new files, +and other for modified. +We can achieve that with different `` definitions, but for the same directory: + +[source,xml] ---- -@Bean -public DirectoryScanner scanner() { - return new WatchServiceDirectoryScanner("/tmp/myDir"); -} + + + ---- ==== Limiting Memory Consumption diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 0a34ad6848..e57098c9f9 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -94,6 +94,11 @@ The generated file name for the `FileWritingMessageHandler` can represent _sub-p structure for file in the target directory. See <> for more information. +The `FileReadingMessageSource` now hides the `WatchService` directory scanning logic in the inner class. +The `use-watch-service` and `watch-events` options are provided to enable such a behaviour. +The top level `WatchServiceDirectoryScanner` has been deprecated because of inconsistency around API. +See <> for more information. + ===== Buffer Size When writing files, you can now specify the buffer size to use.