diff --git a/spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedEntryReaderMessageSource.java b/spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedEntryReaderMessageSource.java index e3b22ad1e4..22e3ef1a2f 100644 --- a/spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedEntryReaderMessageSource.java +++ b/spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedEntryReaderMessageSource.java @@ -137,7 +137,7 @@ public class FeedEntryReaderMessageSource extends IntegrationObjectSupport imple @SuppressWarnings("unchecked") private void populateEntryList() { - SyndFeed syndFeed = this.feedReaderMessageSource.receiveSyndFeed(); + SyndFeed syndFeed = this.feedReaderMessageSource.receiveFeed(); if (syndFeed != null) { List retrievedEntries = (List) syndFeed.getEntries(); if (!CollectionUtils.isEmpty(retrievedEntries)) { diff --git a/spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedReaderMessageSource.java b/spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedReaderMessageSource.java index 7f1dfa5910..d72aaa6beb 100644 --- a/spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedReaderMessageSource.java +++ b/spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedReaderMessageSource.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.feed; import java.net.URL; @@ -34,99 +35,103 @@ 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 {@link SyndFeed} for a feed identified - * with 'feedUrl' attribute. - * + * 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 { - +public class FeedReaderMessageSource extends IntegrationObjectSupport implements InitializingBean, MessageSource { + + private final URL feedUrl; + private final AbstractFeedFetcher fetcher; - private final Object syndFeedMonitor = new Object(); - - private volatile URL feedUrl; - private volatile FeedFetcherCache fetcherCache; - private volatile ConcurrentLinkedQueue syndFeeds = new ConcurrentLinkedQueue(); - private volatile MyFetcherListener myFetcherListener; - - - 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 URL getFeedUrl() { - return feedUrl; - } - - public SyndFeed receiveSyndFeed() { - SyndFeed returnedSyndFeed = null; - try { - synchronized (syndFeedMonitor) { - returnedSyndFeed = fetcher.retrieveFeed(this.feedUrl); - logger.debug("attempted to retrieve feed '" + this.feedUrl + "'"); + private volatile FeedFetcherCache fetcherCache; - if (returnedSyndFeed == null) { - logger.debug("no feeds updated, return null!"); - return null; - } - } - } catch (Exception e) { - throw new MessagingException("Exception thrown when trying to retrive feed at url '" + this.feedUrl + "'", e); - } + private final ConcurrentLinkedQueue feeds = new ConcurrentLinkedQueue(); - return returnedSyndFeed; - } + private final Object feedMonitor = new Object(); - public Message receive() { - SyndFeed syndFeed = this.receiveSyndFeed(); - if (null == syndFeed) { - return null; - } + 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()); + } + } - return MessageBuilder.withPayload(syndFeed).setHeader(FeedConstants.FEED_URL, this.feedUrl).build(); - } - @Override - protected void onInit() throws Exception { + 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"); + } - fetcher.addFetcherEventListener(myFetcherListener); - Assert.notNull(this.feedUrl, "the feedURL can't be null"); - } - - class MyFetcherListener implements FetcherListener { - /** - * @see com.sun.syndication.fetcher.FetcherListener#fetcherEvent(com.sun.syndication.fetcher.FetcherEvent) - */ - public void fetcherEvent(final FetcherEvent event) { - String eventType = event.getEventType(); + 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; + } - 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()); - syndFeeds.add(event.getFeed()); - } else if (FetcherEvent.EVENT_TYPE_FEED_UNCHANGED.equals(eventType)) { - logger.debug("\tEVENT: Feed Unchanged. URL = " + event.getUrlString()); - } - } - } -} \ No newline at end of file + public Message 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()); + } + } + } + +} diff --git a/spring-integration-feed/src/test/java/org/springframework/integration/feed/FeedEntryReaderMessageSourceTests.java b/spring-integration-feed/src/test/java/org/springframework/integration/feed/FeedEntryReaderMessageSourceTests.java index 64077262a7..187636ff8e 100644 --- a/spring-integration-feed/src/test/java/org/springframework/integration/feed/FeedEntryReaderMessageSourceTests.java +++ b/spring-integration-feed/src/test/java/org/springframework/integration/feed/FeedEntryReaderMessageSourceTests.java @@ -62,7 +62,7 @@ public class FeedEntryReaderMessageSourceTests { public void testReceiveFeedWithNoEntries() { FeedReaderMessageSource feedReaderSource = mock(FeedReaderMessageSource.class); SyndFeed feed = mock(SyndFeed.class); - when(feedReaderSource.receiveSyndFeed()).thenReturn(feed); + when(feedReaderSource.receiveFeed()).thenReturn(feed); FeedEntryReaderMessageSource feedEntrySource = new FeedEntryReaderMessageSource(feedReaderSource); feedEntrySource.setBeanName("feedReader"); feedEntrySource.afterPropertiesSet(); @@ -82,7 +82,7 @@ public class FeedEntryReaderMessageSourceTests { entries.add(entry2); entries.add(entry1); when(feed.getEntries()).thenReturn(entries); - when(feedReaderSource.receiveSyndFeed()).thenReturn(feed); + when(feedReaderSource.receiveFeed()).thenReturn(feed); FeedEntryReaderMessageSource feedEntrySource = new FeedEntryReaderMessageSource(feedReaderSource); feedEntrySource.setComponentName("feedReader");