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

INT-3619: Prevent scanner Options Overrides

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3619

Previously the `FileReadingMessageSource` allowed overriding `filter` and `locker` properties
on the `scanner` through its own setters.
It caused confusing and unexpected behavior.

Provide refactoring to check that provided `filter` and `locker` are applied for an internal `scanner`
instance, not for the injected one.

PR Comments
This commit is contained in:
Artem Bilan
2015-08-11 17:48:22 -04:00
committed by Gary Russell
parent adb0f43aa8
commit 057c004e6c
7 changed files with 167 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
spring-integration-file/src/main/java/org/springframework/integration/file/FileMessageHolder.java
View File

@@ -1,43 +0,0 @@
/*
* Copyright 2002-2013 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.file;
import java.io.File;
import org.springframework.messaging.Message;
/**
* A simple wrapper for a {@code Message<File>;} used for
* file disposition after the send completes, or
* after the transaction commits with a transactional
* poller.
* @author Gary Russell
* @since 2.2
*
*/
public class FileMessageHolder {
private Message<File> message;
Message<File> getMessage() {
return message;
}
void setMessage(Message<File> message) {
this.message = message;
}
}

77
spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 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.
@@ -37,12 +37,10 @@ import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
/**
* {@link MessageSource} that creates messages from a file system directory. To
* prevent messages for certain files, you may supply a
* {@link FileListFilter}. By
* default, an
* {@link AcceptOnceFileListFilter}
* is used. It ensures files are picked up only once from the directory.
* {@link MessageSource} that creates messages from a file system directory.
* To prevent messages for certain files, you may supply a {@link FileListFilter}.
* By default, an {@link AcceptOnceFileListFilter} is used.
* It ensures files are picked up only once from the directory.
* <p>
* A common problem with reading files is that a file may be detected before it
* is ready. The default {@link AcceptOnceFileListFilter}
@@ -64,6 +62,7 @@ import org.springframework.util.Assert;
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Artem Bilan
*/
public class FileReadingMessageSource extends IntegrationObjectSupport implements MessageSource<File> {
@@ -71,13 +70,6 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
private static final Log logger = LogFactory.getLog(FileReadingMessageSource.class);
private volatile File directory;
private volatile DirectoryScanner scanner = new DefaultDirectoryScanner();
private volatile boolean autoCreateDirectory = true;
/*
* {@link PriorityBlockingQueue#iterator()} throws
* {@link java.util.ConcurrentModificationException} in Java 5.
@@ -85,9 +77,19 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
*/
private final Queue<File> toBeReceived;
private volatile File directory;
private volatile DirectoryScanner scanner = new DefaultDirectoryScanner();
private volatile boolean scannerExplicitlySet;
private volatile boolean autoCreateDirectory = true;
private volatile boolean scanEachPoll = false;
private final ThreadLocal<FileMessageHolder> resources = new ThreadLocal<FileMessageHolder>();
private FileListFilter<File> filter;
private FileLocker locker;
/**
* Creates a FileReadingMessageSource with a naturally ordered queue of unbounded capacity.
@@ -114,7 +116,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
this(null);
Assert.isTrue(internalQueueCapacity > 0,
"Cannot create a queue with non positive capacity");
this.setScanner(new HeadDirectoryScanner(internalQueueCapacity));
this.scanner = new HeadDirectoryScanner(internalQueueCapacity);
}
/**
@@ -149,12 +151,25 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
/**
* Optionally specify a custom scanner, for example the
* {@link org.springframework.integration.file.RecursiveLeafOnlyDirectoryScanner}
* {@link WatchServiceDirectoryScanner}
*
* @param scanner scanner implementation
*/
public void setScanner(DirectoryScanner scanner) {
Assert.notNull(scanner, "'scanner' must not be null.");
this.scanner = scanner;
this.scannerExplicitlySet = true;
}
/**
* The {@link #scanner} property accessor to allow to modify its options
* ({@code filter}, {@code locker} etc.) at runtime using the
* {@link FileReadingMessageSource} bean.
* @return the {@link DirectoryScanner} of this {@link FileReadingMessageSource}.
* @since 4.2
*/
public DirectoryScanner getScanner() {
return scanner;
}
/**
@@ -173,21 +188,20 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
}
/**
* Sets a {@link FileListFilter}. By default a
* {@link org.springframework.integration.file.filters.AbstractFileListFilter}
* Sets a {@link FileListFilter}.
* By default a {@link org.springframework.integration.file.filters.AcceptOnceFileListFilter}
* with no bounds is used. In most cases a customized {@link FileListFilter} will
* be needed to deal with modification and duplication concerns. If multiple
* filters are required a
* be needed to deal with modification and duplication concerns.
* If multiple filters are required a
* {@link org.springframework.integration.file.filters.CompositeFileListFilter}
* can be used to group them together.
* <p>
* <b>The supplied filter must be thread safe.</b>.
*
* @param filter a filter
*/
public void setFilter(FileListFilter<File> filter) {
Assert.notNull(filter, "'filter' must not be null");
this.scanner.setFilter(filter);
this.filter = filter;
}
/**
@@ -195,12 +209,11 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
* duplicate processing.
* <p>
* <b>The supplied FileLocker must be thread safe</b>
*
* @param locker a locker
*/
public void setLocker(FileLocker locker) {
Assert.notNull(locker, "'fileLocker' must not be null.");
this.scanner.setLocker(locker);
this.locker = locker;
}
/**
@@ -240,6 +253,15 @@ 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.filter != null || this.locker != null)),
"The 'filter' and 'locker' options must be present on the provided external 'scanner': "
+ this.scanner);
if (this.filter != null) {
this.scanner.setFilter(this.filter);
}
if (this.locker != null) {
this.scanner.setLocker(this.locker);
}
}
public Message<File> receive() throws MessagingException {
@@ -263,11 +285,6 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
if (logger.isInfoEnabled()) {
logger.info("Created message: [" + message + "]");
}
FileMessageHolder resource = this.resources.get();
if (resource == null) {
this.resources.set(new FileMessageHolder());
}
this.resources.get().setMessage(message);
}
return message;
}

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

@@ -39,8 +39,8 @@ public class HeadDirectoryScanner extends DefaultDirectoryScanner {
public HeadDirectoryScanner(int maxNumberOfFiles) {
HeadFilter headFilter = new HeadFilter(maxNumberOfFiles);
this.setFilter(headFilter);
this.headFilter = headFilter;
this.setFilter(headFilter);
}
@Override

15
spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java
View File

@@ -36,6 +36,7 @@ import org.springframework.util.xml.DomUtils;
* @author Mark Fisher
* @author Gary Russell
* @author Gunnar Hillert
* @author Artem Bilan
*/
public class FileInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
@@ -50,10 +51,12 @@ public class FileInboundChannelAdapterParser extends AbstractPollingInboundChann
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "queue-size");
String filterBeanName = this.registerFilter(element, parserContext);
String lockerBeanName = registerLocker(element, parserContext);
if (filterBeanName != null) {
builder.addPropertyReference("filter", filterBeanName);
}
if (lockerBeanName != null) {
builder.addPropertyReference("locker", lockerBeanName);
}
builder.addPropertyReference("filter", filterBeanName);
return builder.getBeanDefinition();
}
@@ -77,11 +80,18 @@ public class FileInboundChannelAdapterParser extends AbstractPollingInboundChann
}
private String registerFilter(Element element, ParserContext parserContext) {
String filenamePattern = element.getAttribute("filename-pattern");
String filenameRegex = element.getAttribute("filename-regex");
String preventDuplicates = element.getAttribute("prevent-duplicates");
String ignoreHidden = element.getAttribute("ignore-hidden");
if (!StringUtils.hasText(filenamePattern) && !StringUtils.hasText(filenameRegex)
&& !StringUtils.hasText(preventDuplicates) && !StringUtils.hasText(ignoreHidden)) {
return null;
}
BeanDefinitionBuilder factoryBeanBuilder =
BeanDefinitionBuilder.genericBeanDefinition(FileListFilterFactoryBean.class);
factoryBeanBuilder.setRole(BeanDefinition.ROLE_SUPPORT);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(factoryBeanBuilder, element, "filter");
String filenamePattern = element.getAttribute("filename-pattern");
if (StringUtils.hasText(filenamePattern)) {
if (element.hasAttribute("filter")) {
parserContext.getReaderContext().error(
@@ -89,7 +99,6 @@ public class FileInboundChannelAdapterParser extends AbstractPollingInboundChann
}
factoryBeanBuilder.addPropertyValue("filenamePattern", filenamePattern);
}
String filenameRegex = element.getAttribute("filename-regex");
if (StringUtils.hasText(filenameRegex)) {
if (element.hasAttribute("filter")) {
parserContext.getReaderContext().error(

11
spring-integration-file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 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.
@@ -123,7 +123,7 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean<FileRead
return true;
}
private void initSource() {
private void initSource() throws Exception {
synchronized (this.initializationMonitor) {
if (this.source != null) {
return;
@@ -158,6 +158,13 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean<FileRead
this.source.setLocker(locker);
}
}
else if (this.locker != null) {
CompositeFileListFilter<File> compositeFileListFilter = new CompositeFileListFilter<File>();
compositeFileListFilter.addFilter(new FileListFilterFactoryBean().getObject());
compositeFileListFilter.addFilter(this.locker);
this.source.setFilter(compositeFileListFilter);
this.source.setLocker(locker);
}
if (this.scanEachPoll != null) {
this.source.setScanEachPoll(this.scanEachPoll);
}

192
spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.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.
@@ -37,123 +37,127 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
import org.springframework.messaging.Message;
/**
* @author Iwein Fuld
* @author Mark Fisher
* @author Artem Bilan
*/
@RunWith(MockitoJUnitRunner.class)
public class FileReadingMessageSourceTests {
private FileReadingMessageSource source;
private FileReadingMessageSource source;
@Mock
private File inputDirectoryMock;
@Mock
private File inputDirectoryMock;
@Mock
private File fileMock;
@Mock
private File fileMock;
@Mock
private FileLocker locker;
@Mock
private FileLocker locker;
@Mock
private Comparator<File> comparator;
@Mock
private Comparator<File> comparator;
public void prepResource() throws Exception {
when(inputDirectoryMock.exists()).thenReturn(true);
when(inputDirectoryMock.canRead()).thenReturn(true);
when(locker.lock(isA(File.class))).thenReturn(true);
}
public void prepResource() throws Exception {
when(inputDirectoryMock.isDirectory()).thenReturn(true);
when(inputDirectoryMock.exists()).thenReturn(true);
when(inputDirectoryMock.canRead()).thenReturn(true);
when(locker.lock(isA(File.class))).thenReturn(true);
}
@Before
public void initialize() throws Exception {
prepResource();
this.source = new FileReadingMessageSource(comparator);
source.setDirectory(inputDirectoryMock);
source.setLocker(locker);
}
@Before
public void initialize() throws Exception {
prepResource();
this.source = new FileReadingMessageSource(comparator);
this.source.setDirectory(inputDirectoryMock);
this.source.setLocker(locker);
this.source.afterPropertiesSet();
}
@Test
public void straightProcess() throws Exception {
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock});
assertThat(source.receive().getPayload(), is(fileMock));
}
@Test
public void straightProcess() throws Exception {
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock});
assertThat(source.receive().getPayload(), is(fileMock));
}
@Test
public void requeueOnFailure() throws Exception {
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock});
Message<File> received = source.receive();
assertNotNull(received);
source.onFailure(received);
assertEquals(received.getPayload(), source.receive().getPayload());
verify(inputDirectoryMock, times(1)).listFiles();
}
@Test
public void requeueOnFailure() throws Exception {
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock});
Message<File> received = source.receive();
assertNotNull(received);
source.onFailure(received);
assertEquals(received.getPayload(), source.receive().getPayload());
verify(inputDirectoryMock, times(1)).listFiles();
}
@Test
public void scanEachPoll() throws Exception {
File anotherFileMock = mock(File.class);
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock, anotherFileMock});
source.setScanEachPoll(true);
assertNotNull(source.receive());
assertNotNull(source.receive());
assertNull(source.receive());
verify(inputDirectoryMock, times(3)).listFiles();
}
@Test
public void scanEachPoll() throws Exception {
File anotherFileMock = mock(File.class);
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock, anotherFileMock});
source.setScanEachPoll(true);
assertNotNull(source.receive());
assertNotNull(source.receive());
assertNull(source.receive());
verify(inputDirectoryMock, times(3)).listFiles();
}
@Test
public void noDuplication() throws Exception {
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock});
Message<File> received = source.receive();
assertNotNull(received);
assertEquals(fileMock, received.getPayload());
assertNull(source.receive());
verify(inputDirectoryMock, times(2)).listFiles();
}
@Test
public void noDuplication() throws Exception {
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock});
Message<File> received = source.receive();
assertNotNull(received);
assertEquals(fileMock, received.getPayload());
assertNull(source.receive());
verify(inputDirectoryMock, times(2)).listFiles();
}
@Test(expected = IllegalArgumentException.class)
public void nullFilter() throws Exception {
source.setFilter(null);
}
@Test(expected = IllegalArgumentException.class)
public void nullFilter() throws Exception {
source.setFilter(null);
}
@Test
public void lockIsAquired() throws IOException {
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock});
Message<File> received = source.receive();
assertNotNull(received);
assertEquals(fileMock, received.getPayload());
verify(locker).lock(fileMock);
}
@Test
public void lockIsAcquired() throws IOException {
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock});
Message<File> received = source.receive();
assertNotNull(received);
assertEquals(fileMock, received.getPayload());
verify(locker).lock(fileMock);
}
@Test
public void lockedFilesAreIgnored() throws IOException {
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock});
when(locker.lock(fileMock)).thenReturn(false);
Message<File> received = source.receive();
assertNull(received);
verify(locker).lock(fileMock);
}
@Test
public void lockedFilesAreIgnored() throws IOException {
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{fileMock});
when(locker.lock(fileMock)).thenReturn(false);
Message<File> received = source.receive();
assertNull(received);
verify(locker).lock(fileMock);
}
@Test
public void orderedReception() throws Exception {
File file1 = mock(File.class);
File file2 = mock(File.class);
File file3 = mock(File.class);
@Test
public void orderedReception() throws Exception {
File file1 = mock(File.class);
File file2 = mock(File.class);
File file3 = mock(File.class);
// record the comparator to reverse order the files
when(comparator.compare(file1, file2)).thenReturn(1);
when(comparator.compare(file1, file3)).thenReturn(1);
when(comparator.compare(file2, file3)).thenReturn(1);
when(comparator.compare(file2, file1)).thenReturn(-1);
when(comparator.compare(file3, file1)).thenReturn(-1);
when(comparator.compare(file3, file2)).thenReturn(-1);
// record the comparator to reverse order the files
when(comparator.compare(file1, file2)).thenReturn(1);
when(comparator.compare(file1, file3)).thenReturn(1);
when(comparator.compare(file2, file3)).thenReturn(1);
when(comparator.compare(file2, file1)).thenReturn(-1);
when(comparator.compare(file3, file1)).thenReturn(-1);
when(comparator.compare(file3, file2)).thenReturn(-1);
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{file2, file3, file1});
assertSame(file3, source.receive().getPayload());
assertSame(file2, source.receive().getPayload());
assertSame(file1, source.receive().getPayload());
assertNull(source.receive());
verify(inputDirectoryMock, times(2)).listFiles();
}
when(inputDirectoryMock.listFiles()).thenReturn(new File[]{file2, file3, file1});
assertSame(file3, source.receive().getPayload());
assertSame(file2, source.receive().getPayload());
assertSame(file1, source.receive().getPayload());
assertNull(source.receive());
verify(inputDirectoryMock, times(2)).listFiles();
}
}

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

@@ -1,20 +1,4 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ 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.
~ You may obtain a copy of the License at
~
~ http://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing, software
~ distributed under the License is distributed on an "AS IS" BASIS,
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
~ See the License for the specific language governing permissions and
~ limitations under the License.
-->
<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"
@@ -33,8 +17,6 @@
<integration:poller fixed-rate="1000" />
</inbound-channel-adapter>
<beans:bean class="org.springframework.integration.file.config.FileListFilterFactoryBean" />
<beans:bean id="filter" class="org.springframework.integration.file.filters.CompositeFileListFilter">
<beans:constructor-arg>
<beans:list />