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

Merge pull request #209 from olegz/INT-2235-updated

Browse Source
This commit is contained in:
Mark Fisher
2011-11-23 15:08:19 -05:00
parent 6fbdd71aae 34c51eb635
commit cadbed3523
5 changed files with 101 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResourceInboundChannelAdapterParser.java
View File

@@ -20,6 +20,9 @@ import org.springframework.beans.BeanMetadataElement;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.resource.ResourceRetrievingMessageSource;
import org.springframework.integration.util.AcceptOnceCollectionFilter;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
@@ -36,7 +39,17 @@ public class ResourceInboundChannelAdapterParser extends AbstractPollingInboundC
BeanDefinitionBuilder sourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(ResourceRetrievingMessageSource.class);
sourceBuilder.addConstructorArgValue(element.getAttribute("pattern"));
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(sourceBuilder, element, "pattern-resolver");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(sourceBuilder, element, "filter");
boolean hasFilter = element.hasAttribute("filter");
if (hasFilter){
String filterValue = element.getAttribute("filter");
if (StringUtils.hasText(filterValue)){
sourceBuilder.addPropertyReference("filter", filterValue);
}
}
else {
BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition(AcceptOnceCollectionFilter.class);
sourceBuilder.addPropertyValue("filter", filterBuilder.getBeanDefinition());
}
return sourceBuilder.getBeanDefinition();
}

46
spring-integration-core/src/main/java/org/springframework/integration/util/AcceptOnceCollectionFilter.java Normal file
View File

@@ -0,0 +1,46 @@
/*
* Copyright 2002-2011 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.util;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
/**
* An implementation of {@link CollectionFilter} that remembers the elements passed in
* the previous invocation in order to avoid returning those elements more than once.
*
* @author Mark Fisher
* @since 2.1
*/
public class AcceptOnceCollectionFilter<T> implements CollectionFilter<T> {
private volatile Collection<T> lastSeenElements = Collections.emptyList();
public synchronized Collection<T> filter(Collection<T> unfilteredElements) {
List<T> filteredElements = new ArrayList<T>();
for (T element : unfilteredElements) {
if (!this.lastSeenElements.contains(element)) {
filteredElements.add(element);
}
}
this.lastSeenElements = unfilteredElements;
return filteredElements;
}
}

89
spring-integration-core/src/main/java/org/springframework/integration/util/AcceptOnceUntilPurgedCollectionFilter.java
View File

@@ -1,89 +0,0 @@
/*
* Copyright 2002-2011 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.util;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.util.Assert;
/**
* An implementation of {@link CollectionFilter} which will queue all items that have been seen until
* the queue reaches its capacity after which one item from the queue will be purged to make room for a
* new item to be added. Note that however unlikely the removed item will now appear as unprocessed
* so it is highly recommended to move/delete resources which correspond to the underlying items once processing
* is done to eliminate duplicate processing.
*
* @author Oleg Zhurakousky
* @author Mark Fisher
* @since 2.1
*/
public class AcceptOnceUntilPurgedCollectionFilter<T> implements CollectionFilter<T> {
private final Log logger = LogFactory.getLog(this.getClass());
private final Queue<T> seenItems;
private final Object seenQueueMonitor = new Object();
public AcceptOnceUntilPurgedCollectionFilter() {
this(Integer.MAX_VALUE);
}
public AcceptOnceUntilPurgedCollectionFilter(int maxCapacity) {
this.seenItems = new LinkedBlockingQueue<T>(maxCapacity);
}
public Collection<T> filter(Collection<T> unfilteredElements) {
Assert.notNull(unfilteredElements, "'unfilteredElements' must not be null");
List<T> filteredElements = new ArrayList<T>();
if (unfilteredElements.size() > 0) {
for (T element : unfilteredElements) {
if (this.accept(element)) {
filteredElements.add(element);
}
}
}
return filteredElements;
}
private boolean accept(T item) {
synchronized (this.seenQueueMonitor) {
boolean accepted = false;
if (!this.seenItems.contains(item)) {
accepted = this.seenItems.offer(item);
if (!accepted){
logger.warn("'seenQueueMonitor' queue of AcceptOnceUntilPurgedElementFilter is at the capacity, " +
"evicting one item to make room for another");
this.seenItems.poll();
accepted = this.seenItems.offer(item);
}
}
return accepted;
}
}
}

27
spring-integration-core/src/test/java/org/springframework/integration/resource/ResourceInboundChannelAdapterParserTests.java
View File

@@ -16,11 +16,6 @@
package org.springframework.integration.resource;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.io.File;
import java.util.Collection;
@@ -36,6 +31,12 @@ import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.util.CollectionFilter;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
/**
* @author Oleg Zhurakousky
* @since 2.1
@@ -114,6 +115,22 @@ public class ResourceInboundChannelAdapterParserTests {
assertNotNull(message);
assertTrue(customFilter.invoked);
}
@Test
public void testUsageWithEmptyFilter() throws Exception{
File baseDir = new File(System.getProperty("java.io.tmpdir"));
for (int i = 0; i < 10; i++) {
File f = new File(baseDir, "testUsageWithRf"+i);
f.createNewFile();
}
ApplicationContext context = new ClassPathXmlApplicationContext("ResourcePatternResolver-config-usage-emptyref.xml", this.getClass());
SourcePollingChannelAdapter resourceAdapter = context.getBean("resourceAdapterDefault", SourcePollingChannelAdapter.class);
ResourceRetrievingMessageSource source = TestUtils.getPropertyValue(resourceAdapter, "source", ResourceRetrievingMessageSource.class);
assertNotNull(source);
assertNull(TestUtils.getPropertyValue(source, "filter"));
}
public static class TestCollectionFilter implements CollectionFilter<Resource> {

19
spring-integration-core/src/test/java/org/springframework/integration/resource/ResourcePatternResolver-config-usage-emptyref.xml Normal file
View File

@@ -0,0 +1,19 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="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-2.1.xsd">
<int:resource-inbound-channel-adapter id="resourceAdapterDefault" channel="resultChannel"
pattern="file:#{T(java.lang.System).getProperty('java.io.tmpdir') + T(java.lang.System).getProperty('file.separator') + 'testUsage*'}"
filter="" auto-startup="false">
<int:poller fixed-rate="500"/>
</int:resource-inbound-channel-adapter>
<int:channel id="resultChannel">
<int:queue/>
</int:channel>
</beans>