From 75093c93b3cbfa60e74efe26d72d60dde37727a6 Mon Sep 17 00:00:00 2001 From: Josh Long Date: Sun, 3 Oct 2010 19:14:19 -0700 Subject: [PATCH] added a working feed adapter that also maintains state using the metadata persister mechanism. INT-786 --- pom.xml | 894 +++++++++--------- .../context/IntegrationObjectSupport.java | 1 + .../PropertiesBasedMetadataPersister.java | 18 +- spring-integration-feed/pom.xml | 68 ++ .../integration/feed/FeedConstants.java | 25 + .../feed/FeedEntryReaderMessageSource.java | 167 ++++ .../feed/FeedReaderMessageSource.java | 163 ++++ ...FeedMessageSourceBeanDefinitionParser.java | 43 + .../feed/config/FeedNamespaceHandler.java | 40 + .../main/resources/META-INF/spring.handlers | 1 + .../main/resources/META-INF/spring.schemas | 2 + .../main/resources/META-INF/spring.tooling | 4 + .../config/spring-integration-feed-2.0.xsd | 54 ++ .../feed/config/spring-integration-feed.gif | Bin 0 -> 565 bytes .../feed/TestFeedEventDelivery.java | 36 + .../FeedDeliveryEventServiceActivator.java | 25 + .../feed/TestFeedEventDelivery-context.xml | 36 + spring-integration-feed/template.mf | 17 + spring-integration-file/pom.xml | 8 +- spring-integration-parent/pom.xml | 12 +- 20 files changed, 1173 insertions(+), 441 deletions(-) create mode 100644 spring-integration-feed/pom.xml create mode 100644 spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedConstants.java create mode 100644 spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedEntryReaderMessageSource.java create mode 100644 spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedReaderMessageSource.java create mode 100644 spring-integration-feed/src/main/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParser.java create mode 100644 spring-integration-feed/src/main/java/org/springframework/integration/feed/config/FeedNamespaceHandler.java create mode 100644 spring-integration-feed/src/main/resources/META-INF/spring.handlers create mode 100644 spring-integration-feed/src/main/resources/META-INF/spring.schemas create mode 100644 spring-integration-feed/src/main/resources/META-INF/spring.tooling create mode 100644 spring-integration-feed/src/main/resources/org/springframework/integration/feed/config/spring-integration-feed-2.0.xsd create mode 100644 spring-integration-feed/src/main/resources/org/springframework/integration/feed/config/spring-integration-feed.gif create mode 100644 spring-integration-feed/src/test/java/org/springframework/integration/feed/TestFeedEventDelivery.java create mode 100644 spring-integration-feed/src/test/resources/org/springframework/integration/feed/FeedDeliveryEventServiceActivator.java create mode 100644 spring-integration-feed/src/test/resources/org/springframework/integration/feed/TestFeedEventDelivery-context.xml create mode 100644 spring-integration-feed/template.mf diff --git a/pom.xml b/pom.xml index df7758a9d6..513a25a498 100644 --- a/pom.xml +++ b/pom.xml @@ -1,429 +1,469 @@ - - 4.0.0 - org.springframework.integration - spring-integration - Spring Integration - 2.0.0.BUILD-SNAPSHOT - pom - - spring-integration-parent - spring-integration-core - spring-integration-event - spring-integration-file - spring-integration-groovy - spring-integration-http - spring-integration-httpinvoker - spring-integration-ip - spring-integration-jdbc - spring-integration-jms - spring-integration-jmx - spring-integration-mail - spring-integration-rmi - spring-integration-security - spring-integration-stream - spring-integration-test - spring-integration-ws - spring-integration-xml - spring-integration-xmpp - spring-integration-ftp - spring-integration-sftp - spring-integration-twitter - - - UTF-8 - - spring-integration - Spring Integration - INT - ${project.version} - ${dist.id}-${dist.version} - ${dist.finalName}.zip - target/${dist.fileName} - dist.springframework.org - - - - - snapshot - - snapshot - - - - spring-snapshot - Spring Snapshot Repository - s3://maven.springframework.org/snapshot - - - - - milestone - - milestone - - - - spring-milestone - Spring Milestone Repository - s3://maven.springframework.org/milestone - - - - - - org.apache.maven.plugins - maven-antrun-plugin - 1.4 - - - upload-dist - deploy - - - - - - - - - run - - - - - - org.springframework.build - org.springframework.build.aws.ant - 3.0.5.RELEASE - - - net.java.dev.jets3t - jets3t - 0.7.2 - - - - - - - - release - - release - - - - spring-release - file://${java.io.tmpdir}/spring-integration-repo - - - - - - org.apache.maven.plugins - maven-antrun-plugin - 1.4 - - - upload-dist - deploy - - - - - - - - - run - - - - - - org.springframework.build - org.springframework.build.aws.ant - 3.0.5.RELEASE - - - net.java.dev.jets3t - jets3t - 0.7.2 - - - - - - - - staging - - - spring-site-staging - file:///${java.io.tmpdir}/spring-integration/docs/${project.version} - - - spring-milestone-staging - file:///${java.io.tmpdir}/spring-integration/milestone - - - spring-snapshot-staging - file:///${java.io.tmpdir}/spring-integration/snapshot - - - - - - - - http://static.springframework.org/spring-integration/site/downloads/releases.html - - static.springframework.org - scp://cbeams@static.springframework.org/var/www/domains/springframework.org/static/htdocs/spring-integration/sites/${project.version} - - - - - - - org.springframework.build.aws - org.springframework.build.aws.maven - 3.0.0.RELEASE - - - - - com.agilejava.docbkx - docbkx-maven-plugin - 2.0.6 - - - - generate-html - generate-pdf - - package - - - - - org.docbook - docbook-xml - 4.4 - runtime - - - - index.xml - true - ${project.basedir}/src/docbkx/resources/xsl/fopdf.xsl - - css/html.css - - false - ${project.basedir}/src/docbkx/resources/xsl/html.xsl - - - - version - ${pom.version} - - - - - - - - - - - - - - - - - - - - - - - maven-javadoc-plugin - 2.5 - - - aggregate - - aggregate - - package - - true - true -
Spring Integration
- 1.5 - true - ${project.basedir}/src/main/javadoc - ${project.basedir}/src/main/javadoc/overview.html - ${project.basedir}/src/main/javadoc/spring-javadoc.css - - true - - - Spring Integration Core - org.springframework.integration:org.springframework.integration.* - - - Spring Integration Event Support - org.springframework.integration.event:org.springframework.integration.event.* - - - Spring Integration File Support - org.springframework.integration.file:org.springframework.integration.file.* - - - Spring Integration FTP Support - org.springframework.integration.ftp:org.springframework.integration.ftp.* - - - Spring Integration Groovy Support - org.springframework.integration.groovy:org.springframework.integration.groovy.* - - - Spring Integration HTTP Support - org.springframework.integration.http:org.springframework.integration.http.* - - - Spring Integration HttpInvoker Support - org.springframework.integration.httpinvoker:org.springframework.integration.httpinvoker.* - - - Spring Integration JDBC Support - org.springframework.integration.jdbc:org.springframework.integration.jdbc.* - - - Spring Integration JMS Support - org.springframework.integration.jms:org.springframework.integration.jms.* - - - Spring Integration JMX Support - org.springframework.integration.control:org.springframework.integration.jmx:org.springframework.integration.jmx.* - - - Spring Integration Mail Support - org.springframework.integration.mail:org.springframework.integration.mail.* - - - Spring Integration RMI Support - org.springframework.integration.rmi:org.springframework.integration.rmi.* - - - Spring Integration Web Services Support - org.springframework.integration.ws:org.springframework.integration.ws.* - - - Spring Integration Security Support - org.springframework.integration.security:org.springframework.integration.security.* - - - Spring Integration SFTP Support - org.springframework.integration.sftp:org.springframework.integration.sftp.* - - - Spring Integration Stream Support - org.springframework.integration.stream:org.springframework.integration.stream.* - - - Spring Integration TCP/UDP Support - org.springframework.integration.ip:org.springframework.integration.ip.* - - - Spring Integration Test Support - org.springframework.integration.test:org.springframework.integration.test.* - - - Spring Integration Twitter Support - org.springframework.integration.twitter:org.springframework.integration.twitter.* - - - Spring Integration XML Support - org.springframework.integration.xml:org.springframework.integration.xml.* - - - Spring Integration XMPP Support - org.springframework.integration.xmpp:org.springframework.integration.xmpp.* - - - - http://static.springframework.org/spring/docs/3.0.3.RELEASE/javadoc-api - http://static.springsource.org/spring-integration/sites/${project.version}/apidocs - http://java.sun.com/javase/6/docs/api - - - - - - - - maven-assembly-plugin - 2.2-beta-5 - false - - - distribution - - single - - package - - - ${project.basedir}/src/assembly/distribution.xml - - false - - - - - - - org.apache.maven.plugins - maven-release-plugin - 2.0 - - - - - - repository.springframework.maven.release - Spring Framework Maven Release Repository - http://maven.springframework.org/release - - - repository.springframework.maven.milestone - Spring Framework Maven Milestone Repository - http://maven.springframework.org/milestone - - - - repository.source.maven.release - SpringSource Maven Release Repository - http://repository.springsource.com/maven/bundles/release - - + + 4.0.0 + org.springframework.integration + spring-integration + Spring Integration + 2.0.0.BUILD-SNAPSHOT + pom + + spring-integration-parent + spring-integration-core + spring-integration-event + spring-integration-file + spring-integration-feed + spring-integration-groovy + spring-integration-http + spring-integration-httpinvoker + spring-integration-ip + spring-integration-jdbc + spring-integration-jms + spring-integration-jmx + spring-integration-mail + spring-integration-rmi + spring-integration-security + spring-integration-stream + spring-integration-test + spring-integration-ws + spring-integration-xml + spring-integration-xmpp + spring-integration-ftp + spring-integration-sftp + spring-integration-twitter + + + UTF-8 + + spring-integration + Spring Integration + INT + ${project.version} + ${dist.id}-${dist.version} + ${dist.finalName}.zip + target/${dist.fileName} + dist.springframework.org + + + + + snapshot + + snapshot + + + + spring-snapshot + Spring Snapshot Repository + s3://maven.springframework.org/snapshot + + + + + milestone + + milestone + + + + spring-milestone + Spring Milestone Repository + s3://maven.springframework.org/milestone + + + + + + org.apache.maven.plugins + maven-antrun-plugin + 1.4 + + + upload-dist + deploy + + + + + + + + + run + + + + + + org.springframework.build + org.springframework.build.aws.ant + 3.0.5.RELEASE + + + net.java.dev.jets3t + jets3t + 0.7.2 + + + + + + + + release + + release + + + + spring-release + file://${java.io.tmpdir}/spring-integration-repo + + + + + + org.apache.maven.plugins + maven-antrun-plugin + 1.4 + + + upload-dist + deploy + + + + + + + + + run + + + + + + org.springframework.build + org.springframework.build.aws.ant + 3.0.5.RELEASE + + + net.java.dev.jets3t + jets3t + 0.7.2 + + + + + + + + staging + + + spring-site-staging + file:///${java.io.tmpdir}/spring-integration/docs/${project.version} + + + spring-milestone-staging + file:///${java.io.tmpdir}/spring-integration/milestone + + + spring-snapshot-staging + file:///${java.io.tmpdir}/spring-integration/snapshot + + + + + + + + http://static.springframework.org/spring-integration/site/downloads/releases.html + + static.springframework.org + + scp://cbeams@static.springframework.org/var/www/domains/springframework.org/static/htdocs/spring-integration/sites/${project.version} + + + + + + + + org.springframework.build.aws + org.springframework.build.aws.maven + 3.0.0.RELEASE + + + + + com.agilejava.docbkx + docbkx-maven-plugin + 2.0.6 + + + + generate-html + generate-pdf + + package + + + + + org.docbook + docbook-xml + 4.4 + runtime + + + + index.xml + true + ${project.basedir}/src/docbkx/resources/xsl/fopdf.xsl + + css/html.css + + false + ${project.basedir}/src/docbkx/resources/xsl/html.xsl + + + + version + ${pom.version} + + + + + + + + + + + + + + + + + + + + + + + maven-javadoc-plugin + 2.5 + + + aggregate + + aggregate + + package + + true + true +
Spring Integration
+ 1.5 + true + ${project.basedir}/src/main/javadoc + ${project.basedir}/src/main/javadoc/overview.html + ${project.basedir}/src/main/javadoc/spring-javadoc.css + + true + + + Spring Integration Core + org.springframework.integration:org.springframework.integration.* + + + + Spring Integration Event Support + + org.springframework.integration.event:org.springframework.integration.event.* + + + + Spring Integration File Support + + org.springframework.integration.file:org.springframework.integration.file.* + + + + Spring Integration FTP Support + org.springframework.integration.ftp:org.springframework.integration.ftp.* + + + + Spring Integration Groovy Support + + org.springframework.integration.groovy:org.springframework.integration.groovy.* + + + + Spring Integration HTTP Support + + org.springframework.integration.http:org.springframework.integration.http.* + + + + Spring Integration HttpInvoker Support + + org.springframework.integration.httpinvoker:org.springframework.integration.httpinvoker.* + + + + Spring Integration JDBC Support + + org.springframework.integration.jdbc:org.springframework.integration.jdbc.* + + + + Spring Integration JMS Support + org.springframework.integration.jms:org.springframework.integration.jms.* + + + + Spring Integration JMX Support + + org.springframework.integration.control:org.springframework.integration.jmx:org.springframework.integration.jmx.* + + + + Spring Integration Mail Support + + org.springframework.integration.mail:org.springframework.integration.mail.* + + + + Spring Integration RMI Support + org.springframework.integration.rmi:org.springframework.integration.rmi.* + + + + Spring Integration Web Services Support + org.springframework.integration.ws:org.springframework.integration.ws.* + + + + Spring Integration Security Support + + org.springframework.integration.security:org.springframework.integration.security.* + + + + Spring Integration SFTP Support + + org.springframework.integration.sftp:org.springframework.integration.sftp.* + + + + Spring Integration Stream Support + + org.springframework.integration.stream:org.springframework.integration.stream.* + + + + Spring Integration TCP/UDP Support + org.springframework.integration.ip:org.springframework.integration.ip.* + + + + Spring Integration Test Support + + org.springframework.integration.test:org.springframework.integration.test.* + + + + Spring Integration Twitter Support + + org.springframework.integration.twitter:org.springframework.integration.twitter.* + + + + Spring Integration XML Support + org.springframework.integration.xml:org.springframework.integration.xml.* + + + + Spring Integration XMPP Support + + org.springframework.integration.xmpp:org.springframework.integration.xmpp.* + + + + + http://static.springframework.org/spring/docs/3.0.3.RELEASE/javadoc-api + http://static.springsource.org/spring-integration/sites/${project.version}/apidocs + + http://java.sun.com/javase/6/docs/api + + + + + + + + maven-assembly-plugin + 2.2-beta-5 + false + + + distribution + + single + + package + + + ${project.basedir}/src/assembly/distribution.xml + + false + + + + + + + org.apache.maven.plugins + maven-release-plugin + 2.0 + + + + + + repository.springframework.maven.release + Spring Framework Maven Release Repository + http://maven.springframework.org/release + + + repository.springframework.maven.milestone + Spring Framework Maven Milestone Repository + http://maven.springframework.org/milestone + + + + repository.source.maven.release + SpringSource Maven Release Repository + http://repository.springsource.com/maven/bundles/release + + diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java index 519832c9c0..a3d27f9bd7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java @@ -125,6 +125,7 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo } if (this.metadataPersister == null) { PropertiesBasedMetadataPersister mp = new PropertiesBasedMetadataPersister(); + try { mp.afterPropertiesSet(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/PropertiesBasedMetadataPersister.java b/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/PropertiesBasedMetadataPersister.java index 9b09b67b31..7809bd4df2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/PropertiesBasedMetadataPersister.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/PropertiesBasedMetadataPersister.java @@ -16,12 +16,6 @@ import java.util.concurrent.Executor; * Implementation of {@link org.springframework.integration.context.metadata.MetadataPersister} that knows how to write metadata * to a {@link java.util.Properties} instance. * - * - * - * TODO could this perhaps participate or at least be aware of our transaction synchronization mechanism? IE: no guarantees, but we at least try to write on commit()s? - * - * - * * @author Josh Long */ public class PropertiesBasedMetadataPersister implements MetadataPersister, InitializingBean { @@ -87,10 +81,6 @@ public class PropertiesBasedMetadataPersister implements MetadataPersister + + 4.0.0 + + org.springframework.integration + spring-integration-parent + 2.0.0.BUILD-SNAPSHOT + ../spring-integration-parent/pom.xml + + spring-integration-feed + Spring Integration Feed Support + jar + + + org.springframework + spring-context + + + org.springframework.integration + spring-integration-core + + + + commons-langcommons-lang2.5 + + + net.java.dev.rome + rome-fetcher + 1.0.0 + + + + net.java.dev.rome + rome + 1.0.0 + + + + cglib + cglib-nodep + + + junit + junit + + + org.easymock + easymock + + + org.easymock + easymockclassextension + + + org.mockito + mockito-all + + + org.springframework + spring-test + + + org.springframework.integration + spring-integration-test + + + + diff --git a/spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedConstants.java b/spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedConstants.java new file mode 100644 index 0000000000..a00f9fb663 --- /dev/null +++ b/spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedConstants.java @@ -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"; +} \ No newline at end of file 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 new file mode 100644 index 0000000000..8c18aebcae --- /dev/null +++ b/spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedEntryReaderMessageSource.java @@ -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. + * NB: this does not somehow detect entry removal from a feed. + * + * @author Josh Long + * @author Mario Gray + */ +public class FeedEntryReaderMessageSource extends IntegrationObjectSupport implements MessageSource, Lifecycle { + + private volatile ConcurrentLinkedQueue 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 entries; + private volatile long lastTime = -1; + + public FeedEntryReaderMessageSource() { + // this.entries = new ConcurrentSkipListSet(new MyComparator()); + this.entries = new ConcurrentLinkedQueue(); + } + + 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 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 syndEntryComparator = new Comparator() { + 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 feedEntries = (List) 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 { + 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; + } + } +} \ No newline at end of file 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 new file mode 100644 index 0000000000..d5f7517153 --- /dev/null +++ b/spring-integration-feed/src/main/java/org/springframework/integration/feed/FeedReaderMessageSource.java @@ -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 { + private volatile boolean running; + private volatile String feedUrl; + private volatile URL feedURLObject; + private volatile FeedFetcherCache fetcherCache; + private volatile HttpURLFeedFetcher fetcher; + private volatile ConcurrentLinkedQueue syndFeeds; + private volatile MyFetcherListener myFetcherListener; + + public FeedReaderMessageSource() { + syndFeeds = new ConcurrentLinkedQueue(); + } + + 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 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()); + } + } + } +} \ No newline at end of file diff --git a/spring-integration-feed/src/main/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParser.java b/spring-integration-feed/src/main/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParser.java new file mode 100644 index 0000000000..32bcd24e6e --- /dev/null +++ b/spring-integration-feed/src/main/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParser.java @@ -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()); + } +} \ No newline at end of file diff --git a/spring-integration-feed/src/main/java/org/springframework/integration/feed/config/FeedNamespaceHandler.java b/spring-integration-feed/src/main/java/org/springframework/integration/feed/config/FeedNamespaceHandler.java new file mode 100644 index 0000000000..ace8004450 --- /dev/null +++ b/spring-integration-feed/src/main/java/org/springframework/integration/feed/config/FeedNamespaceHandler.java @@ -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* + * inbound-channel-adapter. 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()); + } + + +} \ No newline at end of file diff --git a/spring-integration-feed/src/main/resources/META-INF/spring.handlers b/spring-integration-feed/src/main/resources/META-INF/spring.handlers new file mode 100644 index 0000000000..f67ce32e75 --- /dev/null +++ b/spring-integration-feed/src/main/resources/META-INF/spring.handlers @@ -0,0 +1 @@ +http\://www.springframework.org/schema/integration/feed=org.springframework.integration.feed.config.FeedNamespaceHandler \ No newline at end of file diff --git a/spring-integration-feed/src/main/resources/META-INF/spring.schemas b/spring-integration-feed/src/main/resources/META-INF/spring.schemas new file mode 100644 index 0000000000..311d93a42e --- /dev/null +++ b/spring-integration-feed/src/main/resources/META-INF/spring.schemas @@ -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 \ No newline at end of file diff --git a/spring-integration-feed/src/main/resources/META-INF/spring.tooling b/spring-integration-feed/src/main/resources/META-INF/spring.tooling new file mode 100644 index 0000000000..c2a2ba35ad --- /dev/null +++ b/spring-integration-feed/src/main/resources/META-INF/spring.tooling @@ -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 diff --git a/spring-integration-feed/src/main/resources/org/springframework/integration/feed/config/spring-integration-feed-2.0.xsd b/spring-integration-feed/src/main/resources/org/springframework/integration/feed/config/spring-integration-feed-2.0.xsd new file mode 100644 index 0000000000..c3f725ec50 --- /dev/null +++ b/spring-integration-feed/src/main/resources/org/springframework/integration/feed/config/spring-integration-feed-2.0.xsd @@ -0,0 +1,54 @@ + + + + + + + + + SyndFeed or SyndEntry objects. + + ]]> + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/spring-integration-feed/src/main/resources/org/springframework/integration/feed/config/spring-integration-feed.gif b/spring-integration-feed/src/main/resources/org/springframework/integration/feed/config/spring-integration-feed.gif new file mode 100644 index 0000000000000000000000000000000000000000..6464056d479897a40330346eeb62c423dd15a645 GIT binary patch literal 565 zcmZ?wbhEHb6krfwc*Xz%xt^+pzUqa(nw6p2)g^w7v4*V)My(Y-ZL{`uOkLNLX5Lfd z*)#u8Z>B|GlEs8<%Snm0ljd!m+~6^J*|8}N?$bhoW)}L)T620%i_6>R!uFpugais6&oRnL8Ywj&dz5n9-{TDwTY%YFqqW{6UMUP(nc(OS0<+TMb z-~W7lZ~vQ%Q{Eo0e*5X?$1lIW{rvs&*B`Klh7zIplZBCi!IeP=DuoaFyd@m+fp4G_f()WbRb75tr9CP-W~8u(oGYS-Px^TiM7##bup~o=U5Rn2wf` z^G;`DJ*8F_K{W*@2@xlYJqoQ%{4!4GFF47xaY;H#ALZqAl)NHp`{1Fiu&cTh*9##Y vs~aM>tax1QEDa6ybvdkl{kGzG`kYHeLnC2v+rcI?E*Y%@8xjvYFjxZs$j_?M literal 0 HcmV?d00001 diff --git a/spring-integration-feed/src/test/java/org/springframework/integration/feed/TestFeedEventDelivery.java b/spring-integration-feed/src/test/java/org/springframework/integration/feed/TestFeedEventDelivery.java new file mode 100644 index 0000000000..d5c0e88cb6 --- /dev/null +++ b/spring-integration-feed/src/test/java/org/springframework/integration/feed/TestFeedEventDelivery.java @@ -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 entryMessage = feedEntryReaderMessageSource.receive(); + + if (entryMessage != null) { + SyndEntry entry = entryMessage.getPayload(); + System.out.println((entry.getTitle() + "=" + entry.getUri())); + } + + Thread.sleep(1000); + } + } + */ +} diff --git a/spring-integration-feed/src/test/resources/org/springframework/integration/feed/FeedDeliveryEventServiceActivator.java b/spring-integration-feed/src/test/resources/org/springframework/integration/feed/FeedDeliveryEventServiceActivator.java new file mode 100644 index 0000000000..14a2d21ade --- /dev/null +++ b/spring-integration-feed/src/test/resources/org/springframework/integration/feed/FeedDeliveryEventServiceActivator.java @@ -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 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)); + + } + +} diff --git a/spring-integration-feed/src/test/resources/org/springframework/integration/feed/TestFeedEventDelivery-context.xml b/spring-integration-feed/src/test/resources/org/springframework/integration/feed/TestFeedEventDelivery-context.xml new file mode 100644 index 0000000000..13bde9ecb6 --- /dev/null +++ b/spring-integration-feed/src/test/resources/org/springframework/integration/feed/TestFeedEventDelivery-context.xml @@ -0,0 +1,36 @@ + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-feed/template.mf b/spring-integration-feed/template.mf new file mode 100644 index 0000000000..36e6277820 --- /dev/null +++ b/spring-integration-feed/template.mf @@ -0,0 +1,17 @@ +Bundle-SymbolicName: org.springframework.integration.ftp +Bundle-Name: Spring Integration SFTP Support +Bundle-Vendor: SpringSource +Bundle-ManifestVersion: 2 +Import-Template: + org.apache.commons.logging;version="[1.1.1, 2.0.0)", + org.apache.commons.lang.*;version="[2.5.0, 3.0.0)", + org.apache.commons.net.*;version="[2.0.0, 3.0.0)", + org.springframework.integration.*;version="[2.0.0, 2.0.1)", + org.springframework.scheduling.*;version="[3.0.3, 4.0.0)", + org.springframework.beans.*;version="[3.0.3, 4.0.0)", + org.springframework.context;version="[3.0.3, 4.0.0)", + org.springframework.core.*;version="[3.0.3, 4.0.0)", + org.springframework.util;version="[3.0.3, 4.0.0)", + org.jivesoftware.*;version="[3.1.0, 4.0.0)", + javax.*;version="0", + org.w3c.dom.*;version="0" diff --git a/spring-integration-file/pom.xml b/spring-integration-file/pom.xml index 04e1816828..2c7dd22e59 100644 --- a/spring-integration-file/pom.xml +++ b/spring-integration-file/pom.xml @@ -7,7 +7,6 @@ 2.0.0.BUILD-SNAPSHOT ../spring-integration-parent/pom.xml - org.springframework.integration spring-integration-file Spring Integration File Support jar @@ -20,6 +19,10 @@ org.springframework.integration spring-integration-core + + + + cglib @@ -43,11 +46,12 @@ org.springframework - spring-test + spring-test org.springframework.integration spring-integration-test + diff --git a/spring-integration-parent/pom.xml b/spring-integration-parent/pom.xml index 96ddbeb988..762ac20fe2 100644 --- a/spring-integration-parent/pom.xml +++ b/spring-integration-parent/pom.xml @@ -136,7 +136,17 @@ spring-aspects ${org.springframework.version} - + + org.springframework spring-core ${org.springframework.version}