From 34c51eb635b49405d163cf245a10f9eb92de7d8f Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 23 Nov 2011 14:49:46 -0500 Subject: [PATCH] INT-2235-updated added default filter as AcceptOnceCollectionFilter, chanhged parser to allow filter= to allow null filter --- .../ResourceInboundChannelAdapterParser.java | 15 +++- .../util/AcceptOnceCollectionFilter.java | 46 ++++++++++ ...AcceptOnceUntilPurgedCollectionFilter.java | 89 ------------------- ...ourceInboundChannelAdapterParserTests.java | 27 ++++-- ...ePatternResolver-config-usage-emptyref.xml | 19 ++++ 5 files changed, 101 insertions(+), 95 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/util/AcceptOnceCollectionFilter.java delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/util/AcceptOnceUntilPurgedCollectionFilter.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/resource/ResourcePatternResolver-config-usage-emptyref.xml diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResourceInboundChannelAdapterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResourceInboundChannelAdapterParser.java index 7945013557..3896268c8a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResourceInboundChannelAdapterParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ResourceInboundChannelAdapterParser.java @@ -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(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/AcceptOnceCollectionFilter.java b/spring-integration-core/src/main/java/org/springframework/integration/util/AcceptOnceCollectionFilter.java new file mode 100644 index 0000000000..6e6f064896 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/AcceptOnceCollectionFilter.java @@ -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 implements CollectionFilter { + + private volatile Collection lastSeenElements = Collections.emptyList(); + + public synchronized Collection filter(Collection unfilteredElements) { + List filteredElements = new ArrayList(); + for (T element : unfilteredElements) { + if (!this.lastSeenElements.contains(element)) { + filteredElements.add(element); + } + } + this.lastSeenElements = unfilteredElements; + return filteredElements; + } + +} \ No newline at end of file diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/AcceptOnceUntilPurgedCollectionFilter.java b/spring-integration-core/src/main/java/org/springframework/integration/util/AcceptOnceUntilPurgedCollectionFilter.java deleted file mode 100644 index a76280be43..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/AcceptOnceUntilPurgedCollectionFilter.java +++ /dev/null @@ -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 implements CollectionFilter { - - private final Log logger = LogFactory.getLog(this.getClass()); - - private final Queue seenItems; - - private final Object seenQueueMonitor = new Object(); - - - public AcceptOnceUntilPurgedCollectionFilter() { - this(Integer.MAX_VALUE); - } - - public AcceptOnceUntilPurgedCollectionFilter(int maxCapacity) { - this.seenItems = new LinkedBlockingQueue(maxCapacity); - } - - - public Collection filter(Collection unfilteredElements) { - Assert.notNull(unfilteredElements, "'unfilteredElements' must not be null"); - List filteredElements = new ArrayList(); - 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; - } - } - -} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/resource/ResourceInboundChannelAdapterParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/resource/ResourceInboundChannelAdapterParserTests.java index bb3028fe6d..4061ccc24b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/resource/ResourceInboundChannelAdapterParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/resource/ResourceInboundChannelAdapterParserTests.java @@ -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 { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/resource/ResourcePatternResolver-config-usage-emptyref.xml b/spring-integration-core/src/test/java/org/springframework/integration/resource/ResourcePatternResolver-config-usage-emptyref.xml new file mode 100644 index 0000000000..7e02fcc05e --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/resource/ResourcePatternResolver-config-usage-emptyref.xml @@ -0,0 +1,19 @@ + + + + + + + + + + + + +