Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-2507: File: Fix HeadDirectoryScanner

Browse Source 
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).
This commit is contained in:
Gary Russell
2015-06-29 16:13:44 -04:00
committed by Artem Bilan
parent 029bb28cfa
commit cef99f3584
6 changed files with 300 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
spring-integration-file/src/main/java/org/springframework/integration/file/HeadDirectoryScanner.java
View File

@@ -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 <code>maxNumberOfFiles</code>
* 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<File> filter) {
if (filter instanceof CompositeFileListFilter) {
((CompositeFileListFilter<File>) filter).addFilter(this.headFilter);
super.setFilter(filter);
}
else {
CompositeFileListFilter<File> compositeFilter = new CompositeFileListFilter<File>();
compositeFilter.addFilter(filter).addFilter(this.headFilter);
super.setFilter(compositeFilter);
}
}
@@ -45,6 +65,7 @@ public class HeadDirectoryScanner extends DefaultDirectoryScanner {
this.maxNumberOfFiles = maxNumberOfFiles;
}
@Override
public List<File> filterFiles(File[] files) {
return Arrays.asList(files).subList(0, Math.min(files.length, maxNumberOfFiles));
}

296
spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java
View File

@@ -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<File> received1 = pollableFileSource.receive();
System.out.println("receive files round 1");
assertNotNull("This should return the first message", received1);
pollableFileSource.onSend(received1);
Message<File> received2 = pollableFileSource.receive();
assertNotNull(received2);
pollableFileSource.onSend(received2);
Message<File> 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<File> received1 = pollableFileSource.receive();
System.out.println("receive files round 1");
assertNotNull("This should return the first message", received1);
pollableFileSource.onSend(received1);
Message<File> received2 = pollableFileSource.receive();
assertNotNull(received2);
pollableFileSource.onSend(received2);
Message<File> 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<File> received1 = pollableFileSource.receive();
Message<File> received2 = pollableFileSource.receive();
Message<File> 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<File> received1 = pollableFileSource.receive();
Message<File> received2 = pollableFileSource.receive();
Message<File> 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<File> received = pollableFileSource.receive();
while (received == null) {
Thread.yield();
received = pollableFileSource.receive();
}
pollableFileSource.onSend(received);
}
};
Runnable failingConsumer = new Runnable() {
public void run() {
Message<File> 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<File> 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<File> received = pollableFileSource.receive();
while (received == null) {
Thread.yield();
received = pollableFileSource.receive();
}
pollableFileSource.onSend(received);
}
};
Runnable failingConsumer = new Runnable() {
@Override
public void run() {
Message<File> 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<File> 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;
}
}

54
spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests-context.xml
View File

@@ -1,6 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ 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.
@@ -16,34 +16,46 @@
-->
<beans:beans xmlns="http://www.springframework.org/schema/integration/file"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:integration="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:integration="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/file
http://www.springframework.org/schema/integration/file/spring-integration-file.xsd">
<inbound-channel-adapter id="inputDirPoller"
directory="${java.io.tmpdir}"
filter="filter"
queue-size="30"
auto-startup="false">
<integration:poller fixed-rate="1000"/>
</inbound-channel-adapter>
<beans:bean class="org.springframework.integration.file.config.FileListFilterFactoryBean" />
<inbound-channel-adapter id="inputDirPoller"
directory="${java.io.tmpdir}/FileInboundChannelAdapterWithQueueSizeTests"
filter="filter"
queue-size="2"
auto-startup="false">
<integration:poller fixed-rate="1000" />
</inbound-channel-adapter>
<beans:bean id="filter" class="org.springframework.integration.file.filters.CompositeFileListFilter">
<beans:constructor-arg>
<beans:list></beans:list>
</beans:constructor-arg>
</beans:bean>
<beans:bean class="org.springframework.integration.file.config.FileListFilterFactoryBean" />
<beans:bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"/>
<beans:bean id="filter" class="org.springframework.integration.file.filters.CompositeFileListFilter">
<beans:constructor-arg>
<beans:list />
</beans:constructor-arg>
</beans:bean>
<beans:bean id="testComparator"
class="org.springframework.integration.file.config.FileInboundChannelAdapterParserTests$TestComparator"/>
<beans:bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer" />
<beans:bean id="testComparator"
class="org.springframework.integration.file.config.FileInboundChannelAdapterParserTests$TestComparator" />
<!-- test second leg of if test in HeadDS.setFilter - no CompositeFLF -->
<inbound-channel-adapter id="inputDirPollerSimpleFilter"
directory="${java.io.tmpdir}/FileInboundChannelAdapterWithQueueSizeTests"
filter="acceptAll"
ignore-hidden="false"
queue-size="2"
auto-startup="false">
<integration:poller fixed-rate="1000" />
</inbound-channel-adapter>
<beans:bean id="acceptAll" class="org.springframework.integration.file.filters.AcceptAllFileListFilter" />
</beans:beans>

73
spring-integration-file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests.java
View File

@@ -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<File> 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());
}
}

25
src/reference/asciidoc/file.adoc
View File

@@ -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

2
src/reference/asciidoc/whats-new.adoc
View File

@@ -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 <<files>> for more information.
[[x4.2-class-package-change]]