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

INT-4153: Feed Java DSL and other improvements

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4153

* Remove deprecated `FeedFetcher` usage
* Introduce `Resource` based ctor for the `FeedEntryMessageSource`
* Add `SyndFeedInput` option and short-hand `preserveWireFeed` for internal `SyndFeedInput` instance
* Reflect the changes in the XSD for Feed
* Change ROME dependency from deprecated `rome-fetcher` to just `rome` as it is recommended by ROME team
* Port Java DSL for Feed module and reflect aforementioned changes in the `Feed` factory and `FeedEntryMessageSourceSpec` as well
* Document changes and mention Feed Java DSL, too
This commit is contained in:
Artem Bilan
2016-11-08 15:39:25 -05:00
committed by Gary Russell
parent 1871b11e85
commit 507764a3d6
18 changed files with 470 additions and 268 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -317,7 +317,7 @@ project('spring-integration-feed') {
description = 'Spring Integration RSS Feed Support'
dependencies {
compile project(":spring-integration-core")
compile "com.rometools:rome-fetcher:$romeToolsVersion"
compile "com.rometools:rome:$romeToolsVersion"
}
}

2
spring-integration-core/src/main/java/org/springframework/integration/config/xml/HeaderEnricherParserSupport.java
View File

@@ -222,7 +222,7 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar
.error("The 'method' attribute cannot be used when a 'script' sub-element is defined", element);
}
if (!(isValue ^ (isRef ^ (isExpression ^ isCustomBean)))) {
if (isValue == (isRef ^ (isExpression ^ isCustomBean))) {
parserContext.getReaderContext().error(
"Exactly one of the 'ref', 'value', 'expression' or inner bean is required.", element);
}

31
spring-integration-feed/src/main/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2016 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.
@@ -34,6 +34,7 @@ import org.springframework.util.StringUtils;
* @author Mark Fisher
* @author Gunnar Hillert
* @author Artem Bilan
*
* @since 2.0
*/
public class FeedInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
@@ -41,13 +42,31 @@ public class FeedInboundChannelAdapterParser extends AbstractPollingInboundChann
@Override
protected BeanMetadataElement parseSource(final Element element, final ParserContext parserContext) {
BeanDefinitionBuilder sourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(FeedEntryMessageSource.class);
sourceBuilder.addConstructorArgValue(element.getAttribute("url"));
sourceBuilder.addConstructorArgValue(element.getAttribute(ID_ATTRIBUTE));
String feedFetcherRef = element.getAttribute("feed-fetcher");
if (StringUtils.hasText(feedFetcherRef)) {
sourceBuilder.addConstructorArgReference(feedFetcherRef);
String url = element.getAttribute("url");
boolean hasUrl = StringUtils.hasText(url);
String resource = element.getAttribute("resource");
boolean hasResource = StringUtils.hasText(resource);
if (hasUrl == hasResource) {
parserContext.getReaderContext().error(
"Exactly one of the 'url', 'reader' or 'resource' is required.", element);
}
if (hasUrl) {
sourceBuilder.addConstructorArgValue(url);
}
else if (hasResource) {
sourceBuilder.addConstructorArgValue(resource);
}
sourceBuilder.addConstructorArgValue(element.getAttribute(ID_ATTRIBUTE));
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(sourceBuilder, element, "metadata-store");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(sourceBuilder, element, "feed-input", "syndFeedInput");
IntegrationNamespaceUtils.setValueIfAttributeDefined(sourceBuilder, element, "preserve-wire-feed");
return sourceBuilder.getBeanDefinition();
}

43
spring-integration-feed/src/main/java/org/springframework/integration/feed/dsl/Feed.java Normal file
View File

@@ -0,0 +1,43 @@
/*
* Copyright 2016 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.feed.dsl;
import java.net.URL;
import org.springframework.core.io.Resource;
/**
* The Spring Integration Feed components Factory.
*
* @author Artem Bilan
* @since 5.0
*/
public final class Feed {
public static FeedEntryMessageSourceSpec inboundAdapter(URL feedUrl, String metadataKey) {
return new FeedEntryMessageSourceSpec(feedUrl, metadataKey);
}
public static FeedEntryMessageSourceSpec inboundAdapter(Resource feedResource, String metadataKey) {
return new FeedEntryMessageSourceSpec(feedResource, metadataKey);
}
private Feed() {
super();
}
}

59
spring-integration-feed/src/main/java/org/springframework/integration/feed/dsl/FeedEntryMessageSourceSpec.java Normal file
View File

@@ -0,0 +1,59 @@
/*
* Copyright 2016 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.feed.dsl;
import java.net.URL;
import org.springframework.core.io.Resource;
import org.springframework.integration.dsl.MessageSourceSpec;
import org.springframework.integration.feed.inbound.FeedEntryMessageSource;
import org.springframework.integration.metadata.MetadataStore;
import com.rometools.rome.io.SyndFeedInput;
/**
* A {@link MessageSourceSpec} for a {@link FeedEntryMessageSource}.
*
* @author Artem Bilan
*
* @since 5.0
*/
public class FeedEntryMessageSourceSpec extends MessageSourceSpec<FeedEntryMessageSourceSpec, FeedEntryMessageSource> {
FeedEntryMessageSourceSpec(URL feedUrl, String metadataKey) {
this.target = new FeedEntryMessageSource(feedUrl, metadataKey);
}
FeedEntryMessageSourceSpec(Resource feedResource, String metadataKey) {
this.target = new FeedEntryMessageSource(feedResource, metadataKey);
}
public FeedEntryMessageSourceSpec metadataStore(MetadataStore metadataStore) {
this.target.setMetadataStore(metadataStore);
return this;
}
public FeedEntryMessageSourceSpec syndFeedInput(SyndFeedInput syndFeedInput) {
this.target.setSyndFeedInput(syndFeedInput);
return this;
}
public FeedEntryMessageSourceSpec preserveWireFeed(boolean preserveWireFeed) {
this.target.setPreserveWireFeed(preserveWireFeed);
return this;
}
}

4
spring-integration-feed/src/main/java/org/springframework/integration/feed/dsl/package-info.java Normal file
View File

@@ -0,0 +1,4 @@
/**
* Provides Feed Components support for Spring Integration Java DSL.
*/
package org.springframework.integration.feed.dsl;

137
spring-integration-feed/src/main/java/org/springframework/integration/feed/inbound/FeedEntryMessageSource.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.feed.inbound;
import java.io.Reader;
import java.net.URL;
import java.util.Collections;
import java.util.Comparator;
@@ -25,6 +26,7 @@ import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.io.Resource;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
@@ -38,6 +40,8 @@ import org.springframework.util.StringUtils;
import com.rometools.rome.feed.synd.SyndEntry;
import com.rometools.rome.feed.synd.SyndFeed;
import com.rometools.rome.io.SyndFeedInput;
import com.rometools.rome.io.XmlReader;
/**
* This implementation of {@link MessageSource} will produce individual
@@ -48,24 +52,18 @@ import com.rometools.rome.feed.synd.SyndFeed;
* @author Oleg Zhurakousky
* @author Artem Bilan
* @author Aaron Loes
*
* @since 2.0
*/
@SuppressWarnings("deprecation")
public class FeedEntryMessageSource extends IntegrationObjectSupport implements MessageSource<SyndEntry> {
private final URL feedUrl;
private final com.rometools.fetcher.FeedFetcher feedFetcher;
private final Queue<SyndEntry> entries = new ConcurrentLinkedQueue<SyndEntry>();
private final Resource feedResource;
private final String metadataKey;
private volatile MetadataStore metadataStore;
private volatile long lastTime = -1;
private volatile boolean initialized;
private final Queue<SyndEntry> entries = new ConcurrentLinkedQueue<>();
private final Object monitor = new Object();
@@ -73,41 +71,74 @@ public class FeedEntryMessageSource extends IntegrationObjectSupport implements
private final Object feedMonitor = new Object();
private volatile SyndFeedInput syndFeedInput = new SyndFeedInput();
private boolean syndFeedInputSet;
private volatile MetadataStore metadataStore;
private volatile long lastTime = -1;
private volatile boolean initialized;
/**
* Creates a FeedEntryMessageSource that will use a HttpURLFeedFetcher to read feeds from the given URL.
* If the feed URL has a protocol other than http*, consider providing a custom implementation of the
* {@link com.rometools.fetcher.FeedFetcher} via the alternate constructor.
* {@link Resource} via the alternate constructor.
* @param feedUrl The URL.
* @param metadataKey The metadata key.
*/
public FeedEntryMessageSource(URL feedUrl, String metadataKey) {
this(feedUrl, metadataKey,
new com.rometools.fetcher.impl.HttpURLFeedFetcher(
com.rometools.fetcher.impl.HashMapFeedInfoCache.getInstance()));
Assert.notNull(feedUrl, "'feedUrl' must not be null");
Assert.notNull(metadataKey, "'metadataKey' must not be null");
this.feedUrl = feedUrl;
this.metadataKey = metadataKey + "." + feedUrl;
this.feedResource = null;
}
/**
* Creates a FeedEntryMessageSource that will use the provided FeedFetcher to read from the given feed URL.
* @param feedUrl The URL.
* @param metadataKey The metadata key.
* @param feedFetcher The feed fetcher.
* Creates a FeedEntryMessageSource that will read feeds from the given {@link Resource}.
* @param feedResource the {@link Resource} to use.
* @param metadataKey the metadata key.
* @since 5.0
*/
public FeedEntryMessageSource(URL feedUrl, String metadataKey, com.rometools.fetcher.FeedFetcher feedFetcher) {
Assert.notNull(feedUrl, "feedUrl must not be null");
Assert.notNull(metadataKey, "metadataKey must not be null");
Assert.notNull(feedFetcher, "feedFetcher must not be null");
this.feedUrl = feedUrl;
this.metadataKey = metadataKey + "." + this.feedUrl;
this.feedFetcher = feedFetcher;
public FeedEntryMessageSource(Resource feedResource, String metadataKey) {
Assert.notNull(feedResource, "'feedResource' must not be null");
Assert.notNull(metadataKey, "'metadataKey' must not be null");
this.feedResource = feedResource;
this.metadataKey = metadataKey;
this.feedUrl = null;
}
public void setMetadataStore(MetadataStore metadataStore) {
Assert.notNull(metadataStore, "metadataStore must not be null");
Assert.notNull(metadataStore, "'metadataStore' must not be null");
this.metadataStore = metadataStore;
}
/**
* Specify a parser for Feed XML documents.
* @param syndFeedInput the {@link SyndFeedInput} to use.
* @since 5.0
*/
public void setSyndFeedInput(SyndFeedInput syndFeedInput) {
Assert.notNull(syndFeedInput, "'syndFeedInput' must not be null");
this.syndFeedInput = syndFeedInput;
this.syndFeedInputSet = true;
}
/**
* Specify a flag to indication if {@code WireFeed} should be preserved in the target {@link SyndFeed}.
* @param preserveWireFeed the {@code boolean} flag.
* @since 5.0
* @see SyndFeedInput#setPreserveWireFeed(boolean)
*/
public void setPreserveWireFeed(boolean preserveWireFeed) {
Assert.isTrue(!this.syndFeedInputSet,
"'preserveWireFeed' must be configured on the provided [" + this.syndFeedInput + "]");
this.syndFeedInput.setPreserveWireFeed(preserveWireFeed);
}
@Override
public String getComponentType() {
return "feed:inbound-channel-adapter";
@@ -126,7 +157,6 @@ public class FeedEntryMessageSource extends IntegrationObjectSupport implements
@Override
protected void onInit() throws Exception {
this.feedFetcher.addFetcherEventListener(new FeedQueueUpdatingFetcherListener());
if (this.metadataStore == null) {
// first try to look for a 'messageStore' in the context
BeanFactory beanFactory = this.getBeanFactory();
@@ -186,7 +216,7 @@ public class FeedEntryMessageSource extends IntegrationObjectSupport implements
for (SyndEntry entry : retrievedEntries) {
Date entryDate = getLastModifiedDate(entry);
if ((entryDate != null && entryDate.getTime() > this.lastTime)
|| (entryDate == null && withinNewEntries)) {
|| (entryDate == null && withinNewEntries)) {
this.entries.add(entry);
withinNewEntries = true;
}
@@ -196,25 +226,36 @@ public class FeedEntryMessageSource extends IntegrationObjectSupport implements
}
private SyndFeed getFeed() {
SyndFeed feed = null;
try {
synchronized (this.feedMonitor) {
feed = this.feedFetcher.retrieveFeed(this.feedUrl);
Reader reader = this.feedUrl != null
? new XmlReader(this.feedUrl)
: new XmlReader(this.feedResource.getInputStream());
SyndFeed feed = this.syndFeedInput.build(reader);
if (logger.isDebugEnabled()) {
logger.debug("retrieved feed at url '" + this.feedUrl + "'");
logger.debug("Retrieved feed for [" + this + "]");
}
if (feed == null) {
if (logger.isDebugEnabled()) {
logger.debug("no feeds updated, returning null");
logger.debug("No feeds updated for [" + this + "], returning null");
}
}
return feed;
}
}
catch (Exception e) {
throw new MessagingException(
"Failed to retrieve feed at url '" + this.feedUrl + "'", e);
throw new MessagingException("Failed to retrieve feed for '" + this + "'", e);
}
return feed;
}
@Override
public String toString() {
return "FeedEntryMessageSource{" +
"feedUrl=" + this.feedUrl +
", feedResource=" + this.feedResource +
", metadataKey='" + this.metadataKey + '\'' +
", lastTime=" + this.lastTime +
'}';
}
private static Date getLastModifiedDate(SyndEntry entry) {
@@ -222,7 +263,11 @@ public class FeedEntryMessageSource extends IntegrationObjectSupport implements
}
private static class SyndEntryPublishedDateComparator implements Comparator<SyndEntry> {
private static final class SyndEntryPublishedDateComparator implements Comparator<SyndEntry> {
SyndEntryPublishedDateComparator() {
super();
}
@Override
public int compare(SyndEntry entry1, SyndEntry entry2) {
@@ -239,24 +284,4 @@ public class FeedEntryMessageSource extends IntegrationObjectSupport implements
}
private class FeedQueueUpdatingFetcherListener implements com.rometools.fetcher.FetcherListener {
@Override
public void fetcherEvent(final com.rometools.fetcher.FetcherEvent event) {
String eventType = event.getEventType();
if (com.rometools.fetcher.FetcherEvent.EVENT_TYPE_FEED_POLLED.equals(eventType)) {
if (logger.isDebugEnabled()) {
logger.debug("\tEVENT: Feed Polled. URL = " + event.getUrlString());
}
}
else if (com.rometools.fetcher.FetcherEvent.EVENT_TYPE_FEED_UNCHANGED.equals(eventType)) {
if (logger.isDebugEnabled()) {
logger.debug("\tEVENT: Feed Unchanged. URL = " + event.getUrlString());
}
}
}
}
}

35
spring-integration-feed/src/main/resources/org/springframework/integration/feed/config/spring-integration-feed-5.0.xsd
View File

@@ -46,23 +46,35 @@
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
<xsd:attribute name="url" type="xsd:string" use="required">
<xsd:attribute name="url" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The URL for an RSS or ATOM feed.
Mutually exclusive with `resource`.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="feed-fetcher" use="optional" type="xsd:string">
<xsd:attribute name="resource">
<xsd:annotation>
<xsd:documentation>
Reference to a FeedFetcher instance for retrieving Feeds from the provided URL.
By default, the HTTP protocol is supported. For any other protocols or general
customizations, provide a reference to a different implementation.
Reference to a org.springframework.core.io.Resource instance - a Feed XML source.
Mutually exclusive with 'url'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="com.rometools.fetcher.FeedFetcher" />
<tool:expected-type type="org.springframework.core.io.Resource" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="feed-input" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Reference to a SyndFeedInput instance - a Feed XML parser.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="com.rometools.rome.io.SyndFeedInput" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
@@ -82,6 +94,17 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="preserve-wire-feed" use="optional">
<xsd:annotation>
<xsd:documentation>
A flag to indication if 'WireFeed' should be preserved in the target 'SyndFeed'.
Defaults 'false'.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string" />
</xsd:simpleType>
</xsd:attribute>
</xsd:complexType>
</xsd:element>

2
spring-integration-feed/src/test/java/log4j.properties
View File

@@ -5,4 +5,4 @@ log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%c{1}: %m%n
log4j.category.org.springframework.integration=WARN
log4j.category.org.springframework.integration.feed=DEBUG
log4j.category.org.springframework.integration.feed=WARN

14
spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests-file-context.xml
View File

@@ -7,11 +7,11 @@
http://www.springframework.org/schema/integration/feed http://www.springframework.org/schema/integration/feed/spring-integration-feed.xsd">
<feed:inbound-channel-adapter id="feedAdapter"
channel="feedChannel"
auto-startup="false"
feed-fetcher="fileUrlFeedFetcher"
metadata-store="customMetadataStore"
url="classpath:org/springframework/integration/feed/sample.rss">
channel="feedChannel"
auto-startup="false"
feed-input="syndFeedInput"
metadata-store="metadataStore"
resource="classpath:org/springframework/integration/feed/sample.rss">
<int:poller fixed-rate="10000" max-messages-per-poll="100" />
</feed:inbound-channel-adapter>
@@ -19,8 +19,8 @@
<int:queue/>
</int:channel>
<bean id="fileUrlFeedFetcher" class="org.springframework.integration.feed.inbound.FileUrlFeedFetcher"/>
<bean id="syndFeedInput" class="com.rometools.rome.io.SyndFeedInput"/>
<bean id="customMetadataStore" class="org.springframework.integration.feed.config.FeedInboundChannelAdapterParserTests.SampleMetadataStore"/>
<bean id="metadataStore" class="org.springframework.integration.metadata.SimpleMetadataStore"/>
</beans>

13
spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests-file-usage-context.xml
View File

@@ -9,9 +9,9 @@
<int:message-history />
<int-feed:inbound-channel-adapter id="feedAdapterUsage"
channel="feedChannelUsage"
url="classpath:org/springframework/integration/feed/sample.rss"
feed-fetcher="fileUrlFeedFetcher">
channel="feedChannelUsage"
resource="classpath:org/springframework/integration/feed/sample.rss"
preserve-wire-feed="true">
<int:poller fixed-rate="10000" max-messages-per-poll="100"/>
</int-feed:inbound-channel-adapter>
@@ -19,8 +19,9 @@
<bean class="org.springframework.integration.feed.config.FeedInboundChannelAdapterParserTests$SampleService" />
</int:service-activator>
<bean id="fileUrlFeedFetcher" class="org.springframework.integration.feed.inbound.FileUrlFeedFetcher"/>
<bean id="metadataStore" class="org.springframework.integration.metadata.PropertiesPersistingMetadataStore"/>
<bean id="metadataStore" class="org.springframework.integration.metadata.PropertiesPersistingMetadataStore">
<property name="baseDirectory"
value="#{T (org.springframework.integration.feed.config.FeedInboundChannelAdapterParserTests).tempFolder.root.absolutePath}"/>
</bean>
</beans>

62
spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.integration.feed.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
@@ -25,14 +26,14 @@ import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.io.File;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.mockito.Mockito;
import org.springframework.context.ConfigurableApplicationContext;
@@ -48,6 +49,7 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import com.rometools.rome.feed.synd.SyndEntry;
import com.rometools.rome.io.SyndFeedInput;
/**
* @author Oleg Zhurakousky
@@ -55,20 +57,16 @@ import com.rometools.rome.feed.synd.SyndEntry;
* @author Gary Russell
* @author Gunnar Hillert
* @author Artem Bilan
*
* @since 2.0
*/
public class FeedInboundChannelAdapterParserTests {
private static CountDownLatch latch;
@ClassRule
public final static TemporaryFolder tempFolder = new TemporaryFolder();
@Before
public void prepare() {
File persisterFile = new File(System.getProperty("java.io.tmpdir") + "/spring-integration/",
"feedAdapter.last.entry");
if (persisterFile.exists()) {
persisterFile.delete();
}
}
private static CountDownLatch latch;
@Test
public void validateSuccessfulFileConfigurationWithCustomMetadataStore() {
@@ -76,16 +74,14 @@ public class FeedInboundChannelAdapterParserTests {
"FeedInboundChannelAdapterParserTests-file-context.xml", this.getClass());
SourcePollingChannelAdapter adapter = context.getBean("feedAdapter", SourcePollingChannelAdapter.class);
FeedEntryMessageSource source = (FeedEntryMessageSource) TestUtils.getPropertyValue(adapter, "source");
MetadataStore metadataStore = (MetadataStore) TestUtils.getPropertyValue(source, "metadataStore");
assertTrue(metadataStore instanceof SampleMetadataStore);
assertEquals(metadataStore, context.getBean("customMetadataStore"));
Object fetcher = TestUtils.getPropertyValue(source, "feedFetcher");
assertEquals("FileUrlFeedFetcher", fetcher.getClass().getSimpleName());
assertSame(context.getBean(MetadataStore.class), TestUtils.getPropertyValue(source, "metadataStore"));
SyndFeedInput syndFeedInput = TestUtils.getPropertyValue(source, "syndFeedInput", SyndFeedInput.class);
assertSame(context.getBean(SyndFeedInput.class), syndFeedInput);
assertFalse(syndFeedInput.isPreserveWireFeed());
context.close();
}
@SuppressWarnings("deprecation")
@Test
public void validateSuccessfulHttpConfigurationWithCustomMetadataStore() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
@@ -93,18 +89,11 @@ public class FeedInboundChannelAdapterParserTests {
SourcePollingChannelAdapter adapter = context.getBean("feedAdapter", SourcePollingChannelAdapter.class);
FeedEntryMessageSource source = (FeedEntryMessageSource) TestUtils.getPropertyValue(adapter, "source");
assertNotNull(TestUtils.getPropertyValue(source, "metadataStore"));
Object fetcher = TestUtils.getPropertyValue(source, "feedFetcher");
assertTrue(fetcher instanceof com.rometools.fetcher.impl.HttpURLFeedFetcher);
context.close();
}
@Test
public void validateSuccessfulNewsRetrievalWithFileUrlAndMessageHistory() throws Exception {
File persisterFile = new File(System.getProperty("java.io.tmpdir") + "/spring-integration/",
"metadata-store.properties");
if (persisterFile.exists()) {
persisterFile.delete();
}
//Test file samples.rss has 3 news items
latch = spy(new CountDownLatch(3));
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
@@ -120,6 +109,10 @@ public class FeedInboundChannelAdapterParserTests {
"FeedInboundChannelAdapterParserTests-file-usage-context.xml", this.getClass());
latch.await(500, TimeUnit.MILLISECONDS);
verify(latch, times(0)).countDown();
SourcePollingChannelAdapter adapter = context.getBean("feedAdapterUsage", SourcePollingChannelAdapter.class);
assertTrue(TestUtils.getPropertyValue(adapter, "source.syndFeedInput.preserveWireFeed", Boolean.class));
context.close();
}
@@ -127,7 +120,7 @@ public class FeedInboundChannelAdapterParserTests {
@Ignore // goes against the real feed
public void validateSuccessfulNewsRetrievalWithHttpUrl() throws Exception {
final CountDownLatch latch = new CountDownLatch(3);
MessageHandler handler = spy((MessageHandler) message -> latch.countDown());
MessageHandler handler = spy(message -> latch.countDown());
ConfigurableApplicationContext context = new ClassPathXmlApplicationContext(
"FeedInboundChannelAdapterParserTests-http-context.xml", this.getClass());
DirectChannel feedChannel = context.getBean("feedChannel", DirectChannel.class);
@@ -176,23 +169,4 @@ public class FeedInboundChannelAdapterParserTests {
}
public static class SampleMetadataStore implements MetadataStore {
@Override
public void put(String key, String value) {
}
@Override
public String get(String key) {
return null;
}
@Override
public String remove(String key) {
return null;
}
}
}

118
spring-integration-feed/src/test/java/org/springframework/integration/feed/dsl/FeedDslTests.java Normal file
View File

@@ -0,0 +1,118 @@
/*
* Copyright 2016 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.feed.dsl;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.io.FileReader;
import java.util.Properties;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.Resource;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.metadata.MetadataStore;
import org.springframework.integration.metadata.PropertiesPersistingMetadataStore;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringRunner;
import com.rometools.rome.feed.synd.SyndEntry;
/**
* @author Artem Bilan
*
* @since 5.0
*/
@RunWith(SpringRunner.class)
@DirtiesContext
public class FeedDslTests {
@ClassRule
public final static TemporaryFolder tempFolder = new TemporaryFolder();
@Autowired
private PollableChannel entries;
@Autowired
private PropertiesPersistingMetadataStore metadataStore;
@Test
@SuppressWarnings("unchecked")
public void testFeedEntryMessageSourceFlow() throws Exception {
Message<SyndEntry> message1 = (Message<SyndEntry>) this.entries.receive(10000);
Message<SyndEntry> message2 = (Message<SyndEntry>) this.entries.receive(10000);
Message<SyndEntry> message3 = (Message<SyndEntry>) this.entries.receive(10000);
long time1 = message1.getPayload().getPublishedDate().getTime();
long time2 = message2.getPayload().getPublishedDate().getTime();
long time3 = message3.getPayload().getPublishedDate().getTime();
assertTrue(time1 < time2);
assertTrue(time2 < time3);
assertNull(this.entries.receive(10));
this.metadataStore.flush();
FileReader metadataStoreFile =
new FileReader(tempFolder.getRoot().getAbsolutePath() + "/metadata-store.properties");
Properties metadataStoreProperties = new Properties();
metadataStoreProperties.load(metadataStoreFile);
assertFalse(metadataStoreProperties.isEmpty());
assertEquals(1, metadataStoreProperties.size());
assertTrue(metadataStoreProperties.containsKey("feedTest"));
}
@Configuration
@EnableIntegration
public static class ContextConfiguration {
@Value("org/springframework/integration/feed/sample.rss")
private Resource feedResource;
@Bean
public MetadataStore metadataStore() {
PropertiesPersistingMetadataStore metadataStore = new PropertiesPersistingMetadataStore();
metadataStore.setBaseDirectory(tempFolder.getRoot().getAbsolutePath());
return metadataStore;
}
@Bean
public IntegrationFlow feedFlow() {
return IntegrationFlows
.from(Feed.inboundAdapter(this.feedResource, "feedTest")
.metadataStore(metadataStore())
.preserveWireFeed(true),
e -> e.poller(p -> p.fixedDelay(100)))
.channel(c -> c.queue("entries"))
.get();
}
}
}

29
spring-integration-feed/src/test/java/org/springframework/integration/feed/inbound/FeedEntryMessageSourceTests.java
View File

@@ -26,6 +26,7 @@ import java.net.URL;
import org.junit.Before;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.io.ClassPathResource;
import org.springframework.integration.metadata.PropertiesPersistingMetadataStore;
@@ -39,13 +40,11 @@ import com.rometools.rome.feed.synd.SyndEntry;
* @author Gary Russell
* @author Aaron Loes
* @author Artem Bilan
*
* @since 2.0
*/
@SuppressWarnings("deprecation")
public class FeedEntryMessageSourceTests {
private final com.rometools.fetcher.FeedFetcher feedFetcher = new FileUrlFeedFetcher();
@Before
public void prepare() {
File metadataStoreFile = new File(System.getProperty("java.io.tmpdir") + "/spring-integration/",
@@ -65,7 +64,7 @@ public class FeedEntryMessageSourceTests {
@Test
public void testReceiveFeedWithNoEntries() throws Exception {
URL url = new ClassPathResource("org/springframework/integration/feed/empty.rss").getURL();
FeedEntryMessageSource feedEntrySource = new FeedEntryMessageSource(url, "foo", this.feedFetcher);
FeedEntryMessageSource feedEntrySource = new FeedEntryMessageSource(url, "foo");
feedEntrySource.setBeanName("feedReader");
feedEntrySource.setBeanFactory(mock(BeanFactory.class));
feedEntrySource.afterPropertiesSet();
@@ -74,8 +73,8 @@ public class FeedEntryMessageSourceTests {
@Test
public void testReceiveFeedWithEntriesSorted() throws Exception {
URL url = new ClassPathResource("org/springframework/integration/feed/sample.rss").getURL();
FeedEntryMessageSource source = new FeedEntryMessageSource(url, "foo", this.feedFetcher);
ClassPathResource resource = new ClassPathResource("org/springframework/integration/feed/sample.rss");
FeedEntryMessageSource source = new FeedEntryMessageSource(resource, "foo");
source.setComponentName("feedReader");
source.setBeanFactory(mock(BeanFactory.class));
source.afterPropertiesSet();
@@ -94,8 +93,8 @@ public class FeedEntryMessageSourceTests {
// account when determining if the feed entry has been seen before
@Test
public void testEntryHavingBeenUpdatedAfterPublishAndRepeat() throws Exception {
URL url = new ClassPathResource("org/springframework/integration/feed/atom.xml").getURL();
FeedEntryMessageSource feedEntrySource = new FeedEntryMessageSource(url, "foo", this.feedFetcher);
ClassPathResource resource = new ClassPathResource("org/springframework/integration/feed/atom.xml");
FeedEntryMessageSource feedEntrySource = new FeedEntryMessageSource(resource, "foo");
feedEntrySource.setBeanName("feedReader");
PropertiesPersistingMetadataStore metadataStore = new PropertiesPersistingMetadataStore();
metadataStore.afterPropertiesSet();
@@ -114,7 +113,7 @@ public class FeedEntryMessageSourceTests {
metadataStore.afterPropertiesSet();
// now test that what's been read is no longer retrieved
feedEntrySource = new FeedEntryMessageSource(url, "foo", this.feedFetcher);
feedEntrySource = new FeedEntryMessageSource(resource, "foo");
feedEntrySource.setBeanName("feedReader");
metadataStore = new PropertiesPersistingMetadataStore();
metadataStore.afterPropertiesSet();
@@ -128,8 +127,8 @@ public class FeedEntryMessageSourceTests {
// and no duplicate entries are retrieved
@Test
public void testReceiveFeedWithRealEntriesAndRepeatWithPersistentMetadataStore() throws Exception {
URL url = new ClassPathResource("org/springframework/integration/feed/sample.rss").getURL();
FeedEntryMessageSource feedEntrySource = new FeedEntryMessageSource(url, "foo", this.feedFetcher);
ClassPathResource resource = new ClassPathResource("org/springframework/integration/feed/sample.rss");
FeedEntryMessageSource feedEntrySource = new FeedEntryMessageSource(resource, "foo");
feedEntrySource.setBeanName("feedReader");
PropertiesPersistingMetadataStore metadataStore = new PropertiesPersistingMetadataStore();
metadataStore.afterPropertiesSet();
@@ -154,7 +153,7 @@ public class FeedEntryMessageSourceTests {
metadataStore.afterPropertiesSet();
// now test that what's been read is no longer retrieved
feedEntrySource = new FeedEntryMessageSource(url, "foo", this.feedFetcher);
feedEntrySource = new FeedEntryMessageSource(resource, "foo");
feedEntrySource.setBeanName("feedReader");
metadataStore = new PropertiesPersistingMetadataStore();
metadataStore.afterPropertiesSet();
@@ -170,8 +169,8 @@ public class FeedEntryMessageSourceTests {
// no persistent MetadataStore is provided and the same entries are retrieved again
@Test
public void testReceiveFeedWithRealEntriesAndRepeatNoPersistentMetadataStore() throws Exception {
URL url = new ClassPathResource("org/springframework/integration/feed/sample.rss").getURL();
FeedEntryMessageSource feedEntrySource = new FeedEntryMessageSource(url, "foo", this.feedFetcher);
ClassPathResource resource = new ClassPathResource("org/springframework/integration/feed/sample.rss");
FeedEntryMessageSource feedEntrySource = new FeedEntryMessageSource(resource, "foo");
feedEntrySource.setBeanName("feedReader");
feedEntrySource.setBeanFactory(mock(BeanFactory.class));
feedEntrySource.afterPropertiesSet();
@@ -191,7 +190,7 @@ public class FeedEntryMessageSourceTests {
// UNLIKE the previous test
// now test that what's been read is read AGAIN
feedEntrySource = new FeedEntryMessageSource(url, "foo", this.feedFetcher);
feedEntrySource = new FeedEntryMessageSource(resource, "foo");
feedEntrySource.setBeanName("feedReader");
feedEntrySource.setBeanFactory(mock(BeanFactory.class));
feedEntrySource.afterPropertiesSet();

116
spring-integration-feed/src/test/java/org/springframework/integration/feed/inbound/FileUrlFeedFetcher.java
View File

@@ -1,116 +0,0 @@
/*
* Copyright 2002-2014 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.feed.inbound;
import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.URL;
import java.net.URLConnection;
import java.util.zip.GZIPInputStream;
import org.springframework.util.Assert;
import com.rometools.rome.feed.synd.SyndFeed;
import com.rometools.rome.io.FeedException;
import com.rometools.rome.io.SyndFeedInput;
import com.rometools.rome.io.XmlReader;
/**
* @author Oleg Zhurakousky
* @author Mark Fisher
* @author Artem Bilan
* @since 2.0
* @deprecated since 4.3 because 'rome-fetcher-1.6.0' is deprecated.
* Will be revised in 5.0 in favor of ROME 2.0
*
*/
@SuppressWarnings("deprecation")
@Deprecated
class FileUrlFeedFetcher extends com.rometools.fetcher.impl.AbstractFeedFetcher {
@Override
public SyndFeed retrieveFeed(URL feedUrl)
throws IOException, FeedException, com.rometools.fetcher.FetcherException {
Assert.notNull(feedUrl, "feedUrl must not be null");
URLConnection connection = feedUrl.openConnection();
com.rometools.fetcher.impl.SyndFeedInfo syndFeedInfo = new com.rometools.fetcher.impl.SyndFeedInfo();
this.refreshFeedInfo(feedUrl, syndFeedInfo, connection);
return syndFeedInfo.getSyndFeed();
}
@Override
public SyndFeed retrieveFeed(String userAgent, URL url)
throws IllegalArgumentException, IOException, FeedException, com.rometools.fetcher.FetcherException {
return retrieveFeed(url);
}
private void refreshFeedInfo(URL feedUrl, com.rometools.fetcher.impl.SyndFeedInfo syndFeedInfo,
URLConnection connection)
throws IOException, FeedException {
// need to always set the URL because this may have changed due to 3xx redirects
syndFeedInfo.setUrl(connection.getURL());
// the ID is a persistent value that should stay the same
// even if the URL for the feed changes (eg, by 3xx redirects)
syndFeedInfo.setId(feedUrl.toString());
// This will be 0 if the server doesn't support or isn't setting the last modified header
syndFeedInfo.setLastModified(connection.getLastModified());
// get the contents
InputStream inputStream = null;
try {
inputStream = connection.getInputStream();
SyndFeed syndFeed = this.readFeedFromStream(inputStream, connection);
syndFeedInfo.setSyndFeed(syndFeed);
}
finally {
try {
inputStream.close();
}
catch (Exception e) {
// ignore
}
}
}
private SyndFeed readFeedFromStream(InputStream inputStream, URLConnection connection)
throws IOException, FeedException {
BufferedInputStream bufferedInputStream;
if ("gzip".equalsIgnoreCase(connection.getContentEncoding())) {
// handle gzip encoded content
bufferedInputStream = new BufferedInputStream(new GZIPInputStream(inputStream));
}
else {
bufferedInputStream = new BufferedInputStream(inputStream);
}
XmlReader reader = null;
if (connection.getHeaderField("Content-Type") != null) {
reader = new XmlReader(bufferedInputStream, connection.getHeaderField("Content-Type"), true);
}
else {
reader = new XmlReader(bufferedInputStream, true);
}
SyndFeedInput syndFeedInput = new SyndFeedInput();
syndFeedInput.setPreserveWireFeed(isPreserveWireFeed());
SyndFeed feed = syndFeedInput.build(reader);
fireEvent(com.rometools.fetcher.FetcherEvent.EVENT_TYPE_FEED_RETRIEVED, connection, feed);
return feed;
}
}

2
spring-integration-feed/src/test/java/org/springframework/integration/feed/sample.rss
View File

@@ -22,7 +22,7 @@ Spring Integration adapters
</title>
<link>http://www.springsource.org/extensions/se-sia</link>
<description>
Spring Integration adapters are realy cool
Spring Integration adapters are really cool
</description>
<pubDate>Tue, 23 Apr 2010 12:34:58 EST</pubDate>
</item>

61
src/reference/asciidoc/feed.adoc
View File

@@ -1,7 +1,8 @@
[[feed]]
== Feed Adapter
Spring Integration provides support for Syndication via Feed Adapters
Spring Integration provides support for Syndication via Feed Adapters.
The implementation is based on the https://rometools.github.io/rome/[ROME Framework].
[[feed-intro]]
=== Introduction
@@ -26,9 +27,9 @@ Below is an example configuration:
[source,xml]
----
<int-feed:inbound-channel-adapter id="feedAdapter"
channel="feedChannel"
url="http://feeds.bbci.co.uk/news/rss.xml">
<int:poller fixed-rate="10000" max-messages-per-poll="100" />
channel="feedChannel"
url="http://feeds.bbci.co.uk/news/rss.xml">
<int:poller fixed-rate="10000" max-messages-per-poll="100" />
</int-feed:inbound-channel-adapter>
----
@@ -44,7 +45,7 @@ However, one important thing you must understand with regard to Feeds is that it
When an Inbound Feed adapter is started, it does the first poll and receives a `com.sun.syndication.feed.synd.SyndEntryFeed` instance.
That is an object that contains multiple `SyndEntry` objects.
Each entry is stored in the local entry queue and is released based on the value in the `max-messages-per-poll` attribute such that each Message will contain a single entry.
If during retrieval of the entries from the entry queue the queue had become empty, the adapter will attempt to update the Feed thereby populating the queue with more entries (SyndEntry instances) if available.
If during retrieval of the entries from the entry queue the queue had become empty, the adapter will attempt to update the Feed thereby populating the queue with more entries (`SyndEntry` instances) if available.
Otherwise the next attempt to poll for a feed will be determined by the trigger of the poller (e.g., every 10 seconds in the above configuration).
_Duplicate Entries_
@@ -52,6 +53,52 @@ _Duplicate Entries_
Polling for a Feed might result in entries that have already been processed ("I already read that news item, why are you showing it to me again?").
Spring Integration provides a convenient mechanism to eliminate the need to worry about duplicate entries.
Each feed entry will have a _published date_ field.
Every time a new Message is generated and sent, Spring Integration will store the value of the latest _published date_ in an instance of the `MetadataStore` strategy (<<metadata-store>>).
Every time a new `Message` is generated and sent, Spring Integration will store the value of the latest _published date_ in an instance of the `MetadataStore` strategy (<<metadata-store>>).
NOTE: The key used to persist the latest _published date_ is the value of the (required) `id` attribute of the Feed Inbound Channel Adapter component plus the `feedUrl` from the adapter's configuration.
NOTE: The key used to persist the latest _published date_ is the value of the (required) `id` attribute of the Feed Inbound Channel Adapter component plus the `feedUrl` (if any) from the adapter's configuration.
_Other Options_
Starting with _version 5.0_, the deprecated `com.rometools.fetcher.FeedFetcher` option has been removed and an overloaded `FeedEntryMessageSource` constructor for an `org.springframework.core.io.Resource` is provided.
This is useful when Feed source isn't an HTTP endpoint, but any other resource, local or remote on FTP, for example.
In the `FeedEntryMessageSource` logic such a resource (or provided `URL`) is parsed by the `SyndFeedInput` to the `SyndFeed` object for processing mentioned above.
A customized `SyndFeedInput` (for example with the `allowDoctypes` option) instance also can be injected to the `FeedEntryMessageSource`.
[[feed-java-configuration]]
=== Java DSL and Annotation configuration
The following Spring Boot application provides an example of configuring the Inbound Adapter using the Java DSL:
[source, java]
----
@SpringBootApplication
public class FeedJavaApplication {
public static void main(String[] args) {
new SpringApplicationBuilder(FeedJavaApplication.class)
.web(false)
.run(args);
}
@Value("org/springframework/integration/feed/sample.rss")
private Resource feedResource;
@Bean
public MetadataStore metadataStore() {
PropertiesPersistingMetadataStore metadataStore = new PropertiesPersistingMetadataStore();
metadataStore.setBaseDirectory(tempFolder.getRoot().getAbsolutePath());
return metadataStore;
}
@Bean
public IntegrationFlow feedFlow() {
return IntegrationFlows
.from(Feed.inboundAdapter(this.feedResource, "feedTest")
.metadataStore(metadataStore()),
e -> e.poller(p -> p.fixedDelay(100)))
.channel(c -> c.queue("entries"))
.get();
}
}
----

8
src/reference/asciidoc/whats-new.adoc
View File

@@ -40,9 +40,15 @@ See <<gateway-error-handling>> for more information.
Some inconsistencies with rendering IMAP mail content have been resolved.
See <<imap-format-important, the note in the Mail-Receiving Channel Adapter Section>> for more information.
==== Feed Changes
Instead of the `com.rometools.fetcher.FeedFetcher`, which is deprecated in ROME, a new `Resource` property has been introduced to the `FeedEntryMessageSource`.
See <<feed>> for more information.
==== File Changes
The new `FileHeaders.RELATIVE_PATH` Message header has been introduced to prepresent relative path in the `FileReadingMessageSource`.
The new `FileHeaders.RELATIVE_PATH` Message header has been introduced to represent relative path in the `FileReadingMessageSource`.
See <<file-reading>> for more information.
==== (S)FTP Changes