From 32d64b19f42d126e86410a05aa74f67988d514b8 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 22 Nov 2011 12:57:48 -0500 Subject: [PATCH] ElementFilter now CollectionFilter Refactored filter to handle a Collection of Resources Made 'pattern' a constructor-arg since it's mandatory --- .../ResourceInboundChannelAdapterParser.java | 3 +- .../resource/ResourceMessageSource.java | 37 +++++++------- .../AcceptOnceUntilPurgedElementFilter.java | 48 +++++++++++-------- ...ementFilter.java => CollectionFilter.java} | 10 ++-- .../ResourcePatternResolverParserTests.java | 32 +++++++------ 5 files changed, 72 insertions(+), 58 deletions(-) rename spring-integration-core/src/main/java/org/springframework/integration/util/{ElementFilter.java => CollectionFilter.java} (76%) 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 ad02416e98..9284805970 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 @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.config.xml; import org.springframework.beans.BeanMetadataElement; @@ -33,7 +34,7 @@ public class ResourceInboundChannelAdapterParser extends AbstractPollingInboundC @Override protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) { BeanDefinitionBuilder sourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(ResourceMessageSource.class); - IntegrationNamespaceUtils.setValueIfAttributeDefined(sourceBuilder, element, "pattern"); + sourceBuilder.addConstructorArgValue(element.getAttribute("pattern")); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(sourceBuilder, element, "pattern-resolver"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(sourceBuilder, element, "filter"); return sourceBuilder.getBeanDefinition(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/resource/ResourceMessageSource.java b/spring-integration-core/src/main/java/org/springframework/integration/resource/ResourceMessageSource.java index 6a9714a187..7b104b7ca5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/resource/ResourceMessageSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/resource/ResourceMessageSource.java @@ -16,8 +16,8 @@ package org.springframework.integration.resource; -import java.util.ArrayList; -import java.util.List; +import java.util.Arrays; +import java.util.Collection; import org.springframework.beans.factory.InitializingBean; import org.springframework.context.ApplicationContext; @@ -27,8 +27,9 @@ import org.springframework.core.io.Resource; import org.springframework.core.io.support.ResourcePatternResolver; import org.springframework.integration.MessagingException; import org.springframework.integration.endpoint.AbstractMessageSource; -import org.springframework.integration.util.ElementFilter; +import org.springframework.integration.util.CollectionFilter; import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; /** @@ -36,28 +37,31 @@ import org.springframework.util.ObjectUtils; * attempt to resolve {@link Resource}s based on the pattern specified. * * @author Oleg Zhurakousky + * @author Mark Fisher * @since 2.1 */ public class ResourceMessageSource extends AbstractMessageSource implements ApplicationContextAware, InitializingBean { - private volatile String pattern; + private final String pattern; private volatile ApplicationContext applicationContext; private volatile ResourcePatternResolver patternResolver; - private volatile ElementFilter filter; + private volatile CollectionFilter filter; + + + public ResourceMessageSource(String pattern) { + Assert.hasText(pattern, "pattern must not be empty"); + this.pattern = pattern; + } public void setPatternResolver(ResourcePatternResolver patternResolver) { this.patternResolver = patternResolver; } - public void setPattern(String pattern) { - this.pattern = pattern; - } - - public void setFilter(ElementFilter filter) { + public void setFilter(CollectionFilter filter) { this.filter = filter; } @@ -71,8 +75,7 @@ public class ResourceMessageSource extends AbstractMessageSource imp this.patternResolver = this.applicationContext; } } - Assert.notNull(this.patternResolver, "no 'patternResolver' is specified"); - Assert.hasText(this.pattern, "'pattern' must be specified"); + Assert.notNull(this.patternResolver, "no 'patternResolver' available"); } @Override @@ -80,14 +83,8 @@ public class ResourceMessageSource extends AbstractMessageSource imp try { Resource[] resources = this.patternResolver.getResources(this.pattern); if (this.filter != null && !ObjectUtils.isEmpty(resources)) { - List filteredResources = new ArrayList(); - for (Resource resource : resources) { - Resource filteredResource = this.filter.filter(resource); - if (filteredResource != null) { - filteredResources.add(filteredResource); - } - } - if (filteredResources.size() == 0) { + Collection filteredResources = this.filter.filter(Arrays.asList(resources)); + if (CollectionUtils.isEmpty(filteredResources)) { resources = null; } else { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/AcceptOnceUntilPurgedElementFilter.java b/spring-integration-core/src/main/java/org/springframework/integration/util/AcceptOnceUntilPurgedElementFilter.java index e34ceeee97..2369f23c9d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/AcceptOnceUntilPurgedElementFilter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/AcceptOnceUntilPurgedElementFilter.java @@ -13,40 +13,59 @@ * 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 ElementFilter} which will queue all items that's been seen until + * 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 corresponds to the underlying items once processing * is done to eliminate duplicate processing. * * @author Oleg Zhurakousky + * @author Mark Fisher * @since 2.1 */ -public class AcceptOnceUntilPurgedElementFilter implements ElementFilter { - +public class AcceptOnceUntilPurgedElementFilter implements CollectionFilter { + private final Log logger = LogFactory.getLog(this.getClass()); private final Queue seenItems; private final Object seenQueueMonitor = new Object(); - - public AcceptOnceUntilPurgedElementFilter(){ + + + public AcceptOnceUntilPurgedElementFilter() { this(Integer.MAX_VALUE); } - - public AcceptOnceUntilPurgedElementFilter(int maxCapacity){ - seenItems = new LinkedBlockingQueue(maxCapacity); + + public AcceptOnceUntilPurgedElementFilter(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) { @@ -67,15 +86,4 @@ public class AcceptOnceUntilPurgedElementFilter implements ElementFilter { } } - public T filter(T unfilteredElement) { - Assert.notNull(unfilteredElement, "'unfilteredElement' must not be null"); - - if (this.accept(unfilteredElement)){ - return unfilteredElement; - } - else { - return null; - } - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/ElementFilter.java b/spring-integration-core/src/main/java/org/springframework/integration/util/CollectionFilter.java similarity index 76% rename from spring-integration-core/src/main/java/org/springframework/integration/util/ElementFilter.java rename to spring-integration-core/src/main/java/org/springframework/integration/util/CollectionFilter.java index d0aaac9277..b1adc56362 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/ElementFilter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/CollectionFilter.java @@ -13,16 +13,20 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.util; +import java.util.Collection; /** - * Base strategy for filtering out an element + * Base strategy for filtering out a subset of a Collection of elements. * * @author Oleg Zhurakousky + * @author Mark Fisher * @since 2.1 */ -public interface ElementFilter { +public interface CollectionFilter { + + Collection filter(Collection unfilteredElements); - T filter(T unfilteredElement); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/resource/ResourcePatternResolverParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/resource/ResourcePatternResolverParserTests.java index 68da765a4d..49acd12c9f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/resource/ResourcePatternResolverParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/resource/ResourcePatternResolverParserTests.java @@ -13,9 +13,18 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + 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.assertNull; +import static org.junit.Assert.assertTrue; + import java.io.File; +import java.util.Collection; +import java.util.Collections; import org.junit.Test; @@ -27,13 +36,8 @@ import org.springframework.integration.Message; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; import org.springframework.integration.test.util.TestUtils; -import org.springframework.integration.util.ElementFilter; - -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; +import org.springframework.integration.util.CollectionFilter; +import org.springframework.util.CollectionUtils; /** * @author Oleg Zhurakousky @@ -114,19 +118,19 @@ public class ResourcePatternResolverParserTests { assertNull(message); } - - public static class OneItemAndNeverAgainResourceListFilter implements ElementFilter { + + + public static class OneItemAndNeverAgainResourceListFilter implements CollectionFilter { private volatile boolean once = false; - public Resource filter(Resource unfilteredElement) { - - if (!once){ + public Collection filter(Collection unfilteredResources) { + if (!once && !CollectionUtils.isEmpty(unfilteredResources)) { once = true; - return unfilteredElement; + return Collections.singletonList(unfilteredResources.iterator().next()); } return null; } - } + }