diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileMessageHolder.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileMessageHolder.java deleted file mode 100644 index e629a07715..0000000000 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileMessageHolder.java +++ /dev/null @@ -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;} 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 message; - - Message getMessage() { - return message; - } - - void setMessage(Message message) { - this.message = message; - } - -} 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 2fc78e1654..e5fb8a3e80 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 @@ -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. *

* 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 { @@ -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 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 resources = new ThreadLocal(); + private FileListFilter 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. *

* The supplied filter must be thread safe.. - * * @param filter a filter */ public void setFilter(FileListFilter 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. *

* The supplied FileLocker must be thread safe - * * @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 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; } 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 443554d057..38d8536922 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 @@ -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 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 5897d08c0a..0a6383ae3b 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 @@ -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( 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 7097760c32..a4c4d5d14e 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 @@ -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 compositeFileListFilter = new CompositeFileListFilter(); + 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); } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java index 3a4e45b92e..fcb1f1a63a 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceTests.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. @@ -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 comparator; + @Mock + private Comparator 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 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 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 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 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 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 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 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 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(); + } } 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 bed533a311..a9b5651106 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,20 +1,4 @@ - - - -