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

added a working feed adapter that also maintains state using the metadata persister mechanism. INT-786

Browse Source
This commit is contained in:
Josh Long
2010-10-03 19:14:19 -07:00
parent d3a083ecdb
commit 75093c93b3
20 changed files with 1173 additions and 441 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedConstants.java Normal file
View File

@@ -0,0 +1,25 @@
/*
* 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;
/**
* Provides a place to store the header keys for {@link FeedReaderMessageSource}
*
* @author Josh Long
*/
public class FeedConstants {
static public final String FEED_URL = "FEED_URL";
}

167
spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedEntryReaderMessageSource.java Normal file
View File

@@ -0,0 +1,167 @@
package org.springframework.integration.feed;
import com.sun.syndication.feed.synd.SyndEntry;
import com.sun.syndication.feed.synd.SyndFeed;
import org.springframework.context.Lifecycle;
import org.springframework.integration.Message;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.context.metadata.MetadataPersister;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
/**
* this is a slightly different use case than {@link org.springframework.integration.feed.FeedReaderMessageSource}.
* This returns which entries are added, which is a more nuanced use case requiring some of our own caching.
* <em>NB:</em> this does <strong>not</strong> somehow detect entry removal from a feed.
*
* @author Josh Long
* @author Mario Gray
*/
public class FeedEntryReaderMessageSource extends IntegrationObjectSupport implements MessageSource<SyndEntry>, Lifecycle {
private volatile ConcurrentLinkedQueue<SyndEntry> entries;
private volatile MetadataPersister persister;
private volatile FeedReaderMessageSource feedReaderMessageSource;
private final Object monitor = new Object();
private String feedMetadataIdKey;
private String feedUrl;
private volatile boolean running;
public boolean isRunning() {
return running;
}
public void setRunning(boolean running) {
this.running = running;
}
// private Queue<SyndEntry> entries;
private volatile long lastTime = -1;
public FeedEntryReaderMessageSource() {
// this.entries = new ConcurrentSkipListSet<SyndEntry>(new MyComparator());
this.entries = new ConcurrentLinkedQueue<SyndEntry>();
}
public void start() {
this.feedReaderMessageSource.start();
this.setRunning(true);
}
private long sortId(SyndEntry entry) {
return entry.getPublishedDate().getTime();
}
@Override
protected void onInit() throws Exception {
this.persister = this.getRequiredMetadataPersister();
Assert.notNull(this.feedUrl, "the feedUrl can't be null");
this.feedReaderMessageSource = new FeedReaderMessageSource();
this.feedReaderMessageSource.setFeedUrl(this.feedUrl);
this.feedReaderMessageSource.setBeanFactory(this.getBeanFactory());
this.feedReaderMessageSource.setBeanName(this.getComponentName());
this.feedReaderMessageSource.afterPropertiesSet();
// setup persistence of metadata
this.feedMetadataIdKey = FeedEntryReaderMessageSource.class.getName() + "#" + feedUrl;
String lastTime = (String) this.persister.read(this.feedMetadataIdKey);
if (lastTime != null && !lastTime.trim().equalsIgnoreCase("")) {
this.lastTime = Long.parseLong(lastTime);
}
}
public void stop() {
this.feedReaderMessageSource.stop();
this.setRunning(false);
}
public Message<SyndEntry> receive() {
SyndEntry se = receiveSyndEntry();
if (se == null) {
return null;
}
return MessageBuilder.withPayload(se).build();
}
int longToCompare(long l) {
if (l < -1) return -1;
if (l > 1) return 1;
return 0;
}
private Comparator<SyndEntry> syndEntryComparator = new Comparator<SyndEntry>() {
public int compare(SyndEntry syndEntry, SyndEntry syndEntry1) {
long x = sortId(syndEntry) - sortId(syndEntry1);
return longToCompare(x);
}
};
@SuppressWarnings("unchecked")
public SyndEntry receiveSyndEntry() {
synchronized (this.monitor) { // priority goes to the backlog
SyndEntry nextUp = pollAndCache();
if (nextUp != null) {
return nextUp;
}
// otherwise, fill the backlog up
SyndFeed syndFeed = this.feedReaderMessageSource.receiveSyndFeed();
if (syndFeed != null) {
List<SyndEntry> feedEntries = (List<SyndEntry>) syndFeed.getEntries();
if (null != feedEntries) {
Collections.sort(feedEntries, syndEntryComparator);
for (SyndEntry se : feedEntries) {
System.out.println("se: " + se.getPublishedDate().getTime());
long sort = this.sortId(se);
if (sort > this.lastTime)
entries.add(se);
}
}
}
return pollAndCache();
}
}
private SyndEntry pollAndCache() {
SyndEntry next = this.entries.poll();
if (null == next) return null;
this.lastTime = sortId(next);
this.persister.write(this.feedMetadataIdKey, this.lastTime + "");
return next;
}
public String getFeedUrl() {
return feedUrl;
}
public void setFeedUrl(final String feedUrl) {
this.feedUrl = feedUrl;
}
class MyComparator implements Comparator<SyndEntry> {
public int compare(final SyndEntry syndEntry, final SyndEntry syndEntry1) {
long val = sortId(syndEntry) - sortId(syndEntry1);
if (val > 0) return 1;
if (val < 0) return -1;
return 0;
}
}
}

163
spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedReaderMessageSource.java Normal file
View File

@@ -0,0 +1,163 @@
package org.springframework.integration.feed;
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.FeedFetcherCache;
import com.sun.syndication.fetcher.impl.HashMapFeedInfoCache;
import com.sun.syndication.fetcher.impl.HttpURLFeedFetcher;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.integration.Message;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.context.metadata.MetadataPersister;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;
import java.net.URL;
import java.util.concurrent.ConcurrentLinkedQueue;
/**
* The idea behind this class is that {@link org.springframework.integration.core.MessageSource#receive()} will only
* return a {@link SyndFeed} when the event listener tells us that a feed has been updated. If we can ascertain that
* it's been updated, then we can add the item to the {@link java.util.Queue} implementation.
*
* @author Josh Long
* @author Mario Gray
*/
public class FeedReaderMessageSource extends IntegrationObjectSupport
implements InitializingBean, Lifecycle, MessageSource<SyndFeed> {
private volatile boolean running;
private volatile String feedUrl;
private volatile URL feedURLObject;
private volatile FeedFetcherCache fetcherCache;
private volatile HttpURLFeedFetcher fetcher;
private volatile ConcurrentLinkedQueue<SyndFeed> syndFeeds;
private volatile MyFetcherListener myFetcherListener;
public FeedReaderMessageSource() {
syndFeeds = new ConcurrentLinkedQueue<SyndFeed>();
}
private volatile MetadataPersister persister;
@Override
protected void onInit() throws Exception {
this.persister = this.getRequiredMetadataPersister();
myFetcherListener = new MyFetcherListener();
fetcherCache = HashMapFeedInfoCache.getInstance();
fetcher = new HttpURLFeedFetcher(fetcherCache);
// fetcher.set
fetcher.addFetcherEventListener(myFetcherListener);
Assert.notNull(this.feedUrl, "the feedURL can't be null");
feedURLObject = new URL(this.feedUrl);
/*
String id = FeedReaderMessageSource.class.getName() + "#" + feedUrl;
StringBuffer stringBuffer = new StringBuffer();
for (char c : id.toCharArray())
if (Character.isDigit(c) || Character.isLetter(c))
stringBuffer.append(c);
id = stringBuffer.toString();
this.feedMetadataIdKey = id;
long lastTimeNo = -1;
String lastTime = (String) this.persister.read(this.feedMetadataIdKey);
if (lastTime != null && !lastTime.trim().equalsIgnoreCase("")) {
lastTimeNo = Long.parseLong(lastTime);
this.lastTime = lastTimeNo;
}*/
}
private volatile long lastTime = -1;
public void start() {
this.running = true;
}
public void stop() {
this.running = false;
}
private String feedMetadataIdKey;
private final Object syndFeedMonitor = new Object();
public SyndFeed receiveSyndFeed() {
SyndFeed returnedSyndFeed = null;
try {
synchronized (syndFeedMonitor) {
fetcher.retrieveFeed(this.feedURLObject);
logger.debug("attempted to retrieve feed '" + this.feedUrl + "'");
returnedSyndFeed = syndFeeds.poll(); // there wont be things whose pub date is < than the lastTime
if (null == returnedSyndFeed) {
logger.debug("no feeds updated, return null!");
return null;
}
// so its OK to update the lastTime
//
/* this.lastTime = sortId(returnedSyndFeed);if (null != this.persister)
this.persister.write(this.feedMetadataIdKey, this.lastTime + "");
*/
}
} catch (Throwable e) {
logger.debug("Exception thrown when trying to retrive feed at url '" + this.feedURLObject + "'", e);
}
return returnedSyndFeed;
}
public Message<SyndFeed> receive() {
SyndFeed syndFeed = this.receiveSyndFeed();
if (null == syndFeed) {
return null;
}
return MessageBuilder.withPayload(syndFeed).setHeader(FeedConstants.FEED_URL, this.feedURLObject).build();
}
public boolean isRunning() {
return this.running;
}
public String getFeedUrl() {
return feedUrl;
}
public void setFeedUrl(final String feedUrl) {
this.feedUrl = feedUrl;
}
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();
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());
// if (sortId(event.getFeed()) > lastTime) // its true if the lastTime is -1 || N
syndFeeds.add(event.getFeed());
} else if (FetcherEvent.EVENT_TYPE_FEED_UNCHANGED.equals(eventType)) {
logger.debug("\tEVENT: Feed Unchanged. URL = " + event.getUrlString());
}
}
}
}

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

@@ -0,0 +1,43 @@
package org.springframework.integration.feed.config;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.feed.FeedEntryReaderMessageSource;
import org.springframework.integration.feed.FeedReaderMessageSource;
import org.w3c.dom.Element;
/**
* Handles parsing the configuration for the feed inbound channel adapter.
*
* @author Josh Long
*/
public class FeedMessageSourceBeanDefinitionParser extends AbstractPollingInboundChannelAdapterParser {
private String packageName = FeedReaderMessageSource.class.getPackage().getName();
@Override
protected String parseSource(final Element element, final ParserContext parserContext) {
String pftoe = (element.getAttribute("prefer-updated-feed-to-entries"));
pftoe = pftoe == null ? "false" : pftoe.trim().toLowerCase();
boolean preferFeed = pftoe.equalsIgnoreCase(Boolean.TRUE.toString().toLowerCase());
String className = this.packageName + "." + (preferFeed ?
FeedReaderMessageSource.class.getSimpleName() :
FeedEntryReaderMessageSource.class.getSimpleName()
);
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(className);
builder.addPropertyValue("feedUrl", element.getAttribute("feed"));
if (!preferFeed) {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "backlog-cache-size", "maximumBacklogCacheSize");
}
return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
}
}

40
spring-integration-feed/src/main/java/org/springframework/integration/feed/config/FeedNamespaceHandler.java Normal file
View File

@@ -0,0 +1,40 @@
package org.springframework.integration.feed.config;
/*
* 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.
*/
import org.springframework.beans.factory.xml.NamespaceHandlerSupport;
/**
* This is a rather tricky one. I've decided it's best to not get cute about it and to expose *one*
* <em>inbound-channel-adapter</em>. The adapter will let the user pick which type of updated object they'd like to
* return. By default it'll return new {@link com.sun.syndication.feed.synd.SyndEntry} objects (which represent
* individual, new entries in a given feed). One adapter will return updated {@link
* com.sun.syndication.feed.synd.SyndFeed} objects, or it can return updated {@link
* com.sun.syndication.feed.synd.SyndEntry} objects.
*
* @author Josh Long
*/
public class FeedNamespaceHandler extends NamespaceHandlerSupport {
public void init() {
registerBeanDefinitionParser("inbound-channel-adapter", new FeedMessageSourceBeanDefinitionParser());
}
}

1
spring-integration-feed/src/main/resources/META-INF/spring.handlers Normal file
View File

@@ -0,0 +1 @@
http\://www.springframework.org/schema/integration/feed=org.springframework.integration.feed.config.FeedNamespaceHandler

2
spring-integration-feed/src/main/resources/META-INF/spring.schemas Normal file
View File

@@ -0,0 +1,2 @@
http\://www.springframework.org/schema/integration/feed/spring-integration-feed-2.0.xsd=org/springframework/integration/feed/config/spring-integration-feed-2.0.xsd
http\://www.springframework.org/schema/integration/feed/spring-integration-feed.xsd=org/springframework/integration/feed/config/spring-integration-feed-2.0.xsd

4
spring-integration-feed/src/main/resources/META-INF/spring.tooling Normal file
View File

@@ -0,0 +1,4 @@
# Tooling related information for the integration feed namespace
http\://www.springframework.org/schema/integration/feed@name=integration feed Namespace
http\://www.springframework.org/schema/integration/feed@prefix=int-feed
http\://www.springframework.org/schema/integration/feed@icon=org/springframework/integration/feed/config/spring-integration-feed.gif

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

@@ -0,0 +1,54 @@
<?xml version="1.0" encoding="UTF-8"?>
<xsd:schema xmlns="http://www.springframework.org/schema/integration/feed"
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:integration="http://www.springframework.org/schema/integration"
targetNamespace="http://www.springframework.org/schema/integration/feed"
elementFormDefault="qualified"
attributeFormDefault="unqualified">
<xsd:import namespace="http://www.springframework.org/schema/beans"/>
<xsd:import namespace="http://www.springframework.org/schema/integration"
schemaLocation="http://www.springframework.org/schema/integration/spring-integration-2.0.xsd"/>
<xsd:element name="inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation><![CDATA[
This adapter takes a feed URL (either RSS or ATOM) and responds to updates.
Depending on whether you've set the prefer-updated-feeds-to-entries attribute
or not, you will recieve <code>SyndFeed</code> or <code>SyndEntry</code> objects.
]]></xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="channel" use="required" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="backlog-cache-size" type="xsd:int"/>
<xsd:attribute name="feed" type="xsd:string" use="required"/>
<!--
<xsd:attribute name="prefer-updated-feed-to-entries" type="xsd:boolean"/>
-->
</xsd:complexType>
</xsd:element>
</xsd:schema>

BIN
spring-integration-feed/src/main/resources/org/springframework/integration/feed/config/spring-integration-feed.gif Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 565 B

36
spring-integration-feed/src/test/java/org/springframework/integration/feed/TestFeedEventDelivery.java Normal file
View File

@@ -0,0 +1,36 @@
package org.springframework.integration.feed;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class TestFeedEventDelivery {
@Test
public void testDeliveryOfFeed() throws Exception {
Thread.sleep(1000 * 60);
}
/* public static void main(String[] args) throws Throwable {
String siweb = "http://twitter.com/statuses/public_timeline.atom"; //http://localhost:8080/siweb/foo.atom";
FeedEntryReaderMessageSource feedEntryReaderMessageSource = new FeedEntryReaderMessageSource();
feedEntryReaderMessageSource.setFeedUrl(siweb);
feedEntryReaderMessageSource.afterPropertiesSet();
feedEntryReaderMessageSource.start();
while (true) {
Message<SyndEntry> entryMessage = feedEntryReaderMessageSource.receive();
if (entryMessage != null) {
SyndEntry entry = entryMessage.getPayload();
System.out.println((entry.getTitle() + "=" + entry.getUri()));
}
Thread.sleep(1000);
}
}
*/
}

25
spring-integration-feed/src/test/resources/org/springframework/integration/feed/FeedDeliveryEventServiceActivator.java Normal file
View File

@@ -0,0 +1,25 @@
package org.springframework.integration.feed;
import com.sun.syndication.feed.synd.SyndEntry;
import org.apache.commons.lang.builder.ToStringBuilder;
import org.springframework.integration.Message;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.stereotype.Component;
@Component
public class FeedDeliveryEventServiceActivator {
@ServiceActivator
public void activate(Message<SyndEntry> evtMsg) throws Exception {
SyndEntry syndEntry = evtMsg.getPayload();
System.out.println( "Publishing new SyndEntry " + syndEntry.getUri() +":"+
syndEntry.getPublishedDate().toString()+ ":"+ syndEntry.getPublishedDate().getTime());
// System.out.println( syndEntry.toString());
// System.out.println("Delivery! " + ToStringBuilder.reflectionToString(evtMsg));
}
}

36
spring-integration-feed/src/test/resources/org/springframework/integration/feed/TestFeedEventDelivery-context.xml Normal file
View File

@@ -0,0 +1,36 @@
<?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:feed="http://www.springframework.org/schema/integration/feed"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/feed http://www.springframework.org/schema/integration/feed/spring-integration-feed.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<bean id="activator" class="org.springframework.integration.feed.FeedDeliveryEventServiceActivator"/>
<!--
this will keep state in /tmp/feedDemo.properties and not deliver anything until the feed has an updated pub date
to see the feed again, rm /tmp/feedDemo.properties
-->
<bean id="metadataPersister" class="org.springframework.integration.context.metadata.PropertiesBasedMetadataPersister">
<property name="uniqueName" value="feedDemo"/>
</bean>
<!--http://twitter.com/statuses/public_timeline.atom
-->
<feed:inbound-channel-adapter channel="feedChanges" feed="http://feeds.bbci.co.uk/news/rss.xml" >
<int:poller fixed-rate="10000"/>
</feed:inbound-channel-adapter>
<int:channel id="feedChanges"/>
<int:service-activator input-channel="feedChanges" ref="activator" />
</beans>