diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java index 7a672c9297..6da6aec529 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java @@ -19,6 +19,7 @@ package org.springframework.integration.config.xml; import org.w3c.dom.Element; import org.springframework.beans.BeanMetadataElement; +import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; @@ -31,18 +32,25 @@ import org.springframework.util.xml.DomUtils; * @author Mark Fisher * @author Gary Russell * @author Oleg Zhurakousky + * @author Artem Bilan */ public abstract class AbstractPollingInboundChannelAdapterParser extends AbstractChannelAdapterParser { @Override + @SuppressWarnings("unchecked") protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) { BeanMetadataElement source = this.parseSource(element, parserContext); if (source == null) { parserContext.getReaderContext().error("failed to parse source", element); } + + String channelAdapterId = this.resolveId(element, (AbstractBeanDefinition) source, parserContext); + String sourceBeanName = channelAdapterId + ".source"; + parserContext.getRegistry().registerBeanDefinition(sourceBeanName, (BeanDefinition) source); + BeanDefinitionBuilder adapterBuilder = BeanDefinitionBuilder .genericBeanDefinition(SourcePollingChannelAdapterFactoryBean.class); - adapterBuilder.addPropertyValue("source", source); + adapterBuilder.addPropertyReference("source", sourceBeanName); adapterBuilder.addPropertyReference("outputChannel", channelName); IntegrationNamespaceUtils.setValueIfAttributeDefined(adapterBuilder, element, "send-timeout"); Element pollerElement = DomUtils.getChildElementByTagName(element, "poller"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java index 184513990f..a3d6f70a0a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java @@ -20,7 +20,7 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.core.convert.ConversionService; import org.springframework.expression.spel.support.StandardEvaluationContext; import org.springframework.integration.MessageChannel; -import org.springframework.integration.store.metadata.MetadataStore; +import org.springframework.integration.metadata.MetadataStore; import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/MetadataStore.java b/spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStore.java similarity index 69% rename from spring-integration-core/src/main/java/org/springframework/integration/store/metadata/MetadataStore.java rename to spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStore.java index de4a8376aa..06d48c9cab 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/MetadataStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStore.java @@ -14,7 +14,10 @@ * limitations under the License. */ -package org.springframework.integration.store.metadata; +package org.springframework.integration.metadata; + +import org.springframework.jmx.export.annotation.ManagedAttribute; +import org.springframework.jmx.export.annotation.ManagedResource; /** * Strategy interface for storing metadata from certain adapters @@ -25,6 +28,7 @@ package org.springframework.integration.store.metadata; * @author Mark Fisher * @since 2.0 */ +@ManagedResource public interface MetadataStore { /** @@ -35,6 +39,15 @@ public interface MetadataStore { /** * Reads a value for the given key from this MetadataStore. */ + @ManagedAttribute String get(String key); + /** + * Remove a value for the given key from this MetadataStore. + * return the previous value associated with key, or + * null if there was no mapping for key. + */ + @ManagedAttribute + String remove(String key); + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/PropertiesPersistingMetadataStore.java b/spring-integration-core/src/main/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStore.java similarity index 96% rename from spring-integration-core/src/main/java/org/springframework/integration/store/metadata/PropertiesPersistingMetadataStore.java rename to spring-integration-core/src/main/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStore.java index 3393468cfd..26c35231ec 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/PropertiesPersistingMetadataStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStore.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.store.metadata; +package org.springframework.integration.metadata; import java.io.BufferedInputStream; import java.io.BufferedOutputStream; @@ -86,6 +86,12 @@ public class PropertiesPersistingMetadataStore implements MetadataStore, Initial return this.metadata.getProperty(key); } + @Override + @SuppressWarnings("uchecked") + public String remove(String key) { + return (String) this.metadata.remove(key); + } + public void destroy() throws Exception { this.saveMetadata(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/SimpleMetadataStore.java b/spring-integration-core/src/main/java/org/springframework/integration/metadata/SimpleMetadataStore.java similarity index 89% rename from spring-integration-core/src/main/java/org/springframework/integration/store/metadata/SimpleMetadataStore.java rename to spring-integration-core/src/main/java/org/springframework/integration/metadata/SimpleMetadataStore.java index 5a04477715..0bf5b50fdd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/SimpleMetadataStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/metadata/SimpleMetadataStore.java @@ -11,7 +11,7 @@ * specific language governing permissions and limitations under the License. */ -package org.springframework.integration.store.metadata; +package org.springframework.integration.metadata; import java.util.HashMap; import java.util.Map; @@ -37,4 +37,9 @@ public class SimpleMetadataStore implements MetadataStore { return this.metadata.get(key); } + @Override + public String remove(String key) { + return metadata.remove(key); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/package-info.java b/spring-integration-core/src/main/java/org/springframework/integration/metadata/package-info.java similarity index 50% rename from spring-integration-core/src/main/java/org/springframework/integration/store/metadata/package-info.java rename to spring-integration-core/src/main/java/org/springframework/integration/metadata/package-info.java index 18bee7d3e5..d27403d529 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/metadata/package-info.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/metadata/package-info.java @@ -1,4 +1,4 @@ /** * Provides classes supporting metadata stores. */ -package org.springframework.integration.store.metadata; \ No newline at end of file +package org.springframework.integration.metadata; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/metadata/PropertiesPersistingMetadataStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStoreTests.java similarity index 94% rename from spring-integration-core/src/test/java/org/springframework/integration/store/metadata/PropertiesPersistingMetadataStoreTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStoreTests.java index f3ac8a7748..8ed8cc371c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/metadata/PropertiesPersistingMetadataStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStoreTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.store.metadata; +package org.springframework.integration.metadata; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; @@ -27,7 +27,6 @@ import org.junit.Test; import org.springframework.core.io.FileSystemResource; import org.springframework.core.io.support.PropertiesLoaderUtils; -import org.springframework.integration.store.metadata.PropertiesPersistingMetadataStore; /** * @author Oleg Zhurakousky diff --git a/spring-integration-feed/src/main/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParser.java b/spring-integration-feed/src/main/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParser.java index c71a959827..f7f355cd18 100644 --- a/spring-integration-feed/src/main/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParser.java +++ b/spring-integration-feed/src/main/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2013 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. @@ -23,6 +23,7 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder; 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.inbound.FeedEntryMessageSource; import org.springframework.util.StringUtils; /** @@ -31,20 +32,23 @@ import org.springframework.util.StringUtils; * @author Josh Long * @author Oleg Zhurakousky * @author Mark Fisher + * @author Gunnar Hillert + * @author Artem Bilan * @since 2.0 */ public class FeedInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser { @Override protected BeanMetadataElement parseSource(final Element element, final ParserContext parserContext) { - BeanDefinitionBuilder sourceBuilder = BeanDefinitionBuilder.genericBeanDefinition( - "org.springframework.integration.feed.inbound.FeedEntryMessageSource"); + BeanDefinitionBuilder sourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(FeedEntryMessageSource.class); sourceBuilder.addConstructorArgValue(element.getAttribute("url")); + sourceBuilder.addConstructorArgValue(element.getAttribute(ID_ATTRIBUTE)); String feedFetcherRef = element.getAttribute("feed-fetcher"); if (StringUtils.hasText(feedFetcherRef)) { sourceBuilder.addConstructorArgReference(feedFetcherRef); } IntegrationNamespaceUtils.setReferenceIfAttributeDefined(sourceBuilder, element, "metadata-store"); + return sourceBuilder.getBeanDefinition(); } diff --git a/spring-integration-feed/src/main/java/org/springframework/integration/feed/inbound/FeedEntryMessageSource.java b/spring-integration-feed/src/main/java/org/springframework/integration/feed/inbound/FeedEntryMessageSource.java index 38ca53ef82..9e39d23718 100644 --- a/spring-integration-feed/src/main/java/org/springframework/integration/feed/inbound/FeedEntryMessageSource.java +++ b/spring-integration-feed/src/main/java/org/springframework/integration/feed/inbound/FeedEntryMessageSource.java @@ -30,8 +30,8 @@ import org.springframework.integration.MessagingException; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.core.MessageSource; -import org.springframework.integration.store.metadata.MetadataStore; -import org.springframework.integration.store.metadata.SimpleMetadataStore; +import org.springframework.integration.metadata.MetadataStore; +import org.springframework.integration.metadata.SimpleMetadataStore; import org.springframework.integration.support.MessageBuilder; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; @@ -52,6 +52,7 @@ import com.sun.syndication.fetcher.impl.HttpURLFeedFetcher; * @author Josh Long * @author Mario Gray * @author Oleg Zhurakousky + * @author Artem Bilan * @since 2.0 */ public class FeedEntryMessageSource extends IntegrationObjectSupport implements MessageSource { @@ -62,7 +63,7 @@ public class FeedEntryMessageSource extends IntegrationObjectSupport implements private final Queue entries = new ConcurrentLinkedQueue(); - private volatile String metadataKey; + private final String metadataKey; private volatile MetadataStore metadataStore; @@ -82,17 +83,19 @@ public class FeedEntryMessageSource extends IntegrationObjectSupport implements * If the feed URL has a protocol other than http*, consider providing a custom implementation of the * {@link FeedFetcher} via the alternate constructor. */ - public FeedEntryMessageSource(URL feedUrl) { - this(feedUrl, new HttpURLFeedFetcher(HashMapFeedInfoCache.getInstance())); + public FeedEntryMessageSource(URL feedUrl, String metadataKey) { + this(feedUrl, metadataKey, new HttpURLFeedFetcher(HashMapFeedInfoCache.getInstance())); } /** * Creates a FeedEntryMessageSource that will use the provided FeedFetcher to read from the given feed URL. */ - public FeedEntryMessageSource(URL feedUrl, FeedFetcher feedFetcher) { + public FeedEntryMessageSource(URL feedUrl, String metadataKey, FeedFetcher feedFetcher) { Assert.notNull(feedUrl, "feedUrl must not be null"); + Assert.notNull(metadataKey, "metadataKey must not be null"); Assert.notNull(feedFetcher, "feedFetcher must not be null"); this.feedUrl = feedUrl; + this.metadataKey = metadataKey + "." + this.feedUrl; this.feedFetcher = feedFetcher; } @@ -130,18 +133,7 @@ public class FeedEntryMessageSource extends IntegrationObjectSupport implements this.metadataStore = new SimpleMetadataStore(); } } - StringBuilder metadataKeyBuilder = new StringBuilder(); - if (StringUtils.hasText(this.getComponentType())) { - metadataKeyBuilder.append(this.getComponentType() + "."); - } - if (StringUtils.hasText(this.getComponentName())) { - metadataKeyBuilder.append(this.getComponentName() + "."); - } - else if (logger.isWarnEnabled()) { - logger.warn("FeedEntryMessageSource has no name. MetadataStore key might not be unique."); - } - metadataKeyBuilder.append(this.feedUrl); - this.metadataKey = metadataKeyBuilder.toString(); + String lastTimeValue = this.metadataStore.get(this.metadataKey); if (StringUtils.hasText(lastTimeValue)) { this.lastTime = Long.parseLong(lastTimeValue); @@ -237,6 +229,7 @@ public class FeedEntryMessageSource extends IntegrationObjectSupport implements } return (date2 == null) ? 1 : 0; } + } @@ -258,6 +251,7 @@ public class FeedEntryMessageSource extends IntegrationObjectSupport implements } } } + } } diff --git a/spring-integration-feed/src/main/resources/org/springframework/integration/feed/config/spring-integration-feed-3.0.xsd b/spring-integration-feed/src/main/resources/org/springframework/integration/feed/config/spring-integration-feed-3.0.xsd index c3182eaf84..954fe1c681 100644 --- a/spring-integration-feed/src/main/resources/org/springframework/integration/feed/config/spring-integration-feed-3.0.xsd +++ b/spring-integration-feed/src/main/resources/org/springframework/integration/feed/config/spring-integration-feed-3.0.xsd @@ -22,7 +22,28 @@ - + + + + The bean id of this Polling Endpoint; the MessageSource is also registered with this id + plus a suffix '.source'; also used as the + MetaDataStore key with suffix '.' + feedUrl - The URL for an RSS or ATOM feed. + + + + + + + + + + + + Identifies the channel attached to this adapter, to which messages will be sent. + + + + @@ -54,7 +75,7 @@ - + diff --git a/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests-file-usage-context.xml b/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests-file-usage-context.xml index 216a0ee5a9..05b8997914 100644 --- a/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests-file-usage-context.xml +++ b/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests-file-usage-context.xml @@ -21,6 +21,6 @@ - + diff --git a/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests-file-usage-noid-context.xml b/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests-file-usage-noid-context.xml deleted file mode 100644 index 544aa96b6f..0000000000 --- a/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests-file-usage-noid-context.xml +++ /dev/null @@ -1,21 +0,0 @@ - - - - - - - - - - - - - - \ No newline at end of file diff --git a/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests.java b/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests.java index 9d1d973610..0e6ee90862 100644 --- a/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests.java +++ b/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests.java @@ -33,6 +33,7 @@ import org.junit.Before; import org.junit.Ignore; import org.junit.Test; import org.mockito.Mockito; + import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.Message; @@ -43,7 +44,7 @@ import org.springframework.integration.core.MessageHandler; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; import org.springframework.integration.feed.inbound.FeedEntryMessageSource; import org.springframework.integration.history.MessageHistory; -import org.springframework.integration.store.metadata.MetadataStore; +import org.springframework.integration.metadata.MetadataStore; import org.springframework.integration.test.util.TestUtils; import com.sun.syndication.feed.synd.SyndEntry; @@ -83,6 +84,7 @@ public class FeedInboundChannelAdapterParserTests { context.destroy(); } + public void validateSuccessfulHttpConfigurationWithCustomMetadataStore() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "FeedInboundChannelAdapterParserTests-http-context.xml", this.getClass()); @@ -98,7 +100,7 @@ public class FeedInboundChannelAdapterParserTests { @Test public void validateSuccessfulNewsRetrievalWithFileUrlAndMessageHistory() throws Exception { - File persisterFile = new File(System.getProperty("java.io.tmpdir") + "/spring-integration/", "message-store.properties"); + File persisterFile = new File(System.getProperty("java.io.tmpdir") + "/spring-integration/", "metadata-store.properties"); if (persisterFile.exists()) { persisterFile.delete(); } @@ -120,26 +122,6 @@ public class FeedInboundChannelAdapterParserTests { context.destroy(); } - @Test - public void validateSuccessfulNewsRetrievalWithFileUrlNoPersistentIdentifier() throws Exception{ - //Test file samples.rss has 3 news items - latch = spy(new CountDownLatch(3)); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "FeedInboundChannelAdapterParserTests-file-usage-noid-context.xml", this.getClass()); - latch.await(5, TimeUnit.SECONDS); - verify(latch, times(3)).countDown(); - context.destroy(); - - // since we are not deleting the persister file - // in this iteration no new feeds will be received and the latch will timeout - latch = spy(new CountDownLatch(3)); - context = new ClassPathXmlApplicationContext( - "FeedInboundChannelAdapterParserTests-file-usage-noid-context.xml", this.getClass()); - latch.await(5, TimeUnit.SECONDS); - verify(latch, times(3)).countDown(); - context.destroy(); - } - @Test @Ignore // goes against the real feed public void validateSuccessfulNewsRetrievalWithHttpUrl() throws Exception{ @@ -204,6 +186,11 @@ public class FeedInboundChannelAdapterParserTests { public String get(String key) { return null; } + + @Override + public String remove(String key) { + return null; + } } } diff --git a/spring-integration-feed/src/test/java/org/springframework/integration/feed/inbound/FeedEntryMessageSourceTests.java b/spring-integration-feed/src/test/java/org/springframework/integration/feed/inbound/FeedEntryMessageSourceTests.java index 1244cd3e48..d7aad3f25c 100644 --- a/spring-integration-feed/src/test/java/org/springframework/integration/feed/inbound/FeedEntryMessageSourceTests.java +++ b/spring-integration-feed/src/test/java/org/springframework/integration/feed/inbound/FeedEntryMessageSourceTests.java @@ -26,7 +26,7 @@ import java.net.URL; import org.junit.Before; import org.junit.Test; import org.springframework.integration.Message; -import org.springframework.integration.store.metadata.PropertiesPersistingMetadataStore; +import org.springframework.integration.metadata.PropertiesPersistingMetadataStore; import com.sun.syndication.feed.synd.SyndEntry; import com.sun.syndication.fetcher.FeedFetcher; @@ -52,14 +52,14 @@ public class FeedEntryMessageSourceTests { @Test(expected=IllegalArgumentException.class) public void testFailureWhenNotInitialized() throws Exception { URL url = new URL("file:src/test/java/org/springframework/integration/feed/sample.rss"); - FeedEntryMessageSource feedEntrySource = new FeedEntryMessageSource(url); + FeedEntryMessageSource feedEntrySource = new FeedEntryMessageSource(url, "foo"); feedEntrySource.receive(); } @Test public void testReceiveFeedWithNoEntries() throws Exception { URL url = new URL("file:src/test/java/org/springframework/integration/feed/empty.rss"); - FeedEntryMessageSource feedEntrySource = new FeedEntryMessageSource(url, this.feedFetcher); + FeedEntryMessageSource feedEntrySource = new FeedEntryMessageSource(url, "foo", this.feedFetcher); feedEntrySource.setBeanName("feedReader"); feedEntrySource.afterPropertiesSet(); assertNull(feedEntrySource.receive()); @@ -68,7 +68,7 @@ public class FeedEntryMessageSourceTests { @Test public void testReceiveFeedWithEntriesSorted() throws Exception { URL url = new URL("file:src/test/java/org/springframework/integration/feed/sample.rss"); - FeedEntryMessageSource source = new FeedEntryMessageSource(url, this.feedFetcher); + FeedEntryMessageSource source = new FeedEntryMessageSource(url, "foo", this.feedFetcher); source.setComponentName("feedReader"); source.afterPropertiesSet(); Message message1 = source.receive(); @@ -87,7 +87,7 @@ public class FeedEntryMessageSourceTests { @Test public void testReceiveFeedWithRealEntriesAndRepeatWithPersistentMetadataStore() throws Exception { URL url = new URL("file:src/test/java/org/springframework/integration/feed/sample.rss"); - FeedEntryMessageSource feedEntrySource = new FeedEntryMessageSource(url, this.feedFetcher); + FeedEntryMessageSource feedEntrySource = new FeedEntryMessageSource(url, "foo", this.feedFetcher); feedEntrySource.setBeanName("feedReader"); PropertiesPersistingMetadataStore metadataStore = new PropertiesPersistingMetadataStore(); metadataStore.afterPropertiesSet(); @@ -111,7 +111,7 @@ public class FeedEntryMessageSourceTests { metadataStore.afterPropertiesSet(); // now test that what's been read is no longer retrieved - feedEntrySource = new FeedEntryMessageSource(url, this.feedFetcher); + feedEntrySource = new FeedEntryMessageSource(url, "foo", this.feedFetcher); feedEntrySource.setBeanName("feedReader"); metadataStore = new PropertiesPersistingMetadataStore(); metadataStore.afterPropertiesSet(); @@ -127,7 +127,7 @@ public class FeedEntryMessageSourceTests { @Test public void testReceiveFeedWithRealEntriesAndRepeatNoPersistentMetadataStore() throws Exception { URL url = new URL("file:src/test/java/org/springframework/integration/feed/sample.rss"); - FeedEntryMessageSource feedEntrySource = new FeedEntryMessageSource(url, this.feedFetcher); + FeedEntryMessageSource feedEntrySource = new FeedEntryMessageSource(url, "foo", this.feedFetcher); feedEntrySource.setBeanName("feedReader"); feedEntrySource.afterPropertiesSet(); SyndEntry entry1 = feedEntrySource.receive().getPayload(); @@ -146,7 +146,7 @@ public class FeedEntryMessageSourceTests { // UNLIKE the previous test // now test that what's been read is read AGAIN - feedEntrySource = new FeedEntryMessageSource(url, this.feedFetcher); + feedEntrySource = new FeedEntryMessageSource(url, "foo", this.feedFetcher); feedEntrySource.setBeanName("feedReader"); feedEntrySource.afterPropertiesSet(); entry1 = feedEntrySource.receive().getPayload(); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java index d6261d48de..404a58dfa8 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileInboundChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -16,9 +16,10 @@ package org.springframework.integration.file.config; +import org.w3c.dom.Element; + import org.springframework.beans.BeanMetadataElement; import org.springframework.beans.factory.config.BeanDefinition; -import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.xml.ParserContext; @@ -27,7 +28,6 @@ import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.integration.file.locking.NioFileLocker; import org.springframework.util.StringUtils; import org.springframework.util.xml.DomUtils; -import org.w3c.dom.Element; /** * Parser for the <inbound-channel-adapter> element of the 'file' namespace. @@ -53,9 +53,8 @@ public class FileInboundChannelAdapterParser extends AbstractPollingInboundChann builder.addPropertyReference("locker", lockerBeanName); } builder.addPropertyReference("filter", filterBeanName); - String beanName = BeanDefinitionReaderUtils.registerWithGeneratedName( - builder.getBeanDefinition(), parserContext.getRegistry()); - return new RuntimeBeanReference(beanName); + + return builder.getBeanDefinition(); } private String registerLocker(Element element, ParserContext parserContext) { diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpParserInboundTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpParserInboundTests.java index 4ff9043190..68d5193db1 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpParserInboundTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpParserInboundTests.java @@ -60,9 +60,8 @@ public class FtpParserInboundTests { fail("BeansException expected."); } catch (BeansException e) { + assertThat(e, Matchers.instanceOf(BeanCreationException.class)); Throwable cause = e.getCause(); - assertThat(cause, Matchers.instanceOf(BeanCreationException.class)); - cause = cause.getCause(); assertThat(cause, Matchers.instanceOf(MessagingException.class)); cause = cause.getCause(); assertThat(cause, Matchers.instanceOf(FileNotFoundException.class)); diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java index 9c7157b896..a9decccdd7 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2013 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. @@ -19,18 +19,15 @@ package org.springframework.integration.jms.config; import org.w3c.dom.Element; import org.springframework.beans.BeanMetadataElement; -import org.springframework.beans.factory.config.BeanDefinition; -import org.springframework.beans.factory.parsing.BeanComponentDefinition; 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.util.StringUtils; /** - * Parser for the <inbound-channel-adapter/> element of the 'jms' namespace. - * + * Parser for the <inbound-channel-adapter/> element of the 'jms' namespace. + * * @author Mark Fisher */ public class JmsInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser { @@ -87,9 +84,7 @@ public class JmsInboundChannelAdapterParser extends AbstractPollingInboundChanne builder.addPropertyReference(JmsAdapterParserUtils.HEADER_MAPPER_PROPERTY, headerMapper); } IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "selector", "messageSelector"); - BeanDefinition beanDefinition = builder.getBeanDefinition(); - String beanName = BeanDefinitionReaderUtils.generateBeanName(beanDefinition, parserContext.getRegistry()); - return new BeanComponentDefinition(beanDefinition, beanName); + return builder.getBeanDefinition(); } } diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests-context.xml index d19f5e6bf4..0ab4f1ab17 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests-context.xml +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests-context.xml @@ -13,18 +13,18 @@ http://www.springframework.org/schema/integration/jmx/spring-integration-jmx.xsd"> - + - - + + - + - + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests.java index 25faf51a4d..535f8b12ec 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests.java @@ -1,11 +1,11 @@ /* * 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. @@ -36,7 +36,7 @@ public class PollingAdapterMBeanTests { @Autowired private MBeanServer server; - + @Test public void testMessageSourceMBeanExists() throws Exception { // System.err.println(server.queryNames(new ObjectName("*:type=MessageSource,*"), null)); diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParser.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParser.java index 216f642ec6..a236ce1b26 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParser.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterParser.java @@ -15,13 +15,11 @@ */ package org.springframework.integration.mongodb.config; -import org.springframework.beans.factory.config.BeanDefinition; import org.w3c.dom.Element; import org.springframework.beans.BeanMetadataElement; -import org.springframework.beans.factory.config.RuntimeBeanReference; +import org.springframework.beans.factory.config.BeanDefinition; 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; @@ -52,8 +50,7 @@ public class MongoDbInboundChannelAdapterParser extends AbstractPollingInboundCh IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "entity-class"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expect-single-result"); - String beanName = BeanDefinitionReaderUtils.registerWithGeneratedName( - builder.getBeanDefinition(), parserContext.getRegistry()); - return new RuntimeBeanReference(beanName); + return builder.getBeanDefinition(); } + } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisStoreInboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisStoreInboundChannelAdapterParser.java index 1920f4ad97..0e73cd3cd7 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisStoreInboundChannelAdapterParser.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisStoreInboundChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2013 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. @@ -20,9 +20,7 @@ import org.w3c.dom.Element; import org.springframework.beans.BeanMetadataElement; import org.springframework.beans.factory.config.BeanDefinition; -import org.springframework.beans.factory.config.RuntimeBeanReference; 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; @@ -62,9 +60,8 @@ public class RedisStoreInboundChannelAdapterParser extends AbstractPollingInboun parserContext, element, atLeastOneRequired); builder.addConstructorArgValue(expressionDef); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "collection-type"); - String beanName = BeanDefinitionReaderUtils.registerWithGeneratedName( - builder.getBeanDefinition(), parserContext.getRegistry()); - return new RuntimeBeanReference(beanName); + + return builder.getBeanDefinition(); } } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/metadata/RedisMetadataStore.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/metadata/RedisMetadataStore.java similarity index 88% rename from spring-integration-redis/src/main/java/org/springframework/integration/redis/store/metadata/RedisMetadataStore.java rename to spring-integration-redis/src/main/java/org/springframework/integration/redis/metadata/RedisMetadataStore.java index b837625b41..e8aa0609c7 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/metadata/RedisMetadataStore.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/metadata/RedisMetadataStore.java @@ -11,13 +11,13 @@ * specific language governing permissions and limitations under the License. */ -package org.springframework.integration.redis.store.metadata; +package org.springframework.integration.redis.metadata; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.BoundValueOperations; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.StringRedisTemplate; -import org.springframework.integration.store.metadata.MetadataStore; +import org.springframework.integration.metadata.MetadataStore; import org.springframework.util.Assert; /** @@ -65,4 +65,13 @@ public class RedisMetadataStore implements MetadataStore { BoundValueOperations ops = this.redisTemplate.boundValueOps(key); return ops.get(); } + + @Override + public String remove(String key) { + Assert.notNull(key, "'key' must not be null."); + String value = this.get(key); + this.redisTemplate.delete(key); + return value; + } + } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/metadata/package-info.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/metadata/package-info.java new file mode 100644 index 0000000000..65840e9612 --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/metadata/package-info.java @@ -0,0 +1,5 @@ +/** + * Provides support for Redis-based + * {@link org.springframework.integration.metadata.MetadataStore}s. + */ +package org.springframework.integration.redis.metadata; diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/metadata/package-info.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/metadata/package-info.java deleted file mode 100644 index 8e0a51cc48..0000000000 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/metadata/package-info.java +++ /dev/null @@ -1,5 +0,0 @@ -/** - * Provides support for Redis-based - * {@link org.springframework.integration.store.metadata.MetadataStore}s. - */ -package org.springframework.integration.redis.store.metadata; diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/metadata/RedisMetadataStoreTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/metadata/RedisMetadataStoreTests.java similarity index 89% rename from spring-integration-redis/src/test/java/org/springframework/integration/redis/store/metadata/RedisMetadataStoreTests.java rename to spring-integration-redis/src/test/java/org/springframework/integration/redis/metadata/RedisMetadataStoreTests.java index 2e57b9c1b2..c9a48610c8 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/store/metadata/RedisMetadataStoreTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/metadata/RedisMetadataStoreTests.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.integration.redis.store.metadata; +package org.springframework.integration.redis.metadata; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; @@ -29,6 +29,7 @@ import org.springframework.integration.redis.rules.RedisAvailableTests; /** * @author Gunnar Hillert + * @author Artem Bilan * @since 3.0 * */ @@ -143,4 +144,20 @@ public class RedisMetadataStoreTests extends RedisAvailableTests { fail("Expected an IllegalArgumentException to be thrown."); } + + @Test + @RedisAvailable + public void testRemoveFromMetadataStore(){ + RedisConnectionFactory jcf = this.getConnectionFactoryForTest(); + RedisMetadataStore metadataStore = new RedisMetadataStore(jcf); + + String testKey = "RedisMetadataStoreTests-Remove"; + String testValue = "Integration"; + + metadataStore.put(testKey, testValue); + + assertEquals(testValue, metadataStore.remove(testKey)); + assertNull(metadataStore.remove(testKey)); + } + } diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/TwitterInboundChannelAdapterParser.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/TwitterInboundChannelAdapterParser.java index 675c52fe9f..9538f12796 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/TwitterInboundChannelAdapterParser.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/config/TwitterInboundChannelAdapterParser.java @@ -32,7 +32,7 @@ import org.springframework.util.StringUtils; /** * Parser for inbound Twitter Channel Adapters. - * + * * @author Oleg Zhurakousky * @since 2.0 */ @@ -50,7 +50,10 @@ public class TwitterInboundChannelAdapterParser extends AbstractPollingInboundCh BeanDefinitionBuilder templateBuilder = BeanDefinitionBuilder.genericBeanDefinition(TwitterTemplate.class); builder.addConstructorArgValue(templateBuilder.getBeanDefinition()); } + builder.addConstructorArgValue(element.getAttribute(ID_ATTRIBUTE)); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "query"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "metadata-store"); return builder.getBeanDefinition(); } diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractTwitterMessageSource.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractTwitterMessageSource.java index 536c5ab910..3240276365 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractTwitterMessageSource.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractTwitterMessageSource.java @@ -28,9 +28,11 @@ import org.springframework.integration.MessagingException; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.core.MessageSource; -import org.springframework.integration.store.metadata.MetadataStore; -import org.springframework.integration.store.metadata.SimpleMetadataStore; +import org.springframework.integration.metadata.MetadataStore; +import org.springframework.integration.metadata.SimpleMetadataStore; import org.springframework.integration.support.MessageBuilder; +import org.springframework.jmx.export.annotation.ManagedAttribute; +import org.springframework.jmx.export.annotation.ManagedOperation; import org.springframework.social.twitter.api.DirectMessage; import org.springframework.social.twitter.api.Tweet; import org.springframework.social.twitter.api.Twitter; @@ -44,24 +46,29 @@ import org.springframework.util.StringUtils; * messages when using the Twitter API. This class also handles keeping track of * the latest inbound message it has received and avoiding, where possible, * redelivery of duplicate messages. This functionality is enabled using the - * {@link org.springframework.integration.store.MetadataStore} strategy. + * {@link org.springframework.integration.metadata.MetadataStore} strategy. * * @author Josh Long * @author Oleg Zhurakousky * @author Mark Fisher * @author Gunnar Hillert + * @author Artem Bilan * * @since 2.0 */ @SuppressWarnings("rawtypes") abstract class AbstractTwitterMessageSource extends IntegrationObjectSupport implements MessageSource { - private volatile long lastPollForTweet; + private final Twitter twitter; + + private final TweetComparator tweetComparator = new TweetComparator(); + + private final Object lastEnqueuedIdMonitor = new Object(); + + private final String metadataKey; private volatile MetadataStore metadataStore; - private volatile String metadataKey; - private final Queue tweets = new LinkedBlockingQueue(); private volatile int prefetchThreshold = 0; @@ -70,25 +77,36 @@ abstract class AbstractTwitterMessageSource extends IntegrationObjectSupport private volatile long lastProcessedId = -1; - private final Twitter twitter; - private final TweetComparator tweetComparator = new TweetComparator(); - - private final Object lastEnqueuedIdMonitor = new Object(); - - - public AbstractTwitterMessageSource(Twitter twitter) { + public AbstractTwitterMessageSource(Twitter twitter, String metadataKey) { Assert.notNull(twitter, "twitter must not be null"); + Assert.notNull(metadataKey, "metadataKey must not be null"); this.twitter = twitter; + if (this.twitter.isAuthorized()){ + UserOperations userOperations = this.twitter.userOperations(); + String profileId = String.valueOf(userOperations.getProfileId()); + if (profileId != null) { + metadataKey += "." + profileId; + } + } + this.metadataKey = metadataKey; } + public void setMetadataStore(MetadataStore metadataStore) { + this.metadataStore = metadataStore; + } + + public void setPrefetchThreshold(int prefetchThreshold) { + this.prefetchThreshold = prefetchThreshold; + } + protected Twitter getTwitter() { return this.twitter; } @Override - protected void onInit() throws Exception{ + protected void onInit() throws Exception { super.onInit(); if (this.metadataStore == null) { // first try to look for a 'metadataStore' in the context @@ -100,26 +118,7 @@ abstract class AbstractTwitterMessageSource extends IntegrationObjectSupport this.metadataStore = new SimpleMetadataStore(); } } - StringBuilder metadataKeyBuilder = new StringBuilder(); - if (StringUtils.hasText(this.getComponentType())) { - metadataKeyBuilder.append(this.getComponentType()); - } - if (StringUtils.hasText(this.getComponentName())) { - metadataKeyBuilder.append("." + this.getComponentName()); - } - else if (logger.isWarnEnabled()) { - logger.warn(this.getClass().getSimpleName() + " has no name. MetadataStore key might not be unique."); - } - if (this.twitter.isAuthorized()){ - UserOperations userOperations = this.twitter.userOperations(); - String profileId = String.valueOf(userOperations.getProfileId()); - if (profileId != null) { - metadataKeyBuilder.append("." + profileId); - } - } - - this.metadataKey = metadataKeyBuilder.toString(); String lastId = this.metadataStore.get(this.metadataKey); // initialize the last status ID from the metadataStore if (StringUtils.hasText(lastId)) { @@ -132,16 +131,10 @@ abstract class AbstractTwitterMessageSource extends IntegrationObjectSupport public Message receive() { T tweet = this.tweets.poll(); if (tweet == null) { - long currentTime = System.currentTimeMillis(); - long elapsedTime = currentTime - this.lastPollForTweet; - if (elapsedTime < 15000) { - // need to wait longer - return null; - } this.refreshTweetQueueIfNecessary(); tweet = this.tweets.poll(); - this.lastPollForTweet = currentTime; } + if (tweet != null) { this.lastProcessedId = this.getIdForTweet(tweet); this.metadataStore.put(this.metadataKey, String.valueOf(this.lastProcessedId)); @@ -204,6 +197,27 @@ abstract class AbstractTwitterMessageSource extends IntegrationObjectSupport } + /** + * Remove the metadata key and the corresponding value from the Metadata Store. + */ + @ManagedOperation(description="Remove the metadata key and the corresponding value from the Metadata Store.") + void resetMetadataStore() { + synchronized(this) { + this.metadataStore.remove(this.metadataKey); + this.lastProcessedId = -1L; + this.lastEnqueuedId = -1L; + } + } + + /** + * + * @return {@code -1} if lastProcessedId is not set, yet. + */ + @ManagedAttribute + public long getLastProcessedId() { + return this.lastProcessedId; + } + private class TweetComparator implements Comparator { public int compare(T tweet1, T tweet2) { @@ -230,6 +244,7 @@ abstract class AbstractTwitterMessageSource extends IntegrationObjectSupport throw new IllegalArgumentException("Uncomparable Twitter objects: " + tweet1 + " and " + tweet2); } } + } } diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSource.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSource.java index deed9b2be7..758302aa6e 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSource.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2013 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. @@ -31,14 +31,13 @@ import org.springframework.social.twitter.api.Twitter; */ public class DirectMessageReceivingMessageSource extends AbstractTwitterMessageSource { - public DirectMessageReceivingMessageSource(Twitter twitter) { - super(twitter); + public DirectMessageReceivingMessageSource(Twitter twitter, String metadataKey) { + super(twitter, metadataKey); } - @Override public String getComponentType() { - return "twitter:dm-inbound-channel-adapter"; + return "twitter:dm-inbound-channel-adapter"; } @Override diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/MentionsReceivingMessageSource.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/MentionsReceivingMessageSource.java index ad9175e84a..5551f3de1a 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/MentionsReceivingMessageSource.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/MentionsReceivingMessageSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2013 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. @@ -30,11 +30,10 @@ import org.springframework.social.twitter.api.Twitter; */ public class MentionsReceivingMessageSource extends AbstractTwitterMessageSource { - public MentionsReceivingMessageSource(Twitter twitter) { - super(twitter); + public MentionsReceivingMessageSource(Twitter twitter, String metadataKey) { + super(twitter, metadataKey); } - @Override public String getComponentType() { return "twitter:mentions-inbound-channel-adapter"; diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSource.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSource.java index d84ce95b3f..ae8bf4a377 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSource.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2013 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. @@ -35,11 +35,10 @@ public class SearchReceivingMessageSource extends AbstractTwitterMessageSource { - public TimelineReceivingMessageSource(Twitter twitter) { - super(twitter); + public TimelineReceivingMessageSource(Twitter twitter, String metadataKey) { + super(twitter, metadataKey); } - @Override public String getComponentType() { - return "twitter:inbound-channel-adapter"; + return "twitter:inbound-channel-adapter"; } @Override diff --git a/spring-integration-twitter/src/main/resources/org/springframework/integration/twitter/config/spring-integration-twitter-3.0.xsd b/spring-integration-twitter/src/main/resources/org/springframework/integration/twitter/config/spring-integration-twitter-3.0.xsd index 3a72f4b8a9..c98713f305 100644 --- a/spring-integration-twitter/src/main/resources/org/springframework/integration/twitter/config/spring-integration-twitter-3.0.xsd +++ b/spring-integration-twitter/src/main/resources/org/springframework/integration/twitter/config/spring-integration-twitter-3.0.xsd @@ -123,7 +123,28 @@ - + + + + The bean id of this Polling Endpoint; the MessageSource is also registered with this id + plus a suffix '.source'; also used as the + MetaDataStore key with suffix '.' + the profileId from the authorized Twitter user. + + + + + + + + + + + + Identifies the channel the attached to this adapter, to which messages will be sent. + + + + @@ -136,6 +157,21 @@ + + + + Reference to a MetadataStore instance for storing metadata associated with + the retrieved feeds. If the implementation is persistent, it can help to + prevent duplicates between restarts. If shared, it can help coordinate multiple + instances of an adapter across different processes. + + + + + + + + diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestReceivingMessageSourceParserTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestReceivingMessageSourceParserTests.java index 73946ecd36..ec97511362 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestReceivingMessageSourceParserTests.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/config/TestReceivingMessageSourceParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2013 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. @@ -16,7 +16,10 @@ package org.springframework.integration.twitter.config; +import static org.junit.Assert.assertNotNull; + import org.junit.Test; + import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; @@ -25,16 +28,13 @@ import org.springframework.integration.twitter.inbound.DirectMessageReceivingMes import org.springframework.integration.twitter.inbound.MentionsReceivingMessageSource; import org.springframework.integration.twitter.inbound.TimelineReceivingMessageSource; -import static org.junit.Assert.assertNotNull; - /** * @author Oleg Zhurakousky + * @author Gunnar Hillert */ public class TestReceivingMessageSourceParserTests { - - @Test public void testReceivingAdapterConfigurationAutoStartup(){ ApplicationContext ac = new ClassPathXmlApplicationContext("TestReceivingMessageSourceParser-context.xml", this.getClass()); @@ -47,10 +47,24 @@ public class TestReceivingMessageSourceParserTests { assertNotNull(dms); spca = ac.getBean("updateAdapter", SourcePollingChannelAdapter.class); - + spca = ac.getBean("updateAdapter", SourcePollingChannelAdapter.class); TimelineReceivingMessageSource tms = TestUtils.getPropertyValue(spca, "source", TimelineReceivingMessageSource.class); assertNotNull(tms); } + @Test + public void testThatMessageSourcesAreRegisteredAsBeans(){ + ApplicationContext ac = new ClassPathXmlApplicationContext("TestReceivingMessageSourceParser-context.xml", this.getClass()); + + MentionsReceivingMessageSource ms = ac.getBean("mentionAdapter.source", MentionsReceivingMessageSource.class); + assertNotNull(ms); + + DirectMessageReceivingMessageSource dms = ac.getBean("dmAdapter.source", DirectMessageReceivingMessageSource.class); + assertNotNull(dms); + + TimelineReceivingMessageSource tms = ac.getBean("updateAdapter.source", TimelineReceivingMessageSource.class); + assertNotNull(tms); + } + } diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSourceTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSourceTests.java index 743039925c..4c2caec1ea 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSourceTests.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSourceTests.java @@ -31,7 +31,7 @@ import org.springframework.social.twitter.api.impl.TwitterTemplate; */ public class DirectMessageReceivingMessageSourceTests { - + @SuppressWarnings("unchecked") @Test @Ignore public void demoReceiveDm() throws Exception{ @@ -40,11 +40,11 @@ public class DirectMessageReceivingMessageSourceTests { pf.afterPropertiesSet(); Properties prop = pf.getObject(); System.out.println(prop); - TwitterTemplate template = new TwitterTemplate(prop.getProperty("z_oleg.oauth.consumerKey"), - prop.getProperty("z_oleg.oauth.consumerSecret"), - prop.getProperty("z_oleg.oauth.accessToken"), + TwitterTemplate template = new TwitterTemplate(prop.getProperty("z_oleg.oauth.consumerKey"), + prop.getProperty("z_oleg.oauth.consumerSecret"), + prop.getProperty("z_oleg.oauth.accessToken"), prop.getProperty("z_oleg.oauth.accessTokenSecret")); - DirectMessageReceivingMessageSource tSource = new DirectMessageReceivingMessageSource(template); + DirectMessageReceivingMessageSource tSource = new DirectMessageReceivingMessageSource(template, "foo"); tSource.afterPropertiesSet(); for (int i = 0; i < 50; i++) { Message message = (Message) tSource.receive(); diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceTests.java index 7fb524243c..1eb70124ee 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceTests.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceTests.java @@ -35,7 +35,7 @@ import org.junit.Test; import org.springframework.beans.factory.config.PropertiesFactoryBean; import org.springframework.core.io.ClassPathResource; import org.springframework.integration.Message; -import org.springframework.integration.store.metadata.SimpleMetadataStore; +import org.springframework.integration.metadata.SimpleMetadataStore; import org.springframework.integration.test.util.TestUtils; import org.springframework.social.twitter.api.SearchMetadata; import org.springframework.social.twitter.api.SearchOperations; @@ -66,7 +66,7 @@ public class SearchReceivingMessageSourceTests { prop.getProperty("z_oleg.oauth.consumerSecret"), prop.getProperty("z_oleg.oauth.accessToken"), prop.getProperty("z_oleg.oauth.accessTokenSecret")); - SearchReceivingMessageSource tSource = new SearchReceivingMessageSource(template); + SearchReceivingMessageSource tSource = new SearchReceivingMessageSource(template, "foo"); tSource.setQuery(SEARCH_QUERY); tSource.afterPropertiesSet(); for (int i = 0; i < 50; i++) { @@ -84,14 +84,14 @@ public class SearchReceivingMessageSourceTests { @Test public void testSearchReceivingMessageSourceInit() { - final SearchReceivingMessageSource messageSource = new SearchReceivingMessageSource(new TwitterTemplate("test")); + final SearchReceivingMessageSource messageSource = new SearchReceivingMessageSource(new TwitterTemplate("test"), "foo"); messageSource.setComponentName("twitterSearchMessageSource"); final Object metadataStore = TestUtils.getPropertyValue(messageSource, "metadataStore"); final Object metadataKey = TestUtils.getPropertyValue(messageSource, "metadataKey"); assertNull(metadataStore); - assertNull(metadataKey); + assertNotNull(metadataKey); messageSource.afterPropertiesSet(); @@ -101,7 +101,7 @@ public class SearchReceivingMessageSourceTests { assertNotNull(metadataStoreInitialized); assertTrue(metadataStoreInitialized instanceof SimpleMetadataStore); assertNotNull(metadataKeyInitialized); - assertEquals("twitter:search-inbound-channel-adapter.twitterSearchMessageSource", metadataKeyInitialized); + assertEquals("foo", metadataKeyInitialized); final Twitter twitter = TestUtils.getPropertyValue(messageSource, "twitter", Twitter.class); @@ -123,7 +123,7 @@ public class SearchReceivingMessageSourceTests { when(twitterTemplate.searchOperations()).thenReturn(so); when(twitterTemplate.searchOperations().search(SEARCH_QUERY, 20, 0, 0)).thenReturn(null); - final SearchReceivingMessageSource messageSource = new SearchReceivingMessageSource(twitterTemplate); + final SearchReceivingMessageSource messageSource = new SearchReceivingMessageSource(twitterTemplate, "foo"); messageSource.setQuery(SEARCH_QUERY); final String setQuery = TestUtils.getPropertyValue(messageSource, "query", String.class); @@ -165,7 +165,7 @@ public class SearchReceivingMessageSourceTests { SearchParameters params = new SearchParameters(SEARCH_QUERY).count(20).sinceId(0); when(twitterTemplate.searchOperations().search(params)).thenReturn(results); - final SearchReceivingMessageSource messageSource = new SearchReceivingMessageSource(twitterTemplate); + final SearchReceivingMessageSource messageSource = new SearchReceivingMessageSource(twitterTemplate, "foo"); messageSource.setQuery(SEARCH_QUERY); diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceWithRedisTests-context.xml b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceWithRedisTests-context.xml index 1dcc01831e..b750aaf2f8 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceWithRedisTests-context.xml +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceWithRedisTests-context.xml @@ -1,27 +1,27 @@ - - + + + - + - + diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceWithRedisTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceWithRedisTests.java index 4985dfea89..c29762180f 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceWithRedisTests.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/SearchReceivingMessageSourceWithRedisTests.java @@ -19,6 +19,7 @@ package org.springframework.integration.twitter.inbound; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.mockito.Matchers.any; import static org.mockito.Mockito.mock; @@ -30,47 +31,61 @@ import java.util.List; import org.junit.Before; import org.junit.Test; + import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.ImportResource; -import org.springframework.data.redis.connection.RedisConnectionFactory; -import org.springframework.data.redis.core.StringRedisTemplate; -import org.springframework.integration.Message; +import org.springframework.integration.core.PollableChannel; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; import org.springframework.integration.redis.rules.RedisAvailable; import org.springframework.integration.redis.rules.RedisAvailableTests; -import org.springframework.integration.redis.store.metadata.RedisMetadataStore; -import org.springframework.integration.store.metadata.MetadataStore; +import org.springframework.integration.redis.metadata.RedisMetadataStore; +import org.springframework.integration.metadata.MetadataStore; import org.springframework.integration.test.util.TestUtils; import org.springframework.social.twitter.api.SearchMetadata; import org.springframework.social.twitter.api.SearchOperations; import org.springframework.social.twitter.api.SearchResults; import org.springframework.social.twitter.api.Tweet; +import org.springframework.social.twitter.api.UserOperations; import org.springframework.social.twitter.api.impl.SearchParameters; import org.springframework.social.twitter.api.impl.TwitterTemplate; /** * @author Gunnar Hillert + * @author Artem Bilan * @since 3.0 */ public class SearchReceivingMessageSourceWithRedisTests extends RedisAvailableTests { private SourcePollingChannelAdapter twitterSearchAdapter; - private RedisConnectionFactory redisConnectionFactory; - private StringRedisTemplate redisTemplate; - private AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + private AbstractTwitterMessageSource twitterMessageSource; + + private MetadataStore metadataStore; + + private String metadataKey; + + private PollableChannel tweets; @Before public void setup() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); context.register(SearchReceivingMessageSourceWithRedisTestsConfig.class); context.registerShutdownHook(); context.refresh(); - this.redisConnectionFactory = context.getBean(RedisConnectionFactory.class); this.twitterSearchAdapter = context.getBean(SourcePollingChannelAdapter.class); - this.redisTemplate = new StringRedisTemplate(redisConnectionFactory); + this.twitterMessageSource = context.getBean(AbstractTwitterMessageSource.class); + this.metadataStore = context.getBean(MetadataStore.class); + this.tweets = context.getBean("inbound_twitter", PollableChannel.class); + + this.metadataKey = TestUtils.getPropertyValue(twitterSearchAdapter, "source.metadataKey", String.class); + + // There is need to set a value, not 'remove' and re-init 'twitterMessageSource' + this.metadataStore.put(metadataKey, "-1"); + + this.twitterMessageSource.afterPropertiesSet(); } /** @@ -80,51 +95,43 @@ public class SearchReceivingMessageSourceWithRedisTests extends RedisAvailableTe @Test @RedisAvailable public void testPollForTweetsThreeResultsWithRedisMetadataStore() throws Exception { - - final MetadataStore metadataStore = TestUtils.getPropertyValue(twitterSearchAdapter, "source.metadataStore", MetadataStore.class); + MetadataStore metadataStore = TestUtils.getPropertyValue(this.twitterSearchAdapter, "source.metadataStore", MetadataStore.class); assertTrue("Exptected metadataStore to be an instance of RedisMetadataStore", metadataStore instanceof RedisMetadataStore); + assertSame(this.metadataStore, metadataStore); - /* - * The metadataKey is automatically generated. To ensure that we use the - * the correct key, we retrieve it from the adapter. - */ - final String metadataKey = TestUtils.getPropertyValue(twitterSearchAdapter, "source.metadataKey", String.class); + assertEquals("twitterSearchAdapter.74", metadataKey); - /* - * As we had to retrieve the metadataKey from the adapter. The metdataStore - * was already invoked and the id retrieved from Redis before we had a chance - * to reset possibly pre-existing values. - * - * Rather than deleting the value, we have to set a value, because "null" values - * returned from the MetadataStore are ignored by the onInit() method in - * the AbstractTwitterMessageSource. */ - redisTemplate.opsForValue().set(metadataKey, "-1"); - assertEquals("-1", redisTemplate.opsForValue().get(metadataKey)); + this.twitterSearchAdapter.start(); - final SearchReceivingMessageSource source = TestUtils.getPropertyValue(twitterSearchAdapter, "source", SearchReceivingMessageSource.class); - - /* We need to call onInit() in order to update the id from the metadataStore. */ - source.onInit(); - - final Message message1 = source.receive(); - final Message message2 = source.receive(); - final Message message3 = source.receive(); + assertNotNull(this.tweets.receive(10000)); + assertNotNull(this.tweets.receive(1000)); + assertNotNull(this.tweets.receive(1000)); /* We received 3 messages so far. When invoking receive() again the search * will return again the 3 test Tweets but as we already processed them * no message (null) is returned. */ - final Message message4 = source.receive(); + assertNull(this.tweets.receive(0)); - assertNotNull(message1); - assertNotNull(message2); - assertNotNull(message3); - assertNull(message4); - - final String persistedMetadataStoreValue = redisTemplate.opsForValue().get(metadataKey); + String persistedMetadataStoreValue = this.metadataStore.get(metadataKey); assertNotNull(persistedMetadataStoreValue); - assertEquals("3", redisTemplate.opsForValue().get(metadataKey)); + assertEquals("3", persistedMetadataStoreValue); - redisTemplate.delete(metadataKey); + this.twitterSearchAdapter.stop(); + + this.metadataStore.put(metadataKey, "1"); + + this.twitterMessageSource.afterPropertiesSet(); + + this.twitterSearchAdapter.start(); + + assertNotNull(this.tweets.receive(1000)); + assertNotNull(this.tweets.receive(1000)); + + assertNull(this.tweets.receive(0)); + + persistedMetadataStoreValue = this.metadataStore.get(metadataKey); + assertNotNull(persistedMetadataStoreValue); + assertEquals("3", persistedMetadataStoreValue); } @Configuration @@ -152,7 +159,15 @@ public class SearchReceivingMessageSourceWithRedisTests extends RedisAvailableTe when(twitterTemplate.searchOperations()).thenReturn(so); when(twitterTemplate.searchOperations().search(any(SearchParameters.class))).thenReturn(results); + when(twitterTemplate.isAuthorized()).thenReturn(true); + + final UserOperations userOperations = mock(UserOperations.class); + when(twitterTemplate.userOperations()).thenReturn(userOperations); + when(userOperations.getProfileId()).thenReturn(74L); + return twitterTemplate; } + } + } diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/TimelineReceivingMessageSourceTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/TimelineReceivingMessageSourceTests.java index 6782c65222..f568f63c55 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/TimelineReceivingMessageSourceTests.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/TimelineReceivingMessageSourceTests.java @@ -32,7 +32,7 @@ import org.springframework.social.twitter.api.impl.TwitterTemplate; */ public class TimelineReceivingMessageSourceTests { - + @SuppressWarnings("unchecked") @Test @Ignore public void demoReceiveTimeline() throws Exception{ @@ -41,11 +41,11 @@ public class TimelineReceivingMessageSourceTests { pf.afterPropertiesSet(); Properties prop = pf.getObject(); System.out.println(prop); - TwitterTemplate template = new TwitterTemplate(prop.getProperty("z_oleg.oauth.consumerKey"), - prop.getProperty("z_oleg.oauth.consumerSecret"), - prop.getProperty("z_oleg.oauth.accessToken"), + TwitterTemplate template = new TwitterTemplate(prop.getProperty("z_oleg.oauth.consumerKey"), + prop.getProperty("z_oleg.oauth.consumerSecret"), + prop.getProperty("z_oleg.oauth.accessToken"), prop.getProperty("z_oleg.oauth.accessTokenSecret")); - TimelineReceivingMessageSource tSource = new TimelineReceivingMessageSource(template); + TimelineReceivingMessageSource tSource = new TimelineReceivingMessageSource(template, "foo"); tSource.afterPropertiesSet(); for (int i = 0; i < 50; i++) { Message message = (Message) tSource.receive(); diff --git a/src/reference/docbook/feed.xml b/src/reference/docbook/feed.xml index 5215161a09..cb02137b39 100644 --- a/src/reference/docbook/feed.xml +++ b/src/reference/docbook/feed.xml @@ -62,14 +62,14 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/feed Spring Integration provides a convenient mechanism to eliminate the need to worry about duplicate entries. Each feed entry will have a published date field. Every time a new Message is generated and sent, Spring Integration will store the value of the latest published date in an instance of the - org.springframework.integration.store.MetadataStore strategy. The MetadataStore interface is + org.springframework.integration.metadata.MetadataStore strategy. The MetadataStore interface is designed to store various types of generic meta-data (e.g., published date of the last feed entry that has been processed) to help components such as this Feed adapter deal with duplicates. The default rule for locating this metadata store is as follows: Spring Integration will look for a bean of type - org.springframework.integration.store.MetadataStore in + org.springframework.integration.metadata.MetadataStore in the ApplicationContext. If one is found then it will be used, otherwise it will create a new instance of SimpleMetadataStore which is an in-memory implementation that will only persist metadata within @@ -107,5 +107,10 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/feed MetadataStore interface (e.g. JdbcMetadataStore) and configure it as bean in the Application Context. + + The key used to persist the latest published date is the value of the (required) + id attribute of the Feed Inbound Channel Adapter component plus the feedUrl + from the adapter's configuration. + diff --git a/src/reference/docbook/twitter.xml b/src/reference/docbook/twitter.xml index 40c4d7bb97..c78a2e7cc9 100644 --- a/src/reference/docbook/twitter.xml +++ b/src/reference/docbook/twitter.xml @@ -141,11 +141,12 @@ twitter.oauth.accessTokenSecret=AbRxUAvyNCtqQtxFK8w5ZMtMj20KFhB6o]]>org.springframework.integration.store.MetadataStore which is a + The latest Tweet id will be stored in an instance of the org.springframework.integration.metadata.MetadataStore which is a strategy interface designed for storing various types of metadata (e.g., last retrieved tweet in this case). That strategy helps components such as these Twitter adapters avoid duplicates. By default, Spring Integration will look for a bean of type - org.springframework.integration.store.MetadataStore in the ApplicationContext. - If one is found then it will be used, otherwise it will create a new instance of SimpleMetadataStore + org.springframework.integration.metadata.MetadataStore in the ApplicationContext. Alternatively, + you can configure an explicit MetadataStore on the adapter. + If there is no explicit or default store, the adapter will create a new instance of SimpleMetadataStore which is a simple in-memory implementation that will only persist metadata within the lifecycle of the currently running application context. That means upon restart you may end up with duplicate entries. If you need to persist metadata between Application Context restarts, you may use the PropertiesPersistingMetadataStore (which is backed by a properties file, and a persister @@ -165,8 +166,16 @@ twitter.oauth.accessTokenSecret=AbRxUAvyNCtqQtxFK8w5ZMtMj20KFhB6o]]> ]]> -The Poller that is configured as part of any Inbound Twitter Adapter (see below) will simply poll from this MetadataStore to determine the latest tweet -received. + + If the MetadataStore is persistent, during initialization, any Inbound Twitter Adapter (see below) + will retrieve the latest tweet id that has already been sent by the adapter. + + + The key used to persist the latest twitter id is the value of the (required) + id attribute of the Twitter Inbound Channel Adapter component plus the profileId + of the Twitter user. + +
Inbound Message Channel Adapter