From cef99f3584d98f8aa1c77aeee7e3ebe60a29f5ec Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 29 Jun 2015 16:13:44 -0400 Subject: [PATCH] INT-2507: File: Fix HeadDirectoryScanner JIRA: https://jira.spring.io/browse/INT-2507 When using a `HeadDirectoryScanner`, the use of any filter removed the size limitation passed into the contstructor (or `queue-size` attribute of a file inbound channel adapter. The `HeadFilter` was overwritten. Combine the head filter into a `CompositeFileListFilter` (creating one if necessary). --- .../file/HeadDirectoryScanner.java | 29 +- ...eReadingMessageSourceIntegrationTests.java | 296 ++++++++++-------- ...annelAdapterWithQueueSizeTests-context.xml | 54 ++-- ...boundChannelAdapterWithQueueSizeTests.java | 73 ++++- src/reference/asciidoc/file.adoc | 25 +- src/reference/asciidoc/whats-new.adoc | 2 + 6 files changed, 300 insertions(+), 179 deletions(-) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/HeadDirectoryScanner.java b/spring-integration-file/src/main/java/org/springframework/integration/file/HeadDirectoryScanner.java index fbe7d58e1a..443554d057 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/HeadDirectoryScanner.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/HeadDirectoryScanner.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2015 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,20 +20,40 @@ import java.io.File; import java.util.Arrays; import java.util.List; +import org.springframework.integration.file.filters.CompositeFileListFilter; import org.springframework.integration.file.filters.FileListFilter; /** * A custom scanner that only returns the first maxNumberOfFiles * elements from a directory listing. This is useful to limit the number of File - * objects in memory and therefore mutually exclusive with AcceptOnceFileListFilter. - * + * objects in memory and therefore mutually exclusive with {@code AcceptOnceFileListFilter}. + * It should not be used in conjunction with an {@code AcceptOnceFileListFilter}. + * * @author Iwein Fuld + * @author Gary Russell * @since 2.0 */ public class HeadDirectoryScanner extends DefaultDirectoryScanner { + private final HeadFilter headFilter; + public HeadDirectoryScanner(int maxNumberOfFiles) { - this.setFilter(new HeadFilter(maxNumberOfFiles)); + HeadFilter headFilter = new HeadFilter(maxNumberOfFiles); + this.setFilter(headFilter); + this.headFilter = headFilter; + } + + @Override + public void setFilter(FileListFilter filter) { + if (filter instanceof CompositeFileListFilter) { + ((CompositeFileListFilter) filter).addFilter(this.headFilter); + super.setFilter(filter); + } + else { + CompositeFileListFilter compositeFilter = new CompositeFileListFilter(); + compositeFilter.addFilter(filter).addFilter(this.headFilter); + super.setFilter(compositeFilter); + } } @@ -45,6 +65,7 @@ public class HeadDirectoryScanner extends DefaultDirectoryScanner { this.maxNumberOfFiles = maxNumberOfFiles; } + @Override public List filterFiles(File[] files) { return Arrays.asList(files).subList(0, Math.min(files.length, maxNumberOfFiles)); } 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 7d9de0bed5..9311637b69 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2015 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. @@ -40,161 +40,179 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Iwein Fuld + * @author Gary Russell */ @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration public class FileReadingMessageSourceIntegrationTests { - @Autowired - FileReadingMessageSource pollableFileSource; + @Autowired + FileReadingMessageSource pollableFileSource; - private static File inputDir; + private static File inputDir; - @AfterClass - public static void cleanUp() throws Throwable { - if(inputDir.exists()) { - inputDir.delete(); - } - } + @AfterClass + public static void cleanUp() throws Throwable { + if(inputDir.exists()) { + inputDir.delete(); + } + } - @BeforeClass - public static void setupInputDir() { - inputDir = new File(System.getProperty("java.io.tmpdir") + "/" - + FileReadingMessageSourceIntegrationTests.class.getSimpleName()); - inputDir.mkdir(); - } + @BeforeClass + public static void setupInputDir() { + inputDir = new File(System.getProperty("java.io.tmpdir") + "/" + + FileReadingMessageSourceIntegrationTests.class.getSimpleName()); + inputDir.mkdir(); + clean(); + } - @Before - public void generateTestFiles() throws Exception { - File.createTempFile("test", null, inputDir).setLastModified(System.currentTimeMillis() - 1000); - File.createTempFile("test", null, inputDir).setLastModified(System.currentTimeMillis() - 1000); - File.createTempFile("test", null, inputDir).setLastModified(System.currentTimeMillis() - 1000); - } + @AfterClass + public static void tearDown() { + clean(); + inputDir.delete(); + } - @After - public void cleanoutInputDir() throws Exception { - File[] listFiles = inputDir.listFiles(); - for (int i = 0; i < listFiles.length; i++) { - listFiles[i].delete(); - } - } + private static void clean() { + File[] files = inputDir.listFiles(); + for (File file : files) { + file.delete(); + } + } - @AfterClass - public static void removeInputDir() throws Exception { - inputDir.delete(); - } + @Before + public void generateTestFiles() throws Exception { + File.createTempFile("test", null, inputDir).setLastModified(System.currentTimeMillis() - 1000); + File.createTempFile("test", null, inputDir).setLastModified(System.currentTimeMillis() - 1000); + File.createTempFile("test", null, inputDir).setLastModified(System.currentTimeMillis() - 1000); + } + + @After + public void cleanoutInputDir() throws Exception { + File[] listFiles = inputDir.listFiles(); + for (int i = 0; i < listFiles.length; i++) { + listFiles[i].delete(); + } + } + + @AfterClass + public static void removeInputDir() throws Exception { + inputDir.delete(); + } - @Test - public void configured() throws Exception { - DirectFieldAccessor accessor = new DirectFieldAccessor(pollableFileSource); - assertEquals(inputDir, accessor.getPropertyValue("directory")); - } + @Test + public void configured() throws Exception { + DirectFieldAccessor accessor = new DirectFieldAccessor(pollableFileSource); + assertEquals(inputDir, accessor.getPropertyValue("directory")); + } - @Test - public void getFiles() throws Exception { - Message received1 = pollableFileSource.receive(); - System.out.println("receive files round 1"); - 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()); - } + @Test + public void getFiles() throws Exception { + Message received1 = pollableFileSource.receive(); + System.out.println("receive files round 1"); + 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()); + } - @Test - public void parallelRetrieval() throws Exception { - Message received1 = pollableFileSource.receive(); - Message received2 = pollableFileSource.receive(); - Message received3 = pollableFileSource.receive(); - assertNotSame(received1 + " == " + received2, received1, received2); - assertNotSame(received1 + " == " + received3, received1, received3); - assertNotSame(received2 + " == " + received3, received2, received3); - } + @Test + public void parallelRetrieval() throws Exception { + Message received1 = pollableFileSource.receive(); + Message received2 = pollableFileSource.receive(); + Message received3 = pollableFileSource.receive(); + assertNotSame(received1 + " == " + received2, received1, received2); + assertNotSame(received1 + " == " + received3, received1, received3); + assertNotSame(received2 + " == " + received3, received2, received3); + } - @Test - public void inputDirExhausted() throws Exception { - assertNotNull(pollableFileSource.receive()); - assertNotNull(pollableFileSource.receive()); - assertNotNull(pollableFileSource.receive()); - assertNull(pollableFileSource.receive()); - } + @Test + public void inputDirExhausted() throws Exception { + assertNotNull(pollableFileSource.receive()); + assertNotNull(pollableFileSource.receive()); + assertNotNull(pollableFileSource.receive()); + assertNull(pollableFileSource.receive()); + } - @Test(timeout = 6000) - @Repeat(5) - public void concurrentProcessing() throws Exception { - CountDownLatch go = new CountDownLatch(1); - Runnable succesfulConsumer = new Runnable() { - public void run() { - Message received = pollableFileSource.receive(); - while (received == null) { - Thread.yield(); - received = pollableFileSource.receive(); - } - pollableFileSource.onSend(received); - } - }; - Runnable failingConsumer = new Runnable() { - public void run() { - Message received = pollableFileSource.receive(); - if (received != null) { - pollableFileSource.onFailure(received); - } - } - }; - CountDownLatch succesfulDone = doConcurrently(3, succesfulConsumer, go); - CountDownLatch failingDone = doConcurrently(10, failingConsumer, go); - go.countDown(); - try { - succesfulDone.await(); - failingDone.await(); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - // make sure three different files were taken - Message received = pollableFileSource.receive(); - if (received != null) { - pollableFileSource.onSend(received); - } - assertNull(received); - } + @Test(timeout = 6000) + @Repeat(5) + public void concurrentProcessing() throws Exception { + CountDownLatch go = new CountDownLatch(1); + Runnable succesfulConsumer = new Runnable() { + @Override + public void run() { + Message received = pollableFileSource.receive(); + while (received == null) { + Thread.yield(); + received = pollableFileSource.receive(); + } + pollableFileSource.onSend(received); + } + }; + Runnable failingConsumer = new Runnable() { + @Override + public void run() { + Message received = pollableFileSource.receive(); + if (received != null) { + pollableFileSource.onFailure(received); + } + } + }; + CountDownLatch succesfulDone = doConcurrently(3, succesfulConsumer, go); + CountDownLatch failingDone = doConcurrently(10, failingConsumer, go); + go.countDown(); + try { + succesfulDone.await(); + failingDone.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + // make sure three different files were taken + Message received = pollableFileSource.receive(); + if (received != null) { + pollableFileSource.onSend(received); + } + assertNull(received); + } - /** - * Convenience method to run part of a test concurrently in multiple threads - * - * @param numberOfThreads how many threads to spawn - * @param runnable the runnable that should be run by all the threads - * @param start the {@link java.util.concurrent.CountDownLatch} instance telling it when to assume everything works - * @return a latch that will be counted down once all threads have run their - * runnable. - */ - private CountDownLatch doConcurrently(int numberOfThreads, final Runnable runnable, final CountDownLatch start) { - final CountDownLatch started = new CountDownLatch(numberOfThreads); - final CountDownLatch done = new CountDownLatch(numberOfThreads); - for (int i = 0; i < numberOfThreads; i++) { - new Thread(new Runnable() { + /** + * Convenience method to run part of a test concurrently in multiple threads + * + * @param numberOfThreads how many threads to spawn + * @param runnable the runnable that should be run by all the threads + * @param start the {@link java.util.concurrent.CountDownLatch} instance telling it when to assume everything works + * @return a latch that will be counted down once all threads have run their + * runnable. + */ + private CountDownLatch doConcurrently(int numberOfThreads, final Runnable runnable, final CountDownLatch start) { + final CountDownLatch started = new CountDownLatch(numberOfThreads); + final CountDownLatch done = new CountDownLatch(numberOfThreads); + for (int i = 0; i < numberOfThreads; i++) { + new Thread(new Runnable() { - public void run() { - started.countDown(); - try { - started.await(); - start.await(); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - runnable.run(); - done.countDown(); - } - }).start(); - } - return done; - } + @Override + public void run() { + started.countDown(); + try { + started.await(); + start.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + runnable.run(); + done.countDown(); + } + }).start(); + } + return done; + } } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests-context.xml index 5d84dd8d18..bed533a311 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests-context.xml +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests-context.xml @@ -1,6 +1,6 @@ - - - - + + + - - - - - + - + + + + + - + + + + + + + + + + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests.java index 36f42a2fe6..e9e60b403a 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,40 +16,87 @@ package org.springframework.integration.file.config; +import static org.junit.Assert.assertEquals; + +import java.io.File; +import java.util.List; + +import org.junit.AfterClass; import org.junit.Before; +import org.junit.BeforeClass; import org.junit.Test; import org.junit.runner.RunWith; -import org.springframework.beans.DirectFieldAccessor; + import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.integration.file.FileReadingMessageSource; import org.springframework.integration.file.HeadDirectoryScanner; +import org.springframework.integration.test.util.TestUtils; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import static org.hamcrest.CoreMatchers.is; -import static org.hamcrest.CoreMatchers.instanceOf; -import static org.junit.Assert.assertThat; - /** * @author Gunnar Hillert + * @author Gary Russell */ @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration public class FileInboundChannelAdapterWithQueueSizeTests { - @Autowired - FileReadingMessageSource source; + private static final String PATHNAME = System.getProperty("java.io.tmpdir") + "/" + + FileInboundChannelAdapterWithQueueSizeTests.class.getSimpleName(); - private DirectFieldAccessor accessor; + private static File inputDir; + + @Autowired + @Qualifier("inputDirPoller.adapter.source") + FileReadingMessageSource source1; + + @Autowired + @Qualifier("inputDirPollerSimpleFilter.adapter.source") + FileReadingMessageSource source2; + + @BeforeClass + public static void setupInputDir() { + inputDir = new File(PATHNAME); + inputDir.mkdir(); + clean(); + } + + @AfterClass + public static void tearDown() { + clean(); + inputDir.delete(); + } + + private static void clean() { + File[] files = inputDir.listFiles(); + for (File file : files) { + file.delete(); + } + } @Before - public void init() { - accessor = new DirectFieldAccessor(source); + public void generateTestFiles() throws Exception { + File.createTempFile("test", null, inputDir).setLastModified(System.currentTimeMillis() - 1000); + File.createTempFile("test", null, inputDir).setLastModified(System.currentTimeMillis() - 1000); + File.createTempFile("test", null, inputDir).setLastModified(System.currentTimeMillis() - 1000); } @Test public void queueSize() { - Object scanner = accessor.getPropertyValue("scanner"); - assertThat(scanner, is(instanceOf(HeadDirectoryScanner.class))); + HeadDirectoryScanner scanner1 = TestUtils.getPropertyValue(source1, "scanner", HeadDirectoryScanner.class); + HeadDirectoryScanner scanner2 = TestUtils.getPropertyValue(source2, "scanner", HeadDirectoryScanner.class); + List files = scanner1.listFiles(new File(PATHNAME)); + assertEquals(2, files.size()); + files = scanner2.listFiles(new File(PATHNAME)); + assertEquals(2, files.size()); + files.get(0).delete(); + files.get(1).delete(); + files = scanner1.listFiles(new File(PATHNAME)); + assertEquals(1, files.size()); + files = scanner2.listFiles(new File(PATHNAME)); + assertEquals(1, files.size()); } + } diff --git a/src/reference/asciidoc/file.adoc b/src/reference/asciidoc/file.adoc index 28dccdb92c..82c0d7ca89 100644 --- a/src/reference/asciidoc/file.adoc +++ b/src/reference/asciidoc/file.adoc @@ -37,7 +37,7 @@ hidden files. [IMPORTANT] ===== -The `IgnoreHiddenFileListFilter` was introduced with _version 4.2_. In prior versions hidden files *were being picked up*. +The `IgnoreHiddenFileListFilter` was introduced with _version 4.2_. In prior versions hidden files were included. With the default configuration, the `IgnoreHiddenFileListFilter` will be triggered first, then the `AcceptOnceFileListFilter`. ===== @@ -136,7 +136,7 @@ Therefore, you can also leave off the 2 attributes `prevent-duplicates` and `ign [IMPORTANT] ===== -The `ignore-hidden` attribute was introduced with _Spring Integration 4.2_. In prior versions hidden files *were being picked up*. +The `ignore-hidden` attribute was introduced with _Spring Integration 4.2_. In prior versions hidden files were included. ===== The second channel adapter example is using a custom filter, the third is using the _filename-pattern_ attribute to @@ -187,6 +187,27 @@ IMPORTANT: It is important to understand that filters (including patterns, regex Any of these attributes set on the adapter are subsequently injected into the scanner. For this reason, if you need to provide a custom scanner and you have multiple file inbound adapters in the same application context, each adapter must be provided with its own instance of the scanner, either by declaring separate beans, or declaring `scope="prototype"` on the scanner bean so that the context will create a new instance for each use. +===== Limiting Memory Consumption + +A `HeadDirectoryScanner` can be used to limit the number of files retained in memory. +This can be useful when scanning large directories. +With XML configuration, this is enabled using the `queue-size` property on the inbound channel adapter. + +Prior to _version 4.2_, this setting was incompatible with the use of any other filters. +Any other filters (including `prevent-duplicates="true"`) overwrote the filter used to limit the size. + +[NOTE] +===== +The use of a `HeadDirectoryScanner` is incompatible with an `AcceptOnceFileListFilter`. +Since all filters are consulted during the poll decision, the `AcceptOnceFileListFilter` does not know +that other filters might be temporarily filtering files. +Even if files that were previously filtered by the `HeadDirectoryScanner.HeadFilter` are now available, the +`AcceptOnceFileListFilter` will filter them. + +Generally, instead of using an `AcceptOnceFileListFilter` in this case, one would simply remove the processed +files so that the previously filtered files will be available on a future poll. +===== + [[file-tailing]] ==== 'Tail'ing Files diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 1de1519638..b2e1fd8ded 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -56,6 +56,8 @@ It is `true` by default. The `FileWritingMessageHandler` now also accepts `InputStream` as a valid message payload type. +The `HeadDirectoryScanner` can now be used with other `FileListFilter` s. + See <> for more information. [[x4.2-class-package-change]]