From ce479439edd902968664b478291801083f36035c Mon Sep 17 00:00:00 2001 From: Iwein Fuld Date: Fri, 9 Apr 2010 12:30:01 +0000 Subject: [PATCH] INT-1073: Added HeadDirectoryScanner as an answer to the queue overflow problem --- .../file/AcceptOnceFileListFilter.java | 3 +- .../file/FileReadingMessageSource.java | 13 ++--- .../file/HeadDirectoryScanner.java | 47 +++++++++++++++++++ .../config/spring-integration-file-2.0.xsd | 15 +++--- ...boundChannelAdapterWithQueueSizeTests.java | 9 ++-- 5 files changed, 66 insertions(+), 21 deletions(-) create mode 100644 org.springframework.integration.file/src/main/java/org/springframework/integration/file/HeadDirectoryScanner.java diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/AcceptOnceFileListFilter.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/AcceptOnceFileListFilter.java index fad4f5bef5..576d168726 100644 --- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/AcceptOnceFileListFilter.java +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/AcceptOnceFileListFilter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2010 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. @@ -28,6 +28,7 @@ import java.util.concurrent.LinkedBlockingQueue; * This implementation is thread safe. * * @author Iwein Fuld + * @since 1.0.0 */ public class AcceptOnceFileListFilter extends AbstractFileListFilter { diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java index 532bb8b34a..8438a23696 100644 --- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java @@ -29,7 +29,6 @@ import org.springframework.util.Assert; import java.io.File; import java.util.*; -import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.PriorityBlockingQueue; /** @@ -76,14 +75,15 @@ public class FileReadingMessageSource implements MessageSource, private boolean scanEachPoll = false; /** - * Creates a FileReadingMessageSource with a naturally ordered queue of default capacity. + * Creates a FileReadingMessageSource with a naturally ordered queue of unbounded capacity. */ public FileReadingMessageSource() { - this(-1); + this(null); } /** - * Creates a FileReadingMessageSource with a bounded queue of the given capacity. + * Creates a FileReadingMessageSource with a bounded queue of the given capacity. This can be used to reduce the + * memory footprint of this component when reading from a large directory. * * @param internalQueueCapacity the size of the queue used to cache files to be received internally. This queue can * be made larger to optimize the directory scanning. With scanEachPoll set to false @@ -92,8 +92,9 @@ public class FileReadingMessageSource implements MessageSource, * of large numbers of files in a directory. */ public FileReadingMessageSource(int internalQueueCapacity) { - toBeReceived = new ArrayBlockingQueue( - internalQueueCapacity < 0 ? DEFAULT_INTERNAL_QUEUE_CAPACITY : internalQueueCapacity); + this(null); + Assert.isTrue(internalQueueCapacity>0, "Cannot create a queue with non positive capacity"); + this.setScanner(new HeadDirectoryScanner(internalQueueCapacity)); } /** diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/HeadDirectoryScanner.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/HeadDirectoryScanner.java new file mode 100644 index 0000000000..7f761687cd --- /dev/null +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/HeadDirectoryScanner.java @@ -0,0 +1,47 @@ +/* + * Copyright 2002-2010 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 java.util.Arrays; +import java.util.List; + +/** + * 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. + * + * @author Iwein Fuld + * @since 2.0.0 + */ +public class HeadDirectoryScanner extends DefaultDirectoryScanner { + + public HeadDirectoryScanner(int maxNumberOfFiles) { + this.setFilter(new HeadFilter(maxNumberOfFiles)); + } + + private class HeadFilter implements FileListFilter { + private final int maxNumberOfFiles; + + public HeadFilter(int maxNumberOfFiles) { + this.maxNumberOfFiles = maxNumberOfFiles; + } + + public List filterFiles(File[] files) { + return Arrays.asList(files).subList(0, Math.min(files.length, maxNumberOfFiles)); + } + } +} diff --git a/org.springframework.integration.file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-2.0.xsd b/org.springframework.integration.file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-2.0.xsd index 73c2f0ef6e..cf960a43d2 100644 --- a/org.springframework.integration.file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-2.0.xsd +++ b/org.springframework.integration.file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-2.0.xsd @@ -100,14 +100,13 @@ - - Specify the queue size used internally by the underlying FileReadingMessageSource to store files - listed on a poll. A larger queue size reduces the number of directory listings needed, but it - increases the chances of the internal queue being out of whack with the actual files listed in - the directory. Use 0 for small but volatile directories, use a large number for large - directories that are only written to. MUTUALLY EXCLUSIVE with comparator, if comparator is set - this attribute will be ignored. - + + Specify the maximum number of files stored in memory by the underlying FileReadingMessageSource. + This is useful to limit the memory footprint of this endpoint. Using a stateful filter would counter + this benefit, so AcceptOnceFileListFilter is not used when this attribute is specified. + MUTUALLY EXCLUSIVE with comparator, if comparator is set this attribute will be ignored. + MUTUALLY EXCLUSIVE with stateful filtering. + diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests.java index fce9daab0c..562159428d 100644 --- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests.java +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests.java @@ -22,11 +22,10 @@ import org.junit.runner.RunWith; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.file.FileReadingMessageSource; +import org.springframework.integration.file.HeadDirectoryScanner; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import java.util.concurrent.BlockingQueue; - import static org.hamcrest.CoreMatchers.is; import static org.junit.Assert.assertThat; @@ -49,9 +48,7 @@ public class FileInboundChannelAdapterWithQueueSizeTests { @Test public void queueSize() { - Object queue = accessor.getPropertyValue("toBeReceived"); - assertThat(queue, is(BlockingQueue.class)); - BlockingQueue blockingQueue = (BlockingQueue) queue; - assertThat(blockingQueue.remainingCapacity()+blockingQueue.size(), is(30)); + Object scanner = accessor.getPropertyValue("scanner"); + assertThat(scanner, is(HeadDirectoryScanner.class)); } }