INT-1073: Added HeadDirectoryScanner as an answer to the queue overflow problem

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 {
 

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<File>,
     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<File>,
      *                              of large numbers of files in a directory.
      */
     public FileReadingMessageSource(int internalQueueCapacity) {
-        toBeReceived = new ArrayBlockingQueue<File>(
-                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));
     }
 
     /**

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 <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.
+ *
+ * @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<File> filterFiles(File[] files) {
+            return Arrays.asList(files).subList(0, Math.min(files.length, maxNumberOfFiles));
+        }
+    }
+}

org.springframework.integration.file/src/main/resources/org/springframework/integration/file/config/spring-integration-file-2.0.xsd

@@ -100,14 +100,13 @@
             </xsd:attribute>
             <xsd:attribute name="queue-size" type="xsd:integer">
                 <xsd:annotation>
-                    <xsd:documentation>
-                        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.
-                    </xsd:documentation>
+                  <xsd:documentation>
+                    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.
+                  </xsd:documentation>
                 </xsd:annotation>
             </xsd:attribute>
         </xsd:complexType>

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));
     }
 }