ElementFilter now CollectionFilter
Refactored filter to handle a Collection of Resources Made 'pattern' a constructor-arg since it's mandatory
This commit is contained in:
@@ -13,6 +13,7 @@
|
|||||||
* See the License for the specific language governing permissions and
|
* See the License for the specific language governing permissions and
|
||||||
* limitations under the License.
|
* limitations under the License.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
package org.springframework.integration.config.xml;
|
package org.springframework.integration.config.xml;
|
||||||
|
|
||||||
import org.springframework.beans.BeanMetadataElement;
|
import org.springframework.beans.BeanMetadataElement;
|
||||||
@@ -33,7 +34,7 @@ public class ResourceInboundChannelAdapterParser extends AbstractPollingInboundC
|
|||||||
@Override
|
@Override
|
||||||
protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
|
protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
|
||||||
BeanDefinitionBuilder sourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(ResourceMessageSource.class);
|
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, "pattern-resolver");
|
||||||
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(sourceBuilder, element, "filter");
|
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(sourceBuilder, element, "filter");
|
||||||
return sourceBuilder.getBeanDefinition();
|
return sourceBuilder.getBeanDefinition();
|
||||||
|
|||||||
@@ -16,8 +16,8 @@
|
|||||||
|
|
||||||
package org.springframework.integration.resource;
|
package org.springframework.integration.resource;
|
||||||
|
|
||||||
import java.util.ArrayList;
|
import java.util.Arrays;
|
||||||
import java.util.List;
|
import java.util.Collection;
|
||||||
|
|
||||||
import org.springframework.beans.factory.InitializingBean;
|
import org.springframework.beans.factory.InitializingBean;
|
||||||
import org.springframework.context.ApplicationContext;
|
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.core.io.support.ResourcePatternResolver;
|
||||||
import org.springframework.integration.MessagingException;
|
import org.springframework.integration.MessagingException;
|
||||||
import org.springframework.integration.endpoint.AbstractMessageSource;
|
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.Assert;
|
||||||
|
import org.springframework.util.CollectionUtils;
|
||||||
import org.springframework.util.ObjectUtils;
|
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.
|
* attempt to resolve {@link Resource}s based on the pattern specified.
|
||||||
*
|
*
|
||||||
* @author Oleg Zhurakousky
|
* @author Oleg Zhurakousky
|
||||||
|
* @author Mark Fisher
|
||||||
* @since 2.1
|
* @since 2.1
|
||||||
*/
|
*/
|
||||||
public class ResourceMessageSource extends AbstractMessageSource<Resource[]> implements ApplicationContextAware, InitializingBean {
|
public class ResourceMessageSource extends AbstractMessageSource<Resource[]> implements ApplicationContextAware, InitializingBean {
|
||||||
|
|
||||||
private volatile String pattern;
|
private final String pattern;
|
||||||
|
|
||||||
private volatile ApplicationContext applicationContext;
|
private volatile ApplicationContext applicationContext;
|
||||||
|
|
||||||
private volatile ResourcePatternResolver patternResolver;
|
private volatile ResourcePatternResolver patternResolver;
|
||||||
|
|
||||||
private volatile ElementFilter<Resource> filter;
|
private volatile CollectionFilter<Resource> filter;
|
||||||
|
|
||||||
|
|
||||||
|
public ResourceMessageSource(String pattern) {
|
||||||
|
Assert.hasText(pattern, "pattern must not be empty");
|
||||||
|
this.pattern = pattern;
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
public void setPatternResolver(ResourcePatternResolver patternResolver) {
|
public void setPatternResolver(ResourcePatternResolver patternResolver) {
|
||||||
this.patternResolver = patternResolver;
|
this.patternResolver = patternResolver;
|
||||||
}
|
}
|
||||||
|
|
||||||
public void setPattern(String pattern) {
|
public void setFilter(CollectionFilter<Resource> filter) {
|
||||||
this.pattern = pattern;
|
|
||||||
}
|
|
||||||
|
|
||||||
public void setFilter(ElementFilter<Resource> filter) {
|
|
||||||
this.filter = filter;
|
this.filter = filter;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -71,8 +75,7 @@ public class ResourceMessageSource extends AbstractMessageSource<Resource[]> imp
|
|||||||
this.patternResolver = this.applicationContext;
|
this.patternResolver = this.applicationContext;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
Assert.notNull(this.patternResolver, "no 'patternResolver' is specified");
|
Assert.notNull(this.patternResolver, "no 'patternResolver' available");
|
||||||
Assert.hasText(this.pattern, "'pattern' must be specified");
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
@@ -80,14 +83,8 @@ public class ResourceMessageSource extends AbstractMessageSource<Resource[]> imp
|
|||||||
try {
|
try {
|
||||||
Resource[] resources = this.patternResolver.getResources(this.pattern);
|
Resource[] resources = this.patternResolver.getResources(this.pattern);
|
||||||
if (this.filter != null && !ObjectUtils.isEmpty(resources)) {
|
if (this.filter != null && !ObjectUtils.isEmpty(resources)) {
|
||||||
List<Resource> filteredResources = new ArrayList<Resource>();
|
Collection<Resource> filteredResources = this.filter.filter(Arrays.asList(resources));
|
||||||
for (Resource resource : resources) {
|
if (CollectionUtils.isEmpty(filteredResources)) {
|
||||||
Resource filteredResource = this.filter.filter(resource);
|
|
||||||
if (filteredResource != null) {
|
|
||||||
filteredResources.add(filteredResource);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if (filteredResources.size() == 0) {
|
|
||||||
resources = null;
|
resources = null;
|
||||||
}
|
}
|
||||||
else {
|
else {
|
||||||
|
|||||||
@@ -13,40 +13,59 @@
|
|||||||
* See the License for the specific language governing permissions and
|
* See the License for the specific language governing permissions and
|
||||||
* limitations under the License.
|
* limitations under the License.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
package org.springframework.integration.util;
|
package org.springframework.integration.util;
|
||||||
|
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.Collection;
|
||||||
|
import java.util.List;
|
||||||
import java.util.Queue;
|
import java.util.Queue;
|
||||||
import java.util.concurrent.LinkedBlockingQueue;
|
import java.util.concurrent.LinkedBlockingQueue;
|
||||||
|
|
||||||
import org.apache.commons.logging.Log;
|
import org.apache.commons.logging.Log;
|
||||||
import org.apache.commons.logging.LogFactory;
|
import org.apache.commons.logging.LogFactory;
|
||||||
|
|
||||||
import org.springframework.util.Assert;
|
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
|
* 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
|
* 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
|
* so it is highly recommended to move/delete resources which corresponds to the underlying items once processing
|
||||||
* is done to eliminate duplicate processing.
|
* is done to eliminate duplicate processing.
|
||||||
*
|
*
|
||||||
* @author Oleg Zhurakousky
|
* @author Oleg Zhurakousky
|
||||||
|
* @author Mark Fisher
|
||||||
* @since 2.1
|
* @since 2.1
|
||||||
*/
|
*/
|
||||||
public class AcceptOnceUntilPurgedElementFilter<T> implements ElementFilter<T> {
|
public class AcceptOnceUntilPurgedElementFilter<T> implements CollectionFilter<T> {
|
||||||
|
|
||||||
private final Log logger = LogFactory.getLog(this.getClass());
|
private final Log logger = LogFactory.getLog(this.getClass());
|
||||||
|
|
||||||
private final Queue<T> seenItems;
|
private final Queue<T> seenItems;
|
||||||
|
|
||||||
private final Object seenQueueMonitor = new Object();
|
private final Object seenQueueMonitor = new Object();
|
||||||
|
|
||||||
public AcceptOnceUntilPurgedElementFilter(){
|
|
||||||
|
public AcceptOnceUntilPurgedElementFilter() {
|
||||||
this(Integer.MAX_VALUE);
|
this(Integer.MAX_VALUE);
|
||||||
}
|
}
|
||||||
|
|
||||||
public AcceptOnceUntilPurgedElementFilter(int maxCapacity){
|
public AcceptOnceUntilPurgedElementFilter(int maxCapacity) {
|
||||||
seenItems = new LinkedBlockingQueue<T>(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) {
|
private boolean accept(T item) {
|
||||||
@@ -67,15 +86,4 @@ public class AcceptOnceUntilPurgedElementFilter<T> implements ElementFilter<T> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
public T filter(T unfilteredElement) {
|
|
||||||
Assert.notNull(unfilteredElement, "'unfilteredElement' must not be null");
|
|
||||||
|
|
||||||
if (this.accept(unfilteredElement)){
|
|
||||||
return unfilteredElement;
|
|
||||||
}
|
|
||||||
else {
|
|
||||||
return null;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -13,16 +13,20 @@
|
|||||||
* See the License for the specific language governing permissions and
|
* See the License for the specific language governing permissions and
|
||||||
* limitations under the License.
|
* limitations under the License.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
package org.springframework.integration.util;
|
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 Oleg Zhurakousky
|
||||||
|
* @author Mark Fisher
|
||||||
* @since 2.1
|
* @since 2.1
|
||||||
*/
|
*/
|
||||||
public interface ElementFilter<T> {
|
public interface CollectionFilter<T> {
|
||||||
|
|
||||||
|
Collection<T> filter(Collection<T> unfilteredElements);
|
||||||
|
|
||||||
T filter(T unfilteredElement);
|
|
||||||
}
|
}
|
||||||
@@ -13,9 +13,18 @@
|
|||||||
* See the License for the specific language governing permissions and
|
* See the License for the specific language governing permissions and
|
||||||
* limitations under the License.
|
* limitations under the License.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
package org.springframework.integration.resource;
|
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.io.File;
|
||||||
|
import java.util.Collection;
|
||||||
|
import java.util.Collections;
|
||||||
|
|
||||||
import org.junit.Test;
|
import org.junit.Test;
|
||||||
|
|
||||||
@@ -27,13 +36,8 @@ import org.springframework.integration.Message;
|
|||||||
import org.springframework.integration.channel.QueueChannel;
|
import org.springframework.integration.channel.QueueChannel;
|
||||||
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
|
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
|
||||||
import org.springframework.integration.test.util.TestUtils;
|
import org.springframework.integration.test.util.TestUtils;
|
||||||
import org.springframework.integration.util.ElementFilter;
|
import org.springframework.integration.util.CollectionFilter;
|
||||||
|
import org.springframework.util.CollectionUtils;
|
||||||
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
|
* @author Oleg Zhurakousky
|
||||||
@@ -114,19 +118,19 @@ public class ResourcePatternResolverParserTests {
|
|||||||
assertNull(message);
|
assertNull(message);
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public static class OneItemAndNeverAgainResourceListFilter implements ElementFilter<Resource> {
|
|
||||||
|
public static class OneItemAndNeverAgainResourceListFilter implements CollectionFilter<Resource> {
|
||||||
|
|
||||||
private volatile boolean once = false;
|
private volatile boolean once = false;
|
||||||
|
|
||||||
public Resource filter(Resource unfilteredElement) {
|
public Collection<Resource> filter(Collection<Resource> unfilteredResources) {
|
||||||
|
if (!once && !CollectionUtils.isEmpty(unfilteredResources)) {
|
||||||
if (!once){
|
|
||||||
once = true;
|
once = true;
|
||||||
return unfilteredElement;
|
return Collections.singletonList(unfilteredResources.iterator().next());
|
||||||
}
|
}
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user