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

INT-2235

Browse Source 
added support for resource-inbound-channel-adapter

  removed pre-fetching logic

  changed ResourcePatternResolvingMessageSource to return multiple Resources

  made filter optional
This commit is contained in:
Oleg Zhurakousky
2011-11-11 12:14:16 -05:00
committed by Mark Fisher
parent 5cac946cf3
commit 193f3d2439
13 changed files with 553 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java
View File

@@ -53,6 +53,7 @@ public class IntegrationNamespaceHandler extends AbstractIntegrationNamespaceHan
registerBeanDefinitionParser("claim-check-in", new ClaimCheckInParser());
registerBeanDefinitionParser("claim-check-out", new ClaimCheckOutParser());
registerBeanDefinitionParser("inbound-channel-adapter", new MethodInvokingInboundChannelAdapterParser());
registerBeanDefinitionParser("resource-inbound-channel-adapter", new ResourceInboundChannelAdapterParser());
registerBeanDefinitionParser("outbound-channel-adapter", new MethodInvokingOutboundChannelAdapterParser());
registerBeanDefinitionParser("logging-channel-adapter", new LoggingChannelAdapterParser());
registerBeanDefinitionParser("gateway", new GatewayParser());

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

@@ -0,0 +1,42 @@
/*
* 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.config.xml;
import org.springframework.beans.BeanMetadataElement;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.resource.ResourceMessageSource;
import org.w3c.dom.Element;
/**
* Parser for 'resource-inbound-channel-adapter'
*
* @author Oleg Zhurakousky
* @since 2.1
*/
public class ResourceInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
@Override
protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
BeanDefinitionBuilder sourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(ResourceMessageSource.class);
IntegrationNamespaceUtils.setValueIfAttributeDefined(sourceBuilder, element, "pattern");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(sourceBuilder, element, "pattern-resolver");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(sourceBuilder, element, "filter");
return sourceBuilder.getBeanDefinition();
}
}

5
spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java
View File

@@ -16,9 +16,6 @@
package org.springframework.integration.endpoint;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.context.NamedComponent;
@@ -36,8 +33,6 @@ import org.springframework.util.Assert;
* @author Oleg Zhurakousky
*/
public class SourcePollingChannelAdapter extends AbstractPollingEndpoint implements TrackableComponent {
private final Log logger = LogFactory.getLog(this.getClass());
private volatile MessageSource<?> source;

104
spring-integration-core/src/main/java/org/springframework/integration/resource/ResourceMessageSource.java Normal file
View File

@@ -0,0 +1,104 @@
/*
* 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.resource;
import java.util.ArrayList;
import java.util.List;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.MessageSource;
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.util.Assert;
import org.springframework.util.ObjectUtils;
/**
* Implementation of {@link MessageSource} based on {@link ResourcePatternResolver} which will
* attempt to resolve {@link Resource}s based on the pattern specified.
*
* @author Oleg Zhurakousky
* @since 2.1
*/
public class ResourceMessageSource extends AbstractMessageSource<Resource[]> implements ApplicationContextAware, InitializingBean {
private volatile String pattern;
private volatile ApplicationContext applicationContext;
private volatile ResourcePatternResolver patternResolver;
private volatile ElementFilter<Resource> filter;
public void setPatternResolver(ResourcePatternResolver patternResolver) {
this.patternResolver = patternResolver;
}
public void setPattern(String pattern) {
this.pattern = pattern;
}
public void setFilter(ElementFilter<Resource> filter) {
this.filter = filter;
}
public void setApplicationContext(ApplicationContext applicationContext) {
this.applicationContext = applicationContext;
}
public void afterPropertiesSet() {
if (this.patternResolver == null) {
if (this.applicationContext instanceof ResourcePatternResolver) {
this.patternResolver = this.applicationContext;
}
}
Assert.notNull(this.patternResolver, "no 'patternResolver' is specified");
Assert.hasText(this.pattern, "'pattern' must be specified");
}
@Override
protected Resource[] doReceive() {
try {
Resource[] resources = this.patternResolver.getResources(this.pattern);
if (this.filter != null && !ObjectUtils.isEmpty(resources)) {
List<Resource> filteredResources = new ArrayList<Resource>();
for (Resource resource : resources) {
Resource filteredResource = this.filter.filter(resource);
if (filteredResource != null) {
filteredResources.add(filteredResource);
}
}
if (filteredResources.size() == 0) {
resources = null;
}
else {
resources = filteredResources.toArray(new Resource[0]);
}
}
return resources;
}
catch (Exception e) {
throw new MessagingException("Attempt to retrieve Resources failed", e);
}
}
}

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

@@ -0,0 +1,81 @@
/*
* 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.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
* 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
* @since 2.1
*/
public class AcceptOnceUntilPurgedElementFilter<T> implements ElementFilter<T> {
private final Log logger = LogFactory.getLog(this.getClass());
private final Queue<T> seenItems;
private final Object seenQueueMonitor = new Object();
public AcceptOnceUntilPurgedElementFilter(){
this(Integer.MAX_VALUE);
}
public AcceptOnceUntilPurgedElementFilter(int maxCapacity){
seenItems = new LinkedBlockingQueue<T>(maxCapacity);
}
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;
}
}
public T filter(T unfilteredElement) {
Assert.notNull(unfilteredElement, "'unfilteredElement' must not be null");
if (this.accept(unfilteredElement)){
return unfilteredElement;
}
else {
return null;
}
}
}

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

@@ -0,0 +1,28 @@
/*
* 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;
/**
* Base strategy for filtering out an element
*
* @author Oleg Zhurakousky
* @since 2.1
*/
public interface ElementFilter<T> {
T filter(T unfilteredElement);
}

72
spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.1.xsd
View File

@@ -716,6 +716,78 @@
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:element name="resource-inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Defines a Channel Adapter that receives Resource(s) and sends them to a
MessageChannel identified via 'channel' attribute.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:sequence>
<xsd:element name="poller" type="basePollerType" />
</xsd:sequence>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Component identifier
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Channel where Message will be sent to
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="filter" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Reference to the implementation of org.springframework.integration.util.ElementFilter.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.util.ElementFilter" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="auto-startup" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Lifecycle attribute signaling if this component should be started during Application Context startup.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="pattern" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Location pattern expression (e.g., "/**/*.txt")
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="pattern-resolver" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Reference to a org.springframework.core.io.support.ResourcePatternResolver.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="send-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Maximum amount of time in milliseconds to wait when sending a message to the channel if such channel may block.
For example, a Queue Channel can block until space is available if its maximum capacity has been reached.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:element name="outbound-channel-adapter">
<xsd:annotation>

20
spring-integration-core/src/test/java/org/springframework/integration/resource/ResourcePatternResolver-config-custom.xml Normal file
View File

@@ -0,0 +1,20 @@
<?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="/**/*"
auto-startup="false" pattern-resolver="customResolver">
<int:poller fixed-rate="1000"/>
</int:resource-inbound-channel-adapter>
<bean id="customResolver" class="org.springframework.core.io.support.PathMatchingResourcePatternResolver"/>
<int:channel id="resultChannel">
<int:queue/>
</int:channel>
</beans>

17
spring-integration-core/src/test/java/org/springframework/integration/resource/ResourcePatternResolver-config-fail.xml Normal file
View File

@@ -0,0 +1,17 @@
<?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" auto-startup="false">
<int:poller fixed-rate="1000"/>
</int:resource-inbound-channel-adapter>
<int:channel id="resultChannel">
<int:queue/>
</int:channel>
</beans>

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

@@ -0,0 +1,18 @@
<?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') + 'testUsage*'}">
<int:poller fixed-rate="2000"/>
</int:resource-inbound-channel-adapter>
<int:channel id="resultChannel">
<int:queue/>
</int:channel>
</beans>

21
spring-integration-core/src/test/java/org/springframework/integration/resource/ResourcePatternResolver-config-usagerf.xml Normal file
View File

@@ -0,0 +1,21 @@
<?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') + 'testUsage*'}"
filter="rlFilter" auto-startup="true">
<int:poller fixed-rate="5000"/>
</int:resource-inbound-channel-adapter>
<bean id="rlFilter" class="org.springframework.integration.resource.ResourcePatternResolverParserTests.OneItemAndNeverAgainResourceListFilter"/>
<int:channel id="resultChannel">
<int:queue/>
</int:channel>
</beans>

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

@@ -0,0 +1,17 @@
<?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="/**/*" auto-startup="false">
<int:poller fixed-rate="1000"/>
</int:resource-inbound-channel-adapter>
<int:channel id="resultChannel">
<int:queue/>
</int:channel>
</beans>

132
spring-integration-core/src/test/java/org/springframework/integration/resource/ResourcePatternResolverParserTests.java Normal file
View File

@@ -0,0 +1,132 @@
/*
* 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.resource;
import java.io.File;
import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.core.io.Resource;
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;
/**
* @author Oleg Zhurakousky
*
*/
public class ResourcePatternResolverParserTests {
@Test
public void testDefaultConfig(){
ApplicationContext context = new ClassPathXmlApplicationContext("ResourcePatternResolver-config.xml", this.getClass());
SourcePollingChannelAdapter resourceAdapter = context.getBean("resourceAdapterDefault", SourcePollingChannelAdapter.class);
ResourceMessageSource source = TestUtils.getPropertyValue(resourceAdapter, "source", ResourceMessageSource.class);
assertNotNull(source);
boolean autoStartup = TestUtils.getPropertyValue(resourceAdapter, "autoStartup", Boolean.class);
assertFalse(autoStartup);
assertEquals("/**/*", TestUtils.getPropertyValue(source, "pattern"));
assertEquals(context, TestUtils.getPropertyValue(source, "patternResolver"));
}
@Test(expected=BeanCreationException.class)
public void testDefaultConfigNoLocationPattern(){
new ClassPathXmlApplicationContext("ResourcePatternResolver-config-fail.xml", this.getClass());
}
@Test
public void testCustomPatternResolver(){
ApplicationContext context = new ClassPathXmlApplicationContext("ResourcePatternResolver-config-custom.xml", this.getClass());
SourcePollingChannelAdapter resourceAdapter = context.getBean("resourceAdapterDefault", SourcePollingChannelAdapter.class);
ResourceMessageSource source = TestUtils.getPropertyValue(resourceAdapter, "source", ResourceMessageSource.class);
assertNotNull(source);
assertEquals(context.getBean("customResolver"), TestUtils.getPropertyValue(source, "patternResolver"));
}
@SuppressWarnings("unchecked")
@Test
public void testUsage() throws Exception{
File baseDir = new File(System.getProperty("java.io.tmpdir"));
for (int i = 0; i < 10; i++) {
File f = new File(baseDir, "testUsage"+i);
f.createNewFile();
}
ApplicationContext context = new ClassPathXmlApplicationContext("ResourcePatternResolver-config-usage.xml", this.getClass());
QueueChannel resultChannel = context.getBean("resultChannel", QueueChannel.class);
Message<Resource[]> message = (Message<Resource[]>) resultChannel.receive(3000);
assertNotNull(message);
Resource[] resources = message.getPayload();
for (Resource resource : resources) {
assertTrue(resource.getURI().toString().contains("testUsage"));
}
}
@SuppressWarnings("unchecked")
@Test
public void testUsageWithCustomResourceFilter() 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-usagerf.xml", this.getClass());
SourcePollingChannelAdapter resourceAdapter = context.getBean("resourceAdapterDefault", SourcePollingChannelAdapter.class);
ResourceMessageSource source = TestUtils.getPropertyValue(resourceAdapter, "source", ResourceMessageSource.class);
assertNotNull(source);
assertEquals(context.getBean("rlFilter"), TestUtils.getPropertyValue(source, "filter"));
QueueChannel resultChannel = context.getBean("resultChannel", QueueChannel.class);
Message<Resource[]> message = (Message<Resource[]>) resultChannel.receive(1000);
assertNotNull(message);
message = (Message<Resource[]>) resultChannel.receive(1000);
assertNull(message);
}
public static class OneItemAndNeverAgainResourceListFilter implements ElementFilter<Resource> {
private volatile boolean once = false;
public Resource filter(Resource unfilteredElement) {
if (!once){
once = true;
return unfilteredElement;
}
return null;
}
}
}