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

polishing

Browse Source
This commit is contained in:
Mark Fisher
2010-10-27 08:36:53 -04:00
parent 7217ba806b
commit b5b0941ea3
3 changed files with 88 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<SyndEntry> retrievedEntries = (List<SyndEntry>) syndFeed.getEntries();
if (!CollectionUtils.isEmpty(retrievedEntries)) {

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

@@ -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<SyndFeed> {
public class FeedReaderMessageSource extends IntegrationObjectSupport implements InitializingBean, MessageSource<SyndFeed> {
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<SyndFeed> syndFeeds = new ConcurrentLinkedQueue<SyndFeed>();
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<SyndFeed> feeds = new ConcurrentLinkedQueue<SyndFeed>();
return returnedSyndFeed;
}
private final Object feedMonitor = new Object();
public Message<SyndFeed> 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());
}
}
}
}
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());
}
}
}
}

4
spring-integration-feed/src/test/java/org/springframework/integration/feed/FeedEntryReaderMessageSourceTests.java
View File

@@ -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");