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

INT-786 initial cleanup, removed MetaPersister in favor of injecting a Map which could be anything (e.g., cache etc.)

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-10-16 08:03:00 -04:00
parent 75d9e701e1
commit 3c5218ec67
9 changed files with 260 additions and 318 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java
View File

@@ -18,15 +18,12 @@ package org.springframework.integration.context;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.convert.ConversionService;
import org.springframework.integration.context.metadata.MetadataPersister;
import org.springframework.integration.context.metadata.PropertiesBasedMetadataPersister;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -50,8 +47,6 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
* Logger that is available to subclasses
*/
protected final Log logger = LogFactory.getLog(getClass());
private volatile MetadataPersister<?> metadataPersister;
private volatile String beanName;
@@ -119,27 +114,6 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
return this.beanFactory;
}
protected MetadataPersister getRequiredMetadataPersister() {
if (this.metadataPersister == null && this.beanFactory != null) {
this.metadataPersister = IntegrationContextUtils.getMetadataPersister(this.beanFactory);
}
if (this.metadataPersister == null) {
PropertiesBasedMetadataPersister mp = new PropertiesBasedMetadataPersister();
try {
mp.afterPropertiesSet();
}
catch (Exception e) {
if (e instanceof RuntimeException) {
throw (RuntimeException) e;
}
throw new BeanInitializationException("failed to obtain reference to MetadataPersister strategy implementation.", e);
}
this.metadataPersister = mp;
}
return this.metadataPersister;
}
protected TaskScheduler getTaskScheduler() {
if (this.taskScheduler == null && this.beanFactory != null) {
this.taskScheduler = IntegrationContextUtils.getTaskScheduler(this.beanFactory);

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

@@ -1,92 +1,129 @@
/*
* 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 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.Map;
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;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;
import com.sun.syndication.feed.synd.SyndEntry;
import com.sun.syndication.feed.synd.SyndFeed;
/**
* 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.
*
* This implementation of {@link MessageSource} will produce individual {@link SyndEntry}s for a feed identified
* with 'feedUrl' attribute.
*
* @author Josh Long
* @author Mario Gray
* @author Oleg Zhurakousky
*/
public class FeedEntryReaderMessageSource extends IntegrationObjectSupport implements MessageSource<SyndEntry>, Lifecycle {
private volatile ConcurrentLinkedQueue<SyndEntry> entries;
private volatile MetadataPersister persister;
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 String feedMetadataIdKey;
private String feedUrl;
private volatile String feedMetadataIdKey;
private volatile String feedUrl;
private volatile boolean running;
public boolean isRunning() {
return running;
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);
if (x < -1) {
return -1;
}
else if (x > 1) {
return 1;
}
return 0;
}
};
public void setFeedUrl(String feedUrl) {
this.feedUrl = feedUrl;
}
public String getFeedUrl() {
return feedUrl;
}
/**
* Allows you to provide your own implementation of 'persisterMap' instead of relying on
* your own which is in-memory.
*
* @param persisterMap
*/
public void setPersisterMap(Map<String, String> persisterMap) {
Assert.notNull(persisterMap, "'persisterMap' can not be null");
this.persisterMap = persisterMap;
}
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 boolean isRunning() {
return running;
}
public void stop() {
this.feedReaderMessageSource.stop();
this.setRunning(false);
}
public String getComponentType(){
return "feed:inbound-channel-adapter";
}
@SuppressWarnings("unchecked")
public SyndEntry receiveSyndEntry() {
synchronized (this.monitor) {
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) {
long sort = this.sortId(se);
if (sort > this.lastTime)
entries.add(se);
}
}
}
return pollAndCache();
}
}
public Message<SyndEntry> receive() {
SyndEntry se = receiveSyndEntry();
@@ -96,72 +133,41 @@ public class FeedEntryReaderMessageSource extends IntegrationObjectSupport imple
return MessageBuilder.withPayload(se).build();
}
int longToCompare(long l) {
if (l < -1) return -1;
if (l > 1) return 1;
return 0;
public void start() {
this.feedReaderMessageSource.start();
this.setRunning(true);
}
private long sortId(SyndEntry entry) {
return entry.getPublishedDate().getTime();
}
private Comparator<SyndEntry> syndEntryComparator = new Comparator<SyndEntry>() {
public int compare(SyndEntry syndEntry, SyndEntry syndEntry1) {
long x = sortId(syndEntry) - sortId(syndEntry1);
return longToCompare(x);
}
};
@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();
@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();
// setup persistence of metadata
this.feedMetadataIdKey = FeedEntryReaderMessageSource.class.getName() + "#" + feedUrl;
String lastTime = (String) this.persisterMap.get(this.feedMetadataIdKey);
if (lastTime != null && !lastTime.trim().equalsIgnoreCase("")) {
this.lastTime = Long.parseLong(lastTime);
}
}
private SyndEntry pollAndCache() {
SyndEntry next = this.entries.poll();
if (null == next) return null;
if (next == null) {
return null;
}
this.lastTime = sortId(next);
this.persister.write(this.feedMetadataIdKey, this.lastTime + "");
this.persisterMap.put(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;
}
}
}

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

@@ -1,86 +1,72 @@
/*
* 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.context.Lifecycle;
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.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.
* This implementation of {@link MessageSource} will produce {@link SyndFeed} for a feed identified
* with 'feedUrl' attribute.
*
* @author Josh Long
* @author Mario Gray
* @author Oleg Zhurakousky
*/
public class FeedReaderMessageSource extends IntegrationObjectSupport
class FeedReaderMessageSource extends IntegrationObjectSupport
implements InitializingBean, Lifecycle, MessageSource<SyndFeed> {
private volatile boolean running;
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;
private final Object syndFeedMonitor = new Object();
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;
}*/
public void setFeedUrl(final String feedUrl) {
this.feedUrl = feedUrl;
}
private volatile long lastTime = -1;
public String getFeedUrl() {
return feedUrl;
}
public void start() {
this.running = true;
}
@@ -88,32 +74,26 @@ public class FeedReaderMessageSource extends IntegrationObjectSupport
public void stop() {
this.running = false;
}
private String feedMetadataIdKey;
private final Object syndFeedMonitor = new Object();
public boolean isRunning() {
return this.running;
}
public SyndFeed receiveSyndFeed() {
SyndFeed returnedSyndFeed = null;
try {
synchronized (syndFeedMonitor) {
fetcher.retrieveFeed(this.feedURLObject);
returnedSyndFeed = 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) {
if (returnedSyndFeed == null) {
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);
} catch (Exception e) {
throw new MessagingException("Exception thrown when trying to retrive feed at url '" + this.feedURLObject + "'", e);
}
return returnedSyndFeed;
@@ -129,19 +109,19 @@ public class FeedReaderMessageSource extends IntegrationObjectSupport
return MessageBuilder.withPayload(syndFeed).setHeader(FeedConstants.FEED_URL, this.feedURLObject).build();
}
public boolean isRunning() {
return this.running;
@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);
}
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)
@@ -153,8 +133,7 @@ public class FeedReaderMessageSource extends IntegrationObjectSupport
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());
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
View File

@@ -1,43 +1,40 @@
/*
* 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 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
* @author Oleg Zhurakousky
*/
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());
BeanDefinitionBuilder feedBuilder =
BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.feed.FeedEntryReaderMessageSource");
feedBuilder.addPropertyValue("feedUrl", element.getAttribute("feedUrl"));
return BeanDefinitionReaderUtils.registerWithGeneratedName(feedBuilder.getBeanDefinition(), parserContext.getRegistry());
}
}

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

@@ -1,33 +1,26 @@
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.
*/
* 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 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.
*
* NamespaceHandler for FEED module
*
* @author Josh Long
*/
public class FeedNamespaceHandler extends NamespaceHandlerSupport {
@@ -35,6 +28,4 @@ public class FeedNamespaceHandler extends NamespaceHandlerSupport {
public void init() {
registerBeanDefinitionParser("inbound-channel-adapter", new FeedMessageSourceBeanDefinitionParser());
}
}

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

@@ -36,17 +36,7 @@
</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:attribute name="feedUrl" type="xsd:string" use="required"/>
</xsd:complexType>
</xsd:element>

15
spring-integration-feed/src/test/resources/org/springframework/integration/feed/FeedDeliveryEventServiceActivator.java → spring-integration-feed/src/test/java/org/springframework/integration/feed/FeedDeliveryEventServiceActivator.java
View File

@@ -1,18 +1,25 @@
package org.springframework.integration.feed;
import com.sun.syndication.feed.synd.SyndEntry;
import org.apache.commons.lang.builder.ToStringBuilder;
import java.util.Properties;
import org.springframework.integration.Message;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.history.MessageHistory;
import org.springframework.stereotype.Component;
import com.sun.syndication.feed.synd.SyndEntry;
@Component
public class FeedDeliveryEventServiceActivator {
@ServiceActivator
public void activate(Message<SyndEntry> evtMsg) throws Exception {
public void activate(Message<SyndEntry> message) throws Exception {
SyndEntry syndEntry = evtMsg.getPayload();
MessageHistory history = MessageHistory.read(message);
for (Properties properties : history) {
System.out.println(properties);
}
SyndEntry syndEntry = message.getPayload();
System.out.println( "Publishing new SyndEntry " + syndEntry.getUri() +":"+
syndEntry.getPublishedDate().toString()+ ":"+ syndEntry.getPublishedDate().getTime());

13
spring-integration-feed/src/test/resources/org/springframework/integration/feed/TestFeedEventDelivery-context.xml → spring-integration-feed/src/test/java/org/springframework/integration/feed/TestFeedEventDelivery-context.xml
View File

@@ -5,10 +5,11 @@
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 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.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<int:message-history/>
<bean id="activator" class="org.springframework.integration.feed.FeedDeliveryEventServiceActivator"/>
@@ -18,14 +19,14 @@
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>
<!-- <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 id="feedAdapter" channel="feedChanges" feedUrl="http://feeds.bbci.co.uk/news/rss.xml" >
<int:poller fixed-rate="10000" max-messages-per-poll="-1"/>
</feed:inbound-channel-adapter>
<int:channel id="feedChanges"/>

37
spring-integration-feed/src/test/java/org/springframework/integration/feed/TestFeedEventDelivery.java
View File

@@ -1,5 +1,21 @@
/*
* 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 org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
@@ -10,27 +26,8 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
public class TestFeedEventDelivery {
@Test
@Ignore
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);
}
}
*/
}