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

INT-985: made sorting and bounding mutually exclusive

Browse Source
This commit is contained in:
Iwein Fuld
2010-02-15 09:34:38 +00:00
parent c95763934e
commit f92eee81b3
8 changed files with 187 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
org.springframework.integration.file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
View File

@@ -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<File>,
}
/**
* 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<File>(
toBeReceived = new ArrayBlockingQueue<File>(
internalQueueCapacity < 0 ? DEFAULT_INTERNAL_QUEUE_CAPACITY : internalQueueCapacity);
}
@@ -101,34 +101,14 @@ public class FileReadingMessageSource implements MessageSource<File>,
* Comparator}
* <p/>
* 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.
* <p/>
*
* @param receptionOrderComparator the comparator to be used to order the files in the internal queue
*/
public FileReadingMessageSource(Comparator<File> receptionOrderComparator) {
this(receptionOrderComparator, -1);
}
/**
* Creates a FileReadingMessageSource with a {@link PriorityBlockingQueue} ordered with the passed in {@link
* Comparator}
* <p/>
* 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.
* <p/>
*
* @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<File> receptionOrderComparator, int internalQueueCapacity) {
toBeReceived = new PriorityBlockingQueue<File>(
internalQueueCapacity < 0 ? DEFAULT_INTERNAL_QUEUE_CAPACITY : internalQueueCapacity,
toBeReceived = new PriorityBlockingQueue<File>(DEFAULT_INTERNAL_QUEUE_CAPACITY,
receptionOrderComparator);
}
@@ -185,9 +165,9 @@ public class FileReadingMessageSource implements MessageSource<File>,
* <p/>
* 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 <code>false</code>, but it will change more often (causing expensive reordering) if it is set to
* <code>true</code>.
* 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 <code>false</code>, but it will change more often (causing expensive reordering) if
* it is set to <code>true</code>.
*/
public void setScanEachPoll(boolean scanEachPoll) {
this.scanEachPoll = scanEachPoll;

1
org.springframework.integration.file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java
View File

@@ -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) {

26
org.springframework.integration.file/src/main/java/org/springframework/integration/file/config/FileReadingMessageSourceFactoryBean.java
View File

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

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

@@ -21,8 +21,8 @@
<xsd:element name="inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
@@ -48,7 +48,7 @@
<xsd:annotation>
<xsd:documentation><![CDATA[
Specify a Comparator to be used when ordering Files. If none is provided, the
order will be determined by the java.io.File implementation of Comparable.
order will be determined by the java.io.File implementation of Comparable. MUTUALLY EXCLUSIVE with queue-size.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -98,6 +98,18 @@
</xsd:documentation>
</xsd:annotation>
</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:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
@@ -128,8 +140,8 @@
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
<xsd:complexContent>
@@ -193,9 +205,9 @@
<xsd:attribute name="auto-create-directory" type="xsd:string" default="true">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>

42
org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests-context.xml
View File

@@ -1,34 +1,34 @@
<?xml version="1.0" encoding="UTF-8"?>
<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"
xsi:schemaLocation="http://www.springframework.org/schema/beans
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"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/file
http://www.springframework.org/schema/integration/file/spring-integration-file.xsd">
<inbound-channel-adapter id="inputDirPoller"
directory="${java.io.tmpdir}"
filter="filter"
comparator="testComparator"
auto-startup="false">
<integration:poller>
<integration:interval-trigger interval="5000"/>
</integration:poller>
</inbound-channel-adapter>
<inbound-channel-adapter id="inputDirPoller"
directory="${java.io.tmpdir}"
filter="filter"
comparator="testComparator"
auto-startup="false">
<integration:poller>
<integration:interval-trigger interval="5000"/>
</integration:poller>
</inbound-channel-adapter>
<beans:bean id="filter" class="org.springframework.integration.file.CompositeFileListFilter">
<beans:constructor-arg>
<beans:list></beans:list>
</beans:constructor-arg>
</beans:bean>
<beans:bean id="filter" class="org.springframework.integration.file.CompositeFileListFilter">
<beans:constructor-arg>
<beans:list></beans:list>
</beans:constructor-arg>
</beans:bean>
<beans:bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer" />
<beans:bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"/>
<beans:bean id="testComparator"
class="org.springframework.integration.file.config.FileInboundChannelAdapterParserTests$TestComparator"/>
<beans:bean id="testComparator"
class="org.springframework.integration.file.config.FileInboundChannelAdapterParserTests$TestComparator"/>
</beans:beans>

7
org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterParserTests.java
View File

@@ -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<File> {
public int compare(File f1, File f2) {

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

@@ -0,0 +1,50 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ 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.
-->
<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"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/file
http://www.springframework.org/schema/integration/file/spring-integration-file.xsd">
<inbound-channel-adapter id="inputDirPoller"
directory="${java.io.tmpdir}"
filter="filter"
queue-size="30"
auto-startup="false">
<integration:poller>
<integration:interval-trigger interval="5000"/>
</integration:poller>
</inbound-channel-adapter>
<beans:bean id="filter" class="org.springframework.integration.file.CompositeFileListFilter">
<beans:constructor-arg>
<beans:list></beans:list>
</beans:constructor-arg>
</beans:bean>
<beans:bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"/>
<beans:bean id="testComparator"
class="org.springframework.integration.file.config.FileInboundChannelAdapterParserTests$TestComparator"/>
</beans:beans>

57
org.springframework.integration.file/src/test/java/org/springframework/integration/file/config/FileInboundChannelAdapterWithQueueSizeTests.java Normal file
View File

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