From f92eee81b3d2666834fa282c1d828f4ed257a0c1 Mon Sep 17 00:00:00 2001 From: Iwein Fuld Date: Mon, 15 Feb 2010 09:34:38 +0000 Subject: [PATCH] INT-985: made sorting and bounding mutually exclusive --- .../file/FileReadingMessageSource.java | 46 +++++---------- .../FileInboundChannelAdapterParser.java | 1 + .../FileReadingMessageSourceFactoryBean.java | 26 ++++++++- .../config/spring-integration-file-2.0.xsd | 28 ++++++--- ...boundChannelAdapterParserTests-context.xml | 42 +++++++------- .../FileInboundChannelAdapterParserTests.java | 7 +-- ...annelAdapterWithQueueSizeTests-context.xml | 50 ++++++++++++++++ ...boundChannelAdapterWithQueueSizeTests.java | 57 +++++++++++++++++++ 8 files changed, 187 insertions(+), 70 deletions(-) create mode 100644 org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests-context.xml create mode 100644 org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests.java 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 0d1437ac0b..532bb8b34a 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,6 +29,7 @@ import org.springframework.util.Assert; import java.io.File; import java.util.*; +import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.PriorityBlockingQueue; /** @@ -82,17 +83,16 @@ public class FileReadingMessageSource implements MessageSource, } /** - * Creates a FileReadingMessageSource with a naturally ordered queue of the given capacity. + * Creates a FileReadingMessageSource with a bounded queue of the given capacity. * * @param internalQueueCapacity the size of the queue used to cache files to be received internally. This queue can - * be made larger to improve the ordering of incoming files and more importantly to - * optimize the directory scanning. With scanEachPoll set to false and the queue to a - * large size, it will be filled once and then completely emptied before a new - * directory listing is done. This is particularly useful to reduce scans of large - * numbers of files in a directory. + * be made larger to optimize the directory scanning. With scanEachPoll set to false + * and the queue to a large size, it will be filled once and then completely emptied + * before a new directory listing is done. This is particularly useful to reduce scans + * of large numbers of files in a directory. */ public FileReadingMessageSource(int internalQueueCapacity) { - toBeReceived = new PriorityBlockingQueue( + toBeReceived = new ArrayBlockingQueue( internalQueueCapacity < 0 ? DEFAULT_INTERNAL_QUEUE_CAPACITY : internalQueueCapacity); } @@ -101,34 +101,14 @@ public class FileReadingMessageSource implements MessageSource, * Comparator} *

* The size of the queue used should be large enough to hold all the files in the input directory in order to sort - * all of them. No guarantees about file delivery order can be made under concurrent access. + * all of them, so restricting the size of the queue is mutually exclusive with ordering. No guarantees about file + * delivery order can be made under concurrent access. *

* * @param receptionOrderComparator the comparator to be used to order the files in the internal queue */ public FileReadingMessageSource(Comparator receptionOrderComparator) { - this(receptionOrderComparator, -1); - } - - /** - * Creates a FileReadingMessageSource with a {@link PriorityBlockingQueue} ordered with the passed in {@link - * Comparator} - *

- * The size of the queue used should be large enough to hold all the files in the input directory in order to sort - * all of them. No guarantees about file delivery order can be made under concurrent access. - *

- * - * @param receptionOrderComparator the comparator to be used to order the files in the internal queue - * @param internalQueueCapacity the size of the queue used to cache files to be received internally. This queue - * can be made larger to improve the ordering of incoming files and more importantly - * to optimize the directory scanning. With scanEachPoll set to false and the queue - * to a large size, it will be filled once and then completely emptied before a new - * directory listing is done. This is particularly useful to reduce scans of large - * numbers of files in a directory. - */ - public FileReadingMessageSource(Comparator receptionOrderComparator, int internalQueueCapacity) { - toBeReceived = new PriorityBlockingQueue( - internalQueueCapacity < 0 ? DEFAULT_INTERNAL_QUEUE_CAPACITY : internalQueueCapacity, + toBeReceived = new PriorityBlockingQueue(DEFAULT_INTERNAL_QUEUE_CAPACITY, receptionOrderComparator); } @@ -185,9 +165,9 @@ public class FileReadingMessageSource implements MessageSource, *

* By default this implementation will empty its queue before looking at the directory again. In cases where order * is relevant it is important to consider the effects of setting this flag. The internal {@link - * PriorityBlockingQueue} that this class is keeping will more likely be out of sync with the file system if this - * flag is set to false, but it will change more often (causing expensive reordering) if it is set to - * true. + * java.util.concurrent.BlockingQueue} that this class is keeping will more likely be out of sync with the file + * system if this flag is set to false, but it will change more often (causing expensive reordering) if + * it is set to true. */ public void setScanEachPoll(boolean scanEachPoll) { this.scanEachPoll = scanEachPoll; diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java index b1d2a12086..1a2bae92fb 100644 --- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java @@ -45,6 +45,7 @@ public class FileInboundChannelAdapterParser extends AbstractPollingInboundChann IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "scanner"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "directory"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-create-directory"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "queue-size"); String filterBeanName = this.registerFilter(element, parserContext); String lockerBeanName = registerLocker(element, parserContext); if (lockerBeanName != null) { diff --git a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java index aaa8da2dfb..6293bc5697 100644 --- a/org.springframework.integration.file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java +++ b/org.springframework.integration.file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java @@ -16,6 +16,8 @@ package org.springframework.integration.file.config; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.FactoryBean; import org.springframework.integration.file.CompositeFileListFilter; import org.springframework.integration.file.DirectoryScanner; @@ -33,6 +35,8 @@ import java.util.Comparator; */ public class FileReadingMessageSourceFactoryBean implements FactoryBean { + private static Log logger = LogFactory.getLog(FileReadingMessageSourceFactoryBean.class); + private volatile FileReadingMessageSource source; private volatile File directory; @@ -49,6 +53,8 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean { private volatile Boolean autoCreateDirectory; + private volatile Integer queueSize; + private final Object initializationMonitor = new Object(); public void setDirectory(File directory) { @@ -78,6 +84,10 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean { this.autoCreateDirectory = autoCreateDirectory; } + public void setQueueSize(Integer queueSize) { + this.queueSize = queueSize; + } + public void setLocker(AbstractFileLockerFilter locker) { this.locker = locker; } @@ -102,8 +112,19 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean { if (this.source != null) { return; } - this.source = (this.comparator != null) ? - new FileReadingMessageSource(this.comparator) : new FileReadingMessageSource(); + boolean comparatorSet = this.comparator != null; + boolean queueSizeSet = this.queueSize != null; + if (comparatorSet) { + if(queueSizeSet){ + logger.warn("'comparator' and 'queueSize' are mutually exclusive. Ignoring 'queueSize'"); + } + this.source = new FileReadingMessageSource(this.comparator); + } else if ( queueSizeSet) { + this.source = new FileReadingMessageSource(queueSize); + } + else { + this.source = new FileReadingMessageSource(); + } this.source.setDirectory(this.directory); if (this.scanner != null) { this.source.setScanner(this.scanner); @@ -125,5 +146,4 @@ public class FileReadingMessageSourceFactoryBean implements FactoryBean { this.source.afterPropertiesSet(); } } - } 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 b018962c8f..73c2f0ef6e 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 @@ -21,8 +21,8 @@ - Configures an inbound Channel Adapter that polls a directory and sends - Messages whose payloads are instances of java.io.File. + Configures an inbound Channel Adapter that polls a directory and sends Messages whose payloads are + instances of java.io.File. @@ -48,7 +48,7 @@ @@ -98,6 +98,18 @@ + + + + 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. + + + @@ -128,8 +140,8 @@ - Configures an outbound Gateway that writes request Message payloads to a File and - then generates a reply Message containing the newly written File as its payload. + Configures an outbound Gateway that writes request Message payloads to a File and then generates a + reply Message containing the newly written File as its payload. @@ -193,9 +205,9 @@ - Specify whether to automatically create the destination directory if it does not yet exist - when this adapter is being initialized. The default value is 'true'. If set to 'false' and - the directory does not exist upon initialization, an Exception will be thrown. + Specify whether to automatically create the destination directory if it does not yet exist when this + adapter is being initialized. The default value is 'true'. If set to 'false' and the directory does + not exist upon initialization, an Exception will be thrown. diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml index add210d348..1bf152ea65 100644 --- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml @@ -1,34 +1,34 @@ - - - - - + + + + + - - - - - + + + + + - + - + \ No newline at end of file diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java index 88195cc58a..efebb19abd 100644 --- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java @@ -46,18 +46,16 @@ public class FileInboundChannelAdapterParserTests { @Autowired(required=true) private ApplicationContext context; - @Autowired(required=true) + @Autowired private FileReadingMessageSource source; private DirectFieldAccessor accessor; - - @Before + @Before public void init() { accessor = new DirectFieldAccessor(source); } - @Test public void channelName() throws Exception { MessageChannel channel = (MessageChannel) context.getBean("inputDirPoller"); @@ -89,7 +87,6 @@ public class FileInboundChannelAdapterParserTests { assertSame("comparator reference not set, ", expected, actual); } - static class TestComparator implements Comparator { public int compare(File f1, File f2) { diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests-context.xml b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests-context.xml new file mode 100644 index 0000000000..62aa45c0e5 --- /dev/null +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests-context.xml @@ -0,0 +1,50 @@ + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file 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 new file mode 100644 index 0000000000..fce9daab0c --- /dev/null +++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests.java @@ -0,0 +1,57 @@ +/* + * 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.config; + +import org.junit.Before; +import org.junit.Test; +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.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; + +/** + * + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class FileInboundChannelAdapterWithQueueSizeTests { + + @Autowired + FileReadingMessageSource source; + + private DirectFieldAccessor accessor; + + @Before + public void init() { + accessor = new DirectFieldAccessor(source); + } + + @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)); + } +}