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

INT-786 refactored Feed MessageSource and related classes

Browse Source
This commit is contained in:
Mark Fisher
2010-10-27 11:37:08 -04:00
parent ffdde36b1b
commit 35f2bc1444
10 changed files with 154 additions and 249 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedConstants.java
View File

@@ -1,28 +0,0 @@
/*
* 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;
/**
* Header keys for {@link FeedReaderMessageSource}.
*
* @author Josh Long
*/
public abstract class FeedConstants {
static public final String FEED_URL = "feedUrl";
}

85
spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedEntryReaderMessageSource.java → spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedEntryMessageSource.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.feed;
import java.net.URL;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
@@ -24,6 +25,7 @@ import java.util.concurrent.ConcurrentLinkedQueue;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.context.metadata.MetadataStore;
@@ -36,6 +38,12 @@ import org.springframework.util.StringUtils;
import com.sun.syndication.feed.synd.SyndEntry;
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;
/**
* This implementation of {@link MessageSource} will produce individual
@@ -46,12 +54,16 @@ import com.sun.syndication.feed.synd.SyndFeed;
* @author Oleg Zhurakousky
* @since 2.0
*/
public class FeedEntryReaderMessageSource extends IntegrationObjectSupport implements MessageSource<SyndEntry> {
public class FeedEntryMessageSource extends IntegrationObjectSupport implements MessageSource<SyndEntry> {
private final URL feedUrl;
private final AbstractFeedFetcher fetcher;
private final Queue<SyndFeed> feeds = new ConcurrentLinkedQueue<SyndFeed>();
private final Queue<SyndEntry> entries = new ConcurrentLinkedQueue<SyndEntry>();
private final FeedReaderMessageSource feedReaderMessageSource;
private volatile String metadataKey;
private volatile MetadataStore metadataStore;
@@ -64,10 +76,22 @@ public class FeedEntryReaderMessageSource extends IntegrationObjectSupport imple
private final Comparator<SyndEntry> syndEntryComparator = new SyndEntryComparator();
private final Object feedMonitor = new Object();
public FeedEntryReaderMessageSource(FeedReaderMessageSource feedReaderMessageSource) {
Assert.notNull(feedReaderMessageSource, "'feedReaderMessageSource' must not be null");
this.feedReaderMessageSource = feedReaderMessageSource;
public FeedEntryMessageSource(URL feedUrl) {
Assert.notNull(feedUrl, "feedUrl must not be null");
this.feedUrl = feedUrl;
if (feedUrl.getProtocol().equals("file")) {
this.fetcher = new FileUrlFeedFetcher();
}
else if (feedUrl.getProtocol().equals("http")) {
FeedFetcherCache fetcherCache = HashMapFeedInfoCache.getInstance();
this.fetcher = new HttpURLFeedFetcher(fetcherCache);
}
else {
throw new IllegalArgumentException("Unsupported URL protocol: " + feedUrl.getProtocol());
}
}
@@ -91,6 +115,7 @@ public class FeedEntryReaderMessageSource extends IntegrationObjectSupport imple
@Override
protected void onInit() throws Exception {
this.fetcher.addFetcherEventListener(new FeedQueueUpdatingFetcherListener());
if (this.metadataStore == null) {
// first try to look for a 'messageStore' in the context
BeanFactory beanFactory = this.getBeanFactory();
@@ -103,8 +128,7 @@ public class FeedEntryReaderMessageSource extends IntegrationObjectSupport imple
}
}
Assert.hasText(this.getComponentName(), "FeedEntryReaderMessageSource must have a name");
this.metadataKey = this.getComponentType() + "." + this.getComponentName()
+ "." + this.feedReaderMessageSource.getFeedUrl();
this.metadataKey = this.getComponentType() + "." + this.getComponentName() + "." + this.feedUrl;
String lastTimeValue = this.metadataStore.get(this.metadataKey);
if (StringUtils.hasText(lastTimeValue)) {
this.lastTime = Long.parseLong(lastTimeValue);
@@ -137,7 +161,7 @@ public class FeedEntryReaderMessageSource extends IntegrationObjectSupport imple
@SuppressWarnings("unchecked")
private void populateEntryList() {
SyndFeed syndFeed = this.feedReaderMessageSource.receiveFeed();
SyndFeed syndFeed = this.getFeed();
if (syndFeed != null) {
List<SyndEntry> retrievedEntries = (List<SyndEntry>) syndFeed.getEntries();
if (!CollectionUtils.isEmpty(retrievedEntries)) {
@@ -151,6 +175,28 @@ public class FeedEntryReaderMessageSource extends IntegrationObjectSupport imple
}
}
private SyndFeed getFeed() {
SyndFeed feed = null;
try {
synchronized (this.feedMonitor) {
feed = this.fetcher.retrieveFeed(this.feedUrl);
if (logger.isDebugEnabled()) {
logger.debug("retrieved feed at url '" + this.feedUrl + "'");
}
if (feed == null) {
if (logger.isDebugEnabled()) {
logger.debug("no feeds updated, returning null");
}
}
}
}
catch (Exception e) {
throw new MessagingException(
"Failed to retrieve feed at url '" + this.feedUrl + "'", e);
}
return feed;
}
private static class SyndEntryComparator implements Comparator<SyndEntry> {
@@ -159,4 +205,25 @@ public class FeedEntryReaderMessageSource extends IntegrationObjectSupport imple
}
}
private class FeedQueueUpdatingFetcherListener implements FetcherListener {
/**
* @see com.sun.syndication.fetcher.FetcherListener#fetcherEvent(com.sun.syndication.fetcher.FetcherEvent)
*/
public void fetcherEvent(final FetcherEvent event) {
String eventType = event.getEventType();
if (FetcherEvent.EVENT_TYPE_FEED_POLLED.equals(eventType)) {
logger.debug("\tEVENT: Feed Polled. URL = " + event.getUrlString());
}
else if (FetcherEvent.EVENT_TYPE_FEED_RETRIEVED.equals(eventType)) {
logger.debug("\tEVENT: Feed Retrieved. URL = " + event.getUrlString());
feeds.add(event.getFeed());
}
else if (FetcherEvent.EVENT_TYPE_FEED_UNCHANGED.equals(eventType)) {
logger.debug("\tEVENT: Feed Unchanged. URL = " + event.getUrlString());
}
}
}
}

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

@@ -1,137 +0,0 @@
/*
* 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 java.net.URL;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.support.MessageBuilder;
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;
/**
* This implementation of {@link MessageSource} will produce a Message whose payload is
* an instance of {@link SyndFeed} for a feed identified with the 'feedUrl' attribute.
*
* @author Josh Long
* @author Mario Gray
* @author Oleg Zhurakousky
* @author Mark Fisher
* @since 2.0
*/
public class FeedReaderMessageSource extends IntegrationObjectSupport implements InitializingBean, MessageSource<SyndFeed> {
private final URL feedUrl;
private final AbstractFeedFetcher fetcher;
private volatile FeedFetcherCache fetcherCache;
private final ConcurrentLinkedQueue<SyndFeed> feeds = new ConcurrentLinkedQueue<SyndFeed>();
private final Object feedMonitor = new Object();
public FeedReaderMessageSource(URL feedUrl) {
this.feedUrl = feedUrl;
if (feedUrl.getProtocol().equals("file")) {
this.fetcher = new FileUrlFeedFetcher();
}
else if (feedUrl.getProtocol().equals("http")) {
this.fetcherCache = HashMapFeedInfoCache.getInstance();
this.fetcher = new HttpURLFeedFetcher(fetcherCache);
}
else {
throw new IllegalArgumentException("Unsupported URL protocol: " + feedUrl.getProtocol());
}
}
URL getFeedUrl() {
return this.feedUrl;
}
@Override
protected void onInit() throws Exception {
fetcher.addFetcherEventListener(new FeedQueueUpdatingFetcherListener());
Assert.notNull(this.feedUrl, "the feedUrl must not be null");
}
SyndFeed receiveFeed() {
SyndFeed feed = null;
try {
synchronized (this.feedMonitor) {
feed = this.fetcher.retrieveFeed(this.feedUrl);
if (logger.isDebugEnabled()) {
logger.debug("retrieved feed at url '" + this.feedUrl + "'");
}
if (feed == null) {
if (logger.isDebugEnabled()) {
logger.debug("no feeds updated, returning null");
}
}
}
}
catch (Exception e) {
throw new MessagingException(
"Failed to retrieve feed at url '" + this.feedUrl + "'", e);
}
return feed;
}
public Message<SyndFeed> receive() {
SyndFeed feed = this.receiveFeed();
if (feed == null) {
return null;
}
return MessageBuilder.withPayload(feed).setHeader(FeedConstants.FEED_URL, this.feedUrl).build();
}
private class FeedQueueUpdatingFetcherListener implements FetcherListener {
/**
* @see com.sun.syndication.fetcher.FetcherListener#fetcherEvent(com.sun.syndication.fetcher.FetcherEvent)
*/
public void fetcherEvent(final FetcherEvent event) {
String eventType = event.getEventType();
if (FetcherEvent.EVENT_TYPE_FEED_POLLED.equals(eventType)) {
logger.debug("\tEVENT: Feed Polled. URL = " + event.getUrlString());
}
else if (FetcherEvent.EVENT_TYPE_FEED_RETRIEVED.equals(eventType)) {
logger.debug("\tEVENT: Feed Retrieved. URL = " + event.getUrlString());
feeds.add(event.getFeed());
}
else if (FetcherEvent.EVENT_TYPE_FEED_UNCHANGED.equals(eventType)) {
logger.debug("\tEVENT: Feed Unchanged. URL = " + event.getUrlString());
}
}
}
}

6
spring-integration-feed/src/main/java/org/springframework/integration/feed/FileUrlFeedFetcher.java
View File

@@ -39,10 +39,10 @@ import com.sun.syndication.io.XmlReader;
* @author Mark Fisher
* @since 2.0
*/
public class FileUrlFeedFetcher extends AbstractFeedFetcher {
class FileUrlFeedFetcher extends AbstractFeedFetcher {
/*
* (non-Javadoc)
/**
* Retrieve a SyndFeed for the given URL.
* @see com.sun.syndication.fetcher.FeedFetcher#retrieveFeed(java.net.URL)
*/
public SyndFeed retrieveFeed(URL feedUrl) throws IOException, FeedException, FetcherException {

18
spring-integration-feed/src/main/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParser.java → spring-integration-feed/src/main/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParser.java
View File

@@ -25,24 +25,22 @@ import org.springframework.integration.config.xml.AbstractPollingInboundChannelA
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
/**
* Handles parsing the configuration for the feed inbound channel adapter.
* Handles parsing the configuration for the feed inbound-channel-adapter.
*
* @author Josh Long
* @author Oleg Zhurakousky
* @author Mark Fisher
* @since 2.0
*/
public class FeedMessageSourceBeanDefinitionParser extends AbstractPollingInboundChannelAdapterParser {
public class FeedInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
@Override
protected String parseSource(final Element element, final ParserContext parserContext) {
BeanDefinitionBuilder feedEntryBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.feed.FeedEntryReaderMessageSource");
BeanDefinitionBuilder feedBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.feed.FeedReaderMessageSource");
feedBuilder.addConstructorArgValue(element.getAttribute("url"));
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(feedEntryBuilder, element, "metadata-store");
feedEntryBuilder.addConstructorArgValue(feedBuilder.getBeanDefinition());
return BeanDefinitionReaderUtils.registerWithGeneratedName(feedEntryBuilder.getBeanDefinition(), parserContext.getRegistry());
BeanDefinitionBuilder sourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.feed.FeedEntryMessageSource");
sourceBuilder.addConstructorArgValue(element.getAttribute("url"));
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(sourceBuilder, element, "metadata-store");
return BeanDefinitionReaderUtils.registerWithGeneratedName(sourceBuilder.getBeanDefinition(), parserContext.getRegistry());
}
}

3
spring-integration-feed/src/main/java/org/springframework/integration/feed/config/FeedNamespaceHandler.java
View File

@@ -22,12 +22,13 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
* NamespaceHandler for the feed module.
*
* @author Josh Long
* @author Mark Fisher
* @since 2.0
*/
public class FeedNamespaceHandler extends AbstractIntegrationNamespaceHandler {
public void init() {
registerBeanDefinitionParser("inbound-channel-adapter", new FeedMessageSourceBeanDefinitionParser());
registerBeanDefinitionParser("inbound-channel-adapter", new FeedInboundChannelAdapterParser());
}
}