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

INT-786, more refactoring and tests, added FileUrlFeedFetcher which will allow feed URLs to be specified as file:// (mainly for testing)

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-10-17 10:05:54 -04:00
parent 24fa72aab2
commit 59c525a490
12 changed files with 461 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-feed/.project
View File

@@ -20,8 +20,14 @@
<arguments>
</arguments>
</buildCommand>
<buildCommand>
<name>org.springframework.ide.eclipse.core.springbuilder</name>
<arguments>
</arguments>
</buildCommand>
</buildSpec>
<natures>
<nature>org.springframework.ide.eclipse.core.springnature</nature>
<nature>org.maven.ide.eclipse.maven2Nature</nature>
<nature>org.eclipse.jdt.core.javanature</nature>
<nature>org.eclipse.wst.common.project.facet.core.nature</nature>

82
spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedEntryReaderMessageSource.java
View File

@@ -23,7 +23,6 @@ import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.springframework.context.Lifecycle;
import org.springframework.integration.Message;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
@@ -41,20 +40,20 @@ import com.sun.syndication.feed.synd.SyndFeed;
* @author Mario Gray
* @author Oleg Zhurakousky
*/
public class FeedEntryReaderMessageSource extends IntegrationObjectSupport implements MessageSource<SyndEntry>, Lifecycle {
public class FeedEntryReaderMessageSource extends IntegrationObjectSupport implements MessageSource<SyndEntry>{
private volatile Map<String, String> persisterMap = new ConcurrentHashMap<String, String>();
private volatile Queue<SyndEntry> entries = new ConcurrentLinkedQueue<SyndEntry>();
private volatile FeedReaderMessageSource feedReaderMessageSource;
private final Object monitor = new Object();
private volatile String feedMetadataIdKey;
private volatile String feedUrl;
private volatile boolean running;
private volatile boolean initialized;
private volatile long lastTime = -1;
private Comparator<SyndEntry> syndEntryComparator = new Comparator<SyndEntry>() {
public int compare(SyndEntry syndEntry, SyndEntry syndEntry1) {
long x = sortId(syndEntry) - sortId(syndEntry1);
long x = syndEntry.getPublishedDate().getTime() -
syndEntry1.getPublishedDate().getTime();
if (x < -1) {
return -1;
}
@@ -65,13 +64,10 @@ public class FeedEntryReaderMessageSource extends IntegrationObjectSupport imple
}
};
public void setFeedUrl(String feedUrl) {
this.feedUrl = feedUrl;
}
public String getFeedUrl() {
return feedUrl;
}
public FeedEntryReaderMessageSource(FeedReaderMessageSource feedReaderMessageSource) {
Assert.notNull(feedReaderMessageSource, "'feedReaderMessageSource' must not be null");
this.feedReaderMessageSource = feedReaderMessageSource;
}
/**
* Allows you to provide your own implementation of 'persisterMap' instead of relying on
* your own which is in-memory.
@@ -83,25 +79,21 @@ public class FeedEntryReaderMessageSource extends IntegrationObjectSupport imple
this.persisterMap = persisterMap;
}
public void setRunning(boolean running) {
this.running = running;
}
public boolean isRunning() {
return running;
}
public void stop() {
this.feedReaderMessageSource.stop();
this.setRunning(false);
}
public String getComponentType(){
return "feed:inbound-channel-adapter";
}
public Message<SyndEntry> receive() {
Assert.isTrue(this.initialized, "'FeedEntryReaderMessageSource' must be initialized before it can produce Messages");
SyndEntry se = doReceieve();
if (se == null) {
return null;
}
return MessageBuilder.withPayload(se).build();
}
@SuppressWarnings("unchecked")
public SyndEntry receiveSyndEntry() {
private SyndEntry doReceieve() {
synchronized (this.monitor) {
SyndEntry nextUp = pollAndCache();
@@ -115,58 +107,36 @@ public class FeedEntryReaderMessageSource extends IntegrationObjectSupport imple
if (null != feedEntries) {
Collections.sort(feedEntries, syndEntryComparator);
for (SyndEntry se : feedEntries) {
long sort = this.sortId(se);
if (sort > this.lastTime)
entries.add(se);
long publishedTime = se.getPublishedDate().getTime();
if (publishedTime > this.lastTime){
entries.add(se);
}
}
}
}
return pollAndCache();
}
}
public Message<SyndEntry> receive() {
SyndEntry se = receiveSyndEntry();
if (se == null) {
return null;
}
return MessageBuilder.withPayload(se).build();
}
public void start() {
this.feedReaderMessageSource.start();
this.setRunning(true);
}
private long sortId(SyndEntry entry) {
return entry.getPublishedDate().getTime();
}
@Override
protected void onInit() throws Exception {
Assert.notNull(this.feedUrl, "the feedUrl can't be null");
this.feedReaderMessageSource = new FeedReaderMessageSource();
this.feedReaderMessageSource.setFeedUrl(this.feedUrl);
this.feedReaderMessageSource.setBeanName(this.getComponentName());
this.feedReaderMessageSource.afterPropertiesSet();
// setup persistence of metadata
this.feedMetadataIdKey = FeedEntryReaderMessageSource.class.getName() + "#" + feedUrl;
this.feedMetadataIdKey = FeedEntryReaderMessageSource.class.getName() + "#" + feedReaderMessageSource.getFeedUrl();
String lastTime = (String) this.persisterMap.get(this.feedMetadataIdKey);
if (lastTime != null && !lastTime.trim().equalsIgnoreCase("")) {
this.lastTime = Long.parseLong(lastTime);
}
this.initialized = true;
}
private SyndEntry pollAndCache() {
private SyndEntry pollAndCache() {
SyndEntry next = this.entries.poll();
if (next == null) {
return null;
}
this.lastTime = sortId(next);
this.lastTime = next.getPublishedDate().getTime();
this.persisterMap.put(this.feedMetadataIdKey, this.lastTime + "");
return next;
}

63
spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedReaderMessageSource.java
View File

@@ -19,7 +19,6 @@ import java.net.URL;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.context.IntegrationObjectSupport;
@@ -30,6 +29,7 @@ import org.springframework.util.Assert;
import com.sun.syndication.feed.synd.SyndFeed;
import com.sun.syndication.fetcher.FetcherEvent;
import com.sun.syndication.fetcher.FetcherListener;
import com.sun.syndication.fetcher.impl.AbstractFeedFetcher;
import com.sun.syndication.fetcher.impl.FeedFetcherCache;
import com.sun.syndication.fetcher.impl.HashMapFeedInfoCache;
import com.sun.syndication.fetcher.impl.HttpURLFeedFetcher;
@@ -43,48 +43,42 @@ import com.sun.syndication.fetcher.impl.HttpURLFeedFetcher;
* @author Mario Gray
* @author Oleg Zhurakousky
*/
class FeedReaderMessageSource extends IntegrationObjectSupport
implements InitializingBean, Lifecycle, MessageSource<SyndFeed> {
public class FeedReaderMessageSource extends IntegrationObjectSupport
implements InitializingBean, MessageSource<SyndFeed> {
private volatile boolean running;
private volatile String feedUrl;
private volatile URL feedURLObject;
private final AbstractFeedFetcher fetcher;
private final Object syndFeedMonitor = new Object();
private volatile URL feedUrl;
private volatile FeedFetcherCache fetcherCache;
private volatile HttpURLFeedFetcher fetcher;
private volatile ConcurrentLinkedQueue<SyndFeed> syndFeeds;
private volatile ConcurrentLinkedQueue<SyndFeed> syndFeeds = new ConcurrentLinkedQueue<SyndFeed>();
private volatile MyFetcherListener myFetcherListener;
private final Object syndFeedMonitor = new Object();
public FeedReaderMessageSource() {
syndFeeds = new ConcurrentLinkedQueue<SyndFeed>();
}
public void setFeedUrl(final String feedUrl) {
public FeedReaderMessageSource(URL feedUrl) {
this.feedUrl = feedUrl;
if (feedUrl.getProtocol().equals("file")){
fetcher = new FileUrlFeedFetcher();
}
else if (feedUrl.getProtocol().equals("http")){
fetcherCache = HashMapFeedInfoCache.getInstance();
fetcher = new HttpURLFeedFetcher(fetcherCache);
}
else{
throw new IllegalArgumentException("Unsupported URL protocol: " + feedUrl.getProtocol());
}
}
public String getFeedUrl() {
public URL getFeedUrl() {
return feedUrl;
}
public void start() {
this.running = true;
}
public void stop() {
this.running = false;
}
public boolean isRunning() {
return this.running;
}
public SyndFeed receiveSyndFeed() {
SyndFeed returnedSyndFeed = null;
try {
synchronized (syndFeedMonitor) {
returnedSyndFeed = fetcher.retrieveFeed(this.feedURLObject);
returnedSyndFeed = fetcher.retrieveFeed(this.feedUrl);
logger.debug("attempted to retrieve feed '" + this.feedUrl + "'");
if (returnedSyndFeed == null) {
@@ -93,7 +87,8 @@ class FeedReaderMessageSource extends IntegrationObjectSupport
}
}
} catch (Exception e) {
throw new MessagingException("Exception thrown when trying to retrive feed at url '" + this.feedURLObject + "'", e);
e.printStackTrace();
throw new MessagingException("Exception thrown when trying to retrive feed at url '" + this.feedUrl + "'", e);
}
return returnedSyndFeed;
@@ -106,20 +101,16 @@ class FeedReaderMessageSource extends IntegrationObjectSupport
return null;
}
return MessageBuilder.withPayload(syndFeed).setHeader(FeedConstants.FEED_URL, this.feedURLObject).build();
return MessageBuilder.withPayload(syndFeed).setHeader(FeedConstants.FEED_URL, this.feedUrl).build();
}
@Override
protected void onInit() throws Exception {
// myFetcherListener = new MyFetcherListener();
fetcherCache = HashMapFeedInfoCache.getInstance();
fetcher = new HttpURLFeedFetcher(fetcherCache);
fetcher.addFetcherEventListener(myFetcherListener);
Assert.notNull(this.feedUrl, "the feedURL can't be null");
feedURLObject = new URL(this.feedUrl);
}
class MyFetcherListener implements FetcherListener {

109
spring-integration-feed/src/main/java/org/springframework/integration/feed/FileUrlFeedFetcher.java Normal file
View File

@@ -0,0 +1,109 @@
/*
* Copyright 2010 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;
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 com.sun.syndication.feed.synd.SyndFeed;
import com.sun.syndication.fetcher.FetcherEvent;
import com.sun.syndication.fetcher.FetcherException;
import com.sun.syndication.fetcher.impl.AbstractFeedFetcher;
import com.sun.syndication.fetcher.impl.SyndFeedInfo;
import com.sun.syndication.io.FeedException;
import com.sun.syndication.io.SyndFeedInput;
import com.sun.syndication.io.XmlReader;
/**
* @author Oleg Zhurakousky
* @since 2.0
*/
public class FileUrlFeedFetcher extends AbstractFeedFetcher {
/* (non-Javadoc)
* @see com.sun.syndication.fetcher.FeedFetcher#retrieveFeed(java.net.URL)
*/
public SyndFeed retrieveFeed(URL feedUrl) throws IllegalArgumentException,
IOException, FeedException, FetcherException {
if (feedUrl == null) {
throw new IllegalArgumentException("null is not a valid URL");
}
URLConnection connection = feedUrl.openConnection();
SyndFeedInfo syndFeedInfo = new SyndFeedInfo();
retrieveAndCacheFeed(feedUrl, syndFeedInfo, connection);
return syndFeedInfo.getSyndFeed();
}
protected void retrieveAndCacheFeed(URL feedUrl, SyndFeedInfo syndFeedInfo, URLConnection connection) throws IllegalArgumentException, FeedException, FetcherException, IOException {
resetFeedInfo(feedUrl, syndFeedInfo, connection);
}
protected void resetFeedInfo(URL orignalUrl, SyndFeedInfo syndFeedInfo, URLConnection connection) throws IllegalArgumentException, 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 persistant value that should stay the same even if the URL for the
// feed changes (eg, by 3xx redirects)
syndFeedInfo.setId(orignalUrl.toString());
// This will be 0 if the server doesn't support or isn't setting the last modified header
syndFeedInfo.setLastModified(new Long(connection.getLastModified()));
// get the contents
InputStream inputStream = null;
try {
inputStream = connection.getInputStream();
SyndFeed syndFeed = getSyndFeedFromStream(inputStream, connection);
syndFeedInfo.setSyndFeed(syndFeed);
} finally {
if (inputStream != null) {
inputStream.close();
}
}
}
private SyndFeed getSyndFeedFromStream(InputStream inputStream, URLConnection connection) throws IOException, IllegalArgumentException, FeedException {
SyndFeed feed = readSyndFeedFromStream(inputStream, connection);
fireEvent(FetcherEvent.EVENT_TYPE_FEED_RETRIEVED, connection, feed);
return feed;
}
private SyndFeed readSyndFeedFromStream(InputStream inputStream, URLConnection connection) throws IOException, IllegalArgumentException, FeedException {
BufferedInputStream is;
if ("gzip".equalsIgnoreCase(connection.getContentEncoding())) {
// handle gzip encoded content
is = new BufferedInputStream(new GZIPInputStream(inputStream));
} else {
is = new BufferedInputStream(inputStream);
}
XmlReader reader = null;
if (connection.getHeaderField("Content-Type") != null) {
reader = new XmlReader(is, connection.getHeaderField("Content-Type"), true);
} else {
reader = new XmlReader(is, true);
}
SyndFeedInput syndFeedInput = new SyndFeedInput();
syndFeedInput.setPreserveWireFeed(isPreserveWireFeed());
return syndFeedInput.build(reader);
}
}

12
spring-integration-feed/src/main/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParser.java
View File

@@ -32,9 +32,15 @@ public class FeedMessageSourceBeanDefinitionParser extends AbstractPollingInboun
@Override
protected String parseSource(final Element element, final ParserContext parserContext) {
BeanDefinitionBuilder feedBuilder =
BeanDefinitionBuilder feedEntryBuilder =
BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.feed.FeedEntryReaderMessageSource");
feedBuilder.addPropertyValue("feedUrl", element.getAttribute("feedUrl"));
return BeanDefinitionReaderUtils.registerWithGeneratedName(feedBuilder.getBeanDefinition(), parserContext.getRegistry());
BeanDefinitionBuilder feedBuilder =
BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.feed.FeedReaderMessageSource");
feedBuilder.addConstructorArgValue(element.getAttribute("feedUrl"));
feedEntryBuilder.addConstructorArgValue(feedBuilder.getBeanDefinition());
return BeanDefinitionReaderUtils.registerWithGeneratedName(feedEntryBuilder.getBeanDefinition(), parserContext.getRegistry());
}
}

12
spring-integration-feed/src/main/resources/org/springframework/integration/feed/config/spring-integration-feed-2.0.xsd
View File

@@ -37,6 +37,18 @@
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="feedUrl" type="xsd:string" use="required"/>
<xsd:attribute name="persisterMap" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation>
Allows you to inject Map
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="java.util.Map"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>

8
spring-integration-feed/src/test/java/log4j.properties Normal file
View File

@@ -0,0 +1,8 @@
log4j.rootCategory=WARN, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
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

80
spring-integration-feed/src/test/java/org/springframework/integration/feed/FeedEntryReaderMessageSourceTests.java Normal file
View File

@@ -0,0 +1,80 @@
/*
* Copyright 2002-2010 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;
import static junit.framework.Assert.assertEquals;
import static junit.framework.Assert.assertNull;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.when;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import org.junit.Test;
import org.springframework.integration.Message;
import com.sun.syndication.feed.synd.SyndEntry;
import com.sun.syndication.feed.synd.SyndFeed;
/**
* @author Oleg Zhurakousky
*
*/
public class FeedEntryReaderMessageSourceTests {
@Test(expected=IllegalArgumentException.class)
public void testFailureWhenNotInitialized(){
FeedEntryReaderMessageSource feedEntrySource = new FeedEntryReaderMessageSource(mock(FeedReaderMessageSource.class));
feedEntrySource.receive();
}
@Test
public void testReceieveFeedWithNoEntries(){
FeedReaderMessageSource feedReaderSource = mock(FeedReaderMessageSource.class);
SyndFeed feed = mock(SyndFeed.class);
when(feedReaderSource.receiveSyndFeed()).thenReturn(feed);
FeedEntryReaderMessageSource feedEntrySource = new FeedEntryReaderMessageSource(feedReaderSource);
feedEntrySource.afterPropertiesSet();
assertNull(feedEntrySource.receive());
}
@Test
public void testReceieveFeedWithEntriesSorted(){
FeedReaderMessageSource feedReaderSource = mock(FeedReaderMessageSource.class);
SyndFeed feed = mock(SyndFeed.class);
SyndEntry entry1 = mock(SyndEntry.class);
SyndEntry entry2 = mock(SyndEntry.class);
when(entry1.getPublishedDate()).thenReturn(new Date(System.currentTimeMillis()));
when(entry2.getPublishedDate()).thenReturn(new Date(System.currentTimeMillis()-10000));
List<SyndEntry> entries = new ArrayList<SyndEntry>();
entries.add(entry2);
entries.add(entry1);
when(feed.getEntries()).thenReturn(entries);
when(feedReaderSource.receiveSyndFeed()).thenReturn(feed);
FeedEntryReaderMessageSource feedEntrySource = new FeedEntryReaderMessageSource(feedReaderSource);
feedEntrySource.afterPropertiesSet();
Message<SyndEntry> entryMessage = feedEntrySource.receive();
assertEquals(entry2, entryMessage.getPayload());
entryMessage = feedEntrySource.receive();
assertEquals(entry1, entryMessage.getPayload());
reset(feed);
entryMessage = feedEntrySource.receive();
assertNull(entryMessage);
}
}

15
spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests-file-context.xml Normal file
View File

@@ -0,0 +1,15 @@
<?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"
xmlns:int-feed="http://www.springframework.org/schema/integration/feed"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
http://www.springframework.org/schema/integration/feed http://www.springframework.org/schema/integration/feed/spring-integration-feed-2.0.xsd">
<int-feed:inbound-channel-adapter id="feedAdapter" channel="feedChannel"
feedUrl="file:src/test/java/org/springframework/integration/feed/config/sample.rss">
<int:poller fixed-rate="10000" max-messages-per-poll="100" />
</int-feed:inbound-channel-adapter>
<int:channel id="feedChannel" />
</beans>

15
spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests-http-context.xml Normal file
View File

@@ -0,0 +1,15 @@
<?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"
xmlns:int-feed="http://www.springframework.org/schema/integration/feed"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
http://www.springframework.org/schema/integration/feed http://www.springframework.org/schema/integration/feed/spring-integration-feed-2.0.xsd">
<int-feed:inbound-channel-adapter id="feedAdapter" channel="feedChannel"
feedUrl="http://feeds.bbci.co.uk/news/rss.xml">
<int:poller fixed-rate="10000" max-messages-per-poll="100" />
</int-feed:inbound-channel-adapter>
<int:channel id="feedChannel" />
</beans>

101
spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests.java Normal file
View File

@@ -0,0 +1,101 @@
/*
* Copyright 2002-2010 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.config;
import static junit.framework.Assert.assertTrue;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.feed.FeedEntryReaderMessageSource;
import org.springframework.integration.feed.FeedReaderMessageSource;
import org.springframework.integration.feed.FileUrlFeedFetcher;
import org.springframework.integration.test.util.TestUtils;
import com.sun.syndication.fetcher.impl.AbstractFeedFetcher;
import com.sun.syndication.fetcher.impl.HttpURLFeedFetcher;
/**
* @author Oleg Zhurakousky
*
*/
public class FeedMessageSourceBeanDefinitionParserTests {
@Test
public void validateSuccessfullConfiguration(){
ApplicationContext context =
new ClassPathXmlApplicationContext("FeedMessageSourceBeanDefinitionParserTests-file-context.xml", this.getClass());
SourcePollingChannelAdapter adapter = context.getBean("feedAdapter", SourcePollingChannelAdapter.class);
FeedEntryReaderMessageSource source = (FeedEntryReaderMessageSource) TestUtils.getPropertyValue(adapter, "source");
FeedReaderMessageSource feedReaderMessageSource = (FeedReaderMessageSource) TestUtils.getPropertyValue(source, "feedReaderMessageSource");
AbstractFeedFetcher fetcher = (AbstractFeedFetcher) TestUtils.getPropertyValue(feedReaderMessageSource, "fetcher");
assertTrue(fetcher instanceof FileUrlFeedFetcher);
context =
new ClassPathXmlApplicationContext("FeedMessageSourceBeanDefinitionParserTests-http-context.xml", this.getClass());
adapter = context.getBean("feedAdapter", SourcePollingChannelAdapter.class);
source = (FeedEntryReaderMessageSource) TestUtils.getPropertyValue(adapter, "source");
feedReaderMessageSource = (FeedReaderMessageSource) TestUtils.getPropertyValue(source, "feedReaderMessageSource");
fetcher = (AbstractFeedFetcher) TestUtils.getPropertyValue(feedReaderMessageSource, "fetcher");
assertTrue(fetcher instanceof HttpURLFeedFetcher);
}
@Test
public void validateSuccessfullNewsRetrievalFile() throws Exception{
//Test file samples.rss has 3 news items
final CountDownLatch latch = new CountDownLatch(3);
MessageHandler handler = spy(new MessageHandler() {
public void handleMessage(Message<?> message) throws MessagingException {
latch.countDown();
}
});
ApplicationContext context =
new ClassPathXmlApplicationContext("FeedMessageSourceBeanDefinitionParserTests-file-context.xml", this.getClass());
DirectChannel feedChannel = context.getBean("feedChannel", DirectChannel.class);
feedChannel.subscribe(handler);
latch.await(5, TimeUnit.SECONDS);
verify(handler, times(3)).handleMessage(Mockito.any(Message.class));
}
@Test
public void validateSuccessfullNewsRetrievalHttp() throws Exception{
//Test file samples.rss has 3 news items
final CountDownLatch latch = new CountDownLatch(3);
MessageHandler handler = spy(new MessageHandler() {
public void handleMessage(Message<?> message) throws MessagingException {
latch.countDown();
}
});
ApplicationContext context =
new ClassPathXmlApplicationContext("FeedMessageSourceBeanDefinitionParserTests-http-context.xml", this.getClass());
DirectChannel feedChannel = context.getBean("feedChannel", DirectChannel.class);
feedChannel.subscribe(handler);
latch.await(5, TimeUnit.SECONDS);
verify(handler, atLeast(3)).handleMessage(Mockito.any(Message.class));
}
}

53
spring-integration-feed/src/test/java/org/springframework/integration/feed/config/sample.rss Normal file
View File

@@ -0,0 +1,53 @@
<rss version="2.0">
<channel>
<title>ASP @ BellaOnline</title>
<link>http://www.bellaonline.com/Site/asp</link>
<description>
Learn to program in ASP, and enhance your ASP skills to add great new functionality to your website!
</description>
<language>en-us</language>
<copyright>Copyright 2001-2005 BellaOnline.com
All Rights Reserved.</copyright>
<lastBuildDate>Tue, 12 Apr 2005 14:21:32 EST</lastBuildDate>
<ttl>240</ttl>
<image>
<url>http://www.bellaonline.com/images/bella.gif</url>
<title>ASP @ BellaOnline</title>
<link>http://asp.bellaonline.com</link>
</image>
<item>
<title>
Using ASP to Code an RSS Feed
</title>
<link>http://www.bellaonline.com/articles/art30646.asp</link>
<description>
RSS feeds let you easily syndicate your content to an end user or another website. ASP can help you easily create your own RSS feed for your website.
</description>
<pubDate>Tue, 12 Apr 2005 13:59:56 EST</pubDate>
</item>
<item>
<title>
RecordCount and Count
</title>
<link>http://www.bellaonline.com/articles/art30403.asp</link>
<description>
If you're trying to figure out how many records are in a given SQL result set, you can use either the RecordCount or Count command. Both work in different ways.
</description>
<pubDate>Sun, 3 Apr 2005 17:12:17 EST</pubDate>
</item>
<item>
<title>
Bubble Sort Code Technique
</title>
<link>http://www.bellaonline.com/articles/art29843.asp</link>
<description>
If you are sorting content into an order, one of the most simple techniques that exists is the bubble sort technique.
</description>
<pubDate>Wed, 16 Mar 2005 00:38:21 EST</pubDate>
</item>
</channel>
</rss>