From 7412bc239c008887ee10dcde25fcd2deb682a076 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 26 Oct 2010 18:39:52 -0400 Subject: [PATCH] INT-1527 MetadataStore refactoring --- .../context/IntegrationContextUtils.java | 9 +- .../context/IntegrationObjectSupport.java | 1 + .../context/metadata/MetadataStore.java | 13 +- ...=> PropertiesPersistingMetadataStore.java} | 147 +++++++----- .../context/metadata/SimpleMetadataStore.java | 39 +++ ...opertiesPersistingMetadataStoreTests.java} | 52 ++-- .../feed/FeedEntryReaderMessageSource.java | 225 +++++++++--------- ...FeedMessageSourceBeanDefinitionParser.java | 27 +-- .../FeedEntryReaderMessageSourceTests.java | 69 ++++-- ...BeanDefinitionParserTests-file-context.xml | 12 +- ...finitionParserTests-file-usage-context.xml | 4 +- ...essageSourceBeanDefinitionParserTests.java | 94 ++++---- ...AbstractInboundTwitterEndpointSupport.java | 47 ++-- ...boundDirectMessageStatusEndpointTests.java | 62 ++--- 14 files changed, 439 insertions(+), 362 deletions(-) rename spring-integration-core/src/main/java/org/springframework/integration/context/metadata/{FileBasedPropertiesStore.java => PropertiesPersistingMetadataStore.java} (50%) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/context/metadata/SimpleMetadataStore.java rename spring-integration-core/src/test/java/org/springframework/integration/context/metadata/{FileBasedPropertiesStoreTests.java => PropertiesPersistingMetadataStoreTests.java} (56%) 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 48c2ee9d4f..7a8428af96 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 @@ -36,20 +36,21 @@ import org.springframework.util.Assert; */ public abstract class IntegrationContextUtils { - public static final String METADATA_PERSISTER_BEAN_NAME = "metadataPersister"; - public static final String TASK_SCHEDULER_BEAN_NAME = "taskScheduler"; public static final String ERROR_CHANNEL_BEAN_NAME = "errorChannel"; public static final String NULL_CHANNEL_BEAN_NAME = "nullChannel"; + public static final String METADATA_STORE_BEAN_NAME = "metadataStore"; + public static final String INTEGRATION_CONVERSION_SERVICE_BEAN_NAME = "integrationConversionService"; public static final String DEFAULT_POLLER_METADATA_BEAN_NAME = "org.springframework.integration.context.defaultPollerMetadata"; - public static MetadataStore getMetadataPersister(BeanFactory beanFactory) { - return getBeanOfType(beanFactory, METADATA_PERSISTER_BEAN_NAME, MetadataStore.class); + + public static MetadataStore getMetadataStore(BeanFactory beanFactory) { + return getBeanOfType(beanFactory, METADATA_STORE_BEAN_NAME, MetadataStore.class); } public static MessageChannel getErrorChannel(BeanFactory beanFactory) { 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 2e063fb60b..9685b80f02 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 @@ -18,6 +18,7 @@ package org.springframework.integration.context; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.BeanInitializationException; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/MetadataStore.java b/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/MetadataStore.java index 5c49742219..bd085dfe7e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/MetadataStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/MetadataStore.java @@ -16,26 +16,25 @@ package org.springframework.integration.context.metadata; -import java.util.Properties; - /** - * Strategy interface for persisting metadata from certain adapters / endpoints + * Strategy interface for storing metadata from certain adapters * to avoid duplicate delivery of messages, for example. * * @author Josh Long * @author Oleg Zhurakousky + * @author Mark Fisher * @since 2.0 */ public interface MetadataStore { /** - * Writes metadata as Properties to this store. + * Writes a key value pair to this MetadataStore. */ - void write(Properties metadata); + void put(String key, String value); /** - * Loads metadata as Properties from this store. + * Reads a value for the given key from this MetadataStore. */ - Properties load(); + String get(String key); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/FileBasedPropertiesStore.java b/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/PropertiesPersistingMetadataStore.java similarity index 50% rename from spring-integration-core/src/main/java/org/springframework/integration/context/metadata/FileBasedPropertiesStore.java rename to spring-integration-core/src/main/java/org/springframework/integration/context/metadata/PropertiesPersistingMetadataStore.java index 2f2036a846..21a2a10fd0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/FileBasedPropertiesStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/PropertiesPersistingMetadataStore.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.context.metadata; import java.io.File; @@ -23,6 +24,8 @@ import java.util.Properties; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + +import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.util.Assert; import org.springframework.util.DefaultPropertiesPersister; @@ -34,87 +37,101 @@ import org.springframework.util.DefaultPropertiesPersister; * Files will be written to the 'java.io.tmpdir' + "/spring-integration/". * * @author Oleg Zhurakousky + * @author Mark Fisher * @since 2.0 */ -public class FileBasedPropertiesStore implements MetadataStore, InitializingBean{ +public class PropertiesPersistingMetadataStore implements MetadataStore, InitializingBean, DisposableBean { + private final Log logger = LogFactory.getLog(getClass()); + + private final Properties metadata = new Properties(); + private final DefaultPropertiesPersister persister = new DefaultPropertiesPersister(); - private final String persistentKey; - private volatile File persistentFile; + + private volatile File file; + private volatile String baseDirectory = System.getProperty("java.io.tmpdir") + "/spring-integration/"; - public FileBasedPropertiesStore(String persistentKey){ - Assert.notNull(persistentKey, "'persistentKey' must not be null"); - this.persistentKey = persistentKey; - } - + public void setBaseDirectory(String baseDirectory) { + Assert.hasText(baseDirectory, "'baseDirectory' must be non-empty"); this.baseDirectory = baseDirectory; } - - public String getBaseDirectory() { - return baseDirectory; - } - - public void write(Properties metadata) { - FileOutputStream fo = null; - try { - fo = new FileOutputStream(persistentFile); - persister.store(metadata, fo, "Last feed entry"); - } - catch (IOException e) { - // not fatal for the functionality of the component - logger.warn("Failed to persist feed entry. This may result in a duplicate " + - "feed entry after this component is restarted", e); - } - finally { - try { - if (fo != null){ - fo.close(); - } - } - catch (IOException e) { - // not fatal for the functionality of he component - logger.warn("Failed to close FileOutputStream to " + persistentFile.getAbsolutePath(), e); - } - } - } - - public Properties load() { - Properties properties = new Properties(); - FileInputStream iStream = null; - try { - iStream = new FileInputStream(persistentFile); - persister.load(properties, iStream); - } catch (Exception e) { - // not fatal for the functionality of the component - logger.warn("Failed to load feed entry from the persistent store. This may result in a duplicate " + - "feed entry after this component is restarted", e); - } finally { - try { - if (iStream != null){ - iStream.close(); - } - } catch (Exception e2) { - // non fatal - logger.warn("Failed to close FileInputStream for: " + persistentFile.getAbsolutePath()); - } - } - return properties; - } public void afterPropertiesSet() throws Exception { - String fileName = this.persistentKey + ".last.entry"; File baseDir = new File(baseDirectory); baseDir.mkdirs(); - persistentFile = new File(baseDir, fileName); + this.file = new File(baseDir, "metadata-store.properties"); try { - if (!persistentFile.exists()){ - persistentFile.createNewFile(); + if (!this.file.exists()) { + this.file.createNewFile(); } - } catch (Exception e) { + } + catch (Exception e) { throw new IllegalArgumentException("Failed to create metadata-store file '" - + persistentFile.getAbsolutePath() + "'", e); + + this.file.getAbsolutePath() + "'", e); + } + this.loadMetadata(); + } + + public void put(String key, String value) { + this.metadata.setProperty(key, value); + } + + public String get(String key) { + return this.metadata.getProperty(key); + } + + public void destroy() throws Exception { + this.saveMetadata(); + } + + private void saveMetadata() { + FileOutputStream outputStream = null; + try { + outputStream = new FileOutputStream(this.file); + this.persister.store(this.metadata, outputStream, "Last feed entry"); + } + catch (IOException e) { + // not fatal for the functionality of the component + logger.warn("Failed to persist feed entry. This may result in a duplicate " + + "feed entry after this component is restarted.", e); + } + finally { + try { + if (outputStream != null) { + outputStream.close(); + } + } + catch (IOException e) { + // not fatal for the functionality of the component + logger.warn("Failed to close FileOutputStream to " + this.file.getAbsolutePath(), e); + } } } + + private void loadMetadata() { + FileInputStream inputStream = null; + try { + inputStream = new FileInputStream(this.file); + this.persister.load(this.metadata, inputStream); + } + catch (Exception e) { + // not fatal for the functionality of the component + logger.warn("Failed to load feed entry from the persistent store. This may result in a duplicate " + + "feed entry after this component is restarted", e); + } + finally { + try { + if (inputStream != null) { + inputStream.close(); + } + } + catch (Exception e2) { + // non fatal + logger.warn("Failed to close FileInputStream for: " + this.file.getAbsolutePath()); + } + } + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/SimpleMetadataStore.java b/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/SimpleMetadataStore.java new file mode 100644 index 0000000000..2ade1e3545 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/metadata/SimpleMetadataStore.java @@ -0,0 +1,39 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ + +package org.springframework.integration.context.metadata; + +import java.util.HashMap; +import java.util.Map; + +/** + * Simple implementation of {@link MetadataStore} that uses an in-memory map only. + * The metadata will not be persisted across application restarts. + * + * @author Mark Fisher + * @since 2.0 + */ +public class SimpleMetadataStore implements MetadataStore { + + private final Map metadata = new HashMap(); + + + public void put(String key, String value) { + this.metadata.put(key, value); + } + + public String get(String key) { + return this.metadata.get(key); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/context/metadata/FileBasedPropertiesStoreTests.java b/spring-integration-core/src/test/java/org/springframework/integration/context/metadata/PropertiesPersistingMetadataStoreTests.java similarity index 56% rename from spring-integration-core/src/test/java/org/springframework/integration/context/metadata/FileBasedPropertiesStoreTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/context/metadata/PropertiesPersistingMetadataStoreTests.java index d557a47ece..4d7d76da20 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/context/metadata/FileBasedPropertiesStoreTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/context/metadata/PropertiesPersistingMetadataStoreTests.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.context.metadata; import static junit.framework.Assert.assertEquals; @@ -24,47 +25,46 @@ import java.util.Properties; import org.junit.Test; +import org.springframework.core.io.FileSystemResource; +import org.springframework.core.io.support.PropertiesLoaderUtils; + /** * @author Oleg Zhurakousky - * + * @author Mark Fisher + * @since 2.0 */ -public class FileBasedPropertiesStoreTests { +public class PropertiesPersistingMetadataStoreTests { - @Test(expected=IllegalArgumentException.class) - public void validateFailureWithNoPersistentKey(){ - new FileBasedPropertiesStore(null); - } - @Test - public void validateWithDefaultBaseDir() throws Exception{ - File file = new File(System.getProperty("java.io.tmpdir") + "/spring-integration/" + "foo.last.entry"); + public void validateWithDefaultBaseDir() throws Exception { + File file = new File(System.getProperty("java.io.tmpdir") + "/spring-integration/metadata-store.properties"); file.delete(); - FileBasedPropertiesStore metaStore = new FileBasedPropertiesStore("foo"); - metaStore.afterPropertiesSet(); + PropertiesPersistingMetadataStore metadataStore = new PropertiesPersistingMetadataStore(); + metadataStore.afterPropertiesSet(); assertTrue(file.exists()); - Properties prop = new Properties(); - prop.setProperty("foo", "bar"); - metaStore.write(prop); - Properties persistentProperties = metaStore.load(); + metadataStore.put("foo", "bar"); + metadataStore.destroy(); + Properties persistentProperties = PropertiesLoaderUtils.loadProperties(new FileSystemResource(file)); assertNotNull(persistentProperties); assertEquals(1, persistentProperties.size()); assertEquals("bar", persistentProperties.get("foo")); - } - @Test - public void validateWithCustomBaseDir() throws Exception{ - File file = new File("foo/" + "foo.last.entry"); file.delete(); + } + + @Test + public void validateWithCustomBaseDir() throws Exception { + File file = new File("foo" + "/metadata-store.properties"); file.deleteOnExit(); - FileBasedPropertiesStore metaStore = new FileBasedPropertiesStore("foo"); - metaStore.setBaseDirectory("foo"); - metaStore.afterPropertiesSet(); + PropertiesPersistingMetadataStore metadataStore = new PropertiesPersistingMetadataStore(); + metadataStore.setBaseDirectory("foo"); + metadataStore.afterPropertiesSet(); + metadataStore.put("foo", "bar"); + metadataStore.destroy(); assertTrue(file.exists()); - Properties prop = new Properties(); - prop.setProperty("foo", "bar"); - metaStore.write(prop); - Properties persistentProperties = metaStore.load(); + Properties persistentProperties = PropertiesLoaderUtils.loadProperties(new FileSystemResource(file)); assertNotNull(persistentProperties); assertEquals(1, persistentProperties.size()); assertEquals("bar", persistentProperties.get("foo")); } + } 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 index 80e578bbec..ab542ce85c 100644 --- 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 @@ -13,19 +13,21 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.feed; import java.util.Collections; import java.util.Comparator; import java.util.List; -import java.util.Properties; import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; +import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.Message; +import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.context.IntegrationObjectSupport; -import org.springframework.integration.context.metadata.FileBasedPropertiesStore; import org.springframework.integration.context.metadata.MetadataStore; +import org.springframework.integration.context.metadata.SimpleMetadataStore; import org.springframework.integration.core.MessageSource; import org.springframework.integration.support.MessageBuilder; import org.springframework.util.Assert; @@ -35,131 +37,128 @@ import com.sun.syndication.feed.synd.SyndEntry; import com.sun.syndication.feed.synd.SyndFeed; /** - * This implementation of {@link MessageSource} will produce individual {@link SyndEntry}s for a feed identified - * with 'feedUrl' attribute. + * This implementation of {@link MessageSource} will produce individual + * {@link SyndEntry}s for a feed identified with the 'feedUrl' attribute. * * @author Josh Long * @author Mario Gray * @author Oleg Zhurakousky + * @since 2.0 */ -public class FeedEntryReaderMessageSource extends IntegrationObjectSupport implements MessageSource{ +public class FeedEntryReaderMessageSource extends IntegrationObjectSupport implements MessageSource { + + private final Queue entries = new ConcurrentLinkedQueue(); + + private final FeedReaderMessageSource feedReaderMessageSource; + + private volatile String metadataKey; + private volatile MetadataStore metadataStore; - private volatile Properties lastPersistentEntry = new Properties(); - private volatile Queue entries = new ConcurrentLinkedQueue(); - private volatile FeedReaderMessageSource feedReaderMessageSource; - private final Object monitor = new Object(); - private volatile String feedMetadataIdKey; - private volatile String persistentIdentifier; - private volatile boolean initialized; - private volatile long lastTime = -1; + private volatile long lastTime = -1; - private Comparator syndEntryComparator = new Comparator() { - public int compare(SyndEntry syndEntry, SyndEntry syndEntry1) { - long x = syndEntry.getPublishedDate().getTime() - - syndEntry1.getPublishedDate().getTime(); - if (x < -1) { - return -1; - } - else if (x > 1) { - return 1; - } - return 0; - } - }; - - public FeedEntryReaderMessageSource(FeedReaderMessageSource feedReaderMessageSource) { - Assert.notNull(feedReaderMessageSource, "'feedReaderMessageSource' must not be null"); + private volatile boolean initialized; + + private final Object monitor = new Object(); + + private final Comparator syndEntryComparator = new SyndEntryComparator(); + + + public FeedEntryReaderMessageSource(FeedReaderMessageSource feedReaderMessageSource) { + Assert.notNull(feedReaderMessageSource, "'feedReaderMessageSource' must not be null"); this.feedReaderMessageSource = feedReaderMessageSource; } - public void setPersistentIdentifier(String persistentIdentifier) { - this.persistentIdentifier = persistentIdentifier; - } - - public void setMetadataStore(MetadataStore metadataStore) { + + public void setMetadataStore(MetadataStore metadataStore) { + Assert.notNull(metadataStore, "metadataStore must not be null"); this.metadataStore = metadataStore; } - - public String getComponentType(){ - return "feed:inbound-channel-adapter"; - } - public Message receive() { - Assert.isTrue(this.initialized, "'FeedEntryReaderMessageSource' must be initialized before it can produce Messages"); - SyndEntry se = doReceieve(); - if (se == null) { - return null; - } - return MessageBuilder.withPayload(se).build(); - } - - @SuppressWarnings("unchecked") - private SyndEntry doReceieve() { - SyndEntry nextUp = null; - synchronized (this.monitor) { - nextUp = pollAndCache(); + public String getComponentType() { + return "feed:inbound-channel-adapter"; + } - 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) { - long publishedTime = se.getPublishedDate().getTime(); - if (publishedTime > this.lastTime){ - entries.add(se); - } - } - } - } - nextUp = pollAndCache(); - } - return nextUp; - } - - @Override - protected void onInit() throws Exception { - if (StringUtils.hasText(this.persistentIdentifier)){ - if (this.metadataStore == null){ - logger.info("Creating FileBasedPropertiesStore"); - metadataStore = new FileBasedPropertiesStore(this.persistentIdentifier); - ((FileBasedPropertiesStore)metadataStore).afterPropertiesSet(); - } - lastPersistentEntry = metadataStore.load(); - } - else { - logger.info("Your '" + this.getComponentType() + "' is anonymous (no ID attribute), therefore no feed entries will be persisted " + - "which may result in a duplicate feed entries once this adapter is restarted"); - } - - this.feedMetadataIdKey = this.getComponentType() + "@" + this.getComponentName() + - "#" + feedReaderMessageSource.getFeedUrl(); - String keyTime = (String) this.lastPersistentEntry.get(this.feedMetadataIdKey); - if (StringUtils.hasText(keyTime)){ - this.lastTime = Long.parseLong(keyTime); - } - this.initialized = true; - } + public Message receive() { + Assert.isTrue(this.initialized, "'FeedEntryReaderMessageSource' must be initialized before it can produce Messages."); + SyndEntry entry = doReceive(); + if (entry == null) { + return null; + } + return MessageBuilder.withPayload(entry).build(); + } + + @Override + protected void onInit() throws Exception { + if (this.metadataStore == null) { + // first try to look for a 'messageStore' in the context + BeanFactory beanFactory = this.getBeanFactory(); + if (beanFactory != null) { + MetadataStore metadataStore = IntegrationContextUtils.getMetadataStore(beanFactory); + if (metadataStore != null) { + this.metadataStore = metadataStore; + } + } + if (this.metadataStore == null) { + this.metadataStore = new SimpleMetadataStore(); + } + } + Assert.hasText(this.getComponentName(), "FeedEntryReaderMessageSource must have a name"); + this.metadataKey = this.getComponentType() + "." + this.getComponentName() + + "." + this.feedReaderMessageSource.getFeedUrl(); + String lastTimeValue = this.metadataStore.get(this.metadataKey); + if (StringUtils.hasText(lastTimeValue)) { + this.lastTime = Long.parseLong(lastTimeValue); + } + this.initialized = true; + } + + @SuppressWarnings("unchecked") + private SyndEntry doReceive() { + SyndEntry nextUp = null; + synchronized (this.monitor) { + nextUp = pollAndCache(); + if (nextUp != null) { + return nextUp; + } + // otherwise, fill the backlog + SyndFeed syndFeed = this.feedReaderMessageSource.receiveSyndFeed(); + if (syndFeed != null) { + List feedEntries = (List) syndFeed.getEntries(); + if (null != feedEntries) { + Collections.sort(feedEntries, syndEntryComparator); + for (SyndEntry se : feedEntries) { + long publishedTime = se.getPublishedDate().getTime(); + if (publishedTime > this.lastTime) { + entries.add(se); + } + } + } + } + nextUp = pollAndCache(); + } + return nextUp; + } private SyndEntry pollAndCache() { - SyndEntry next = this.entries.poll(); - - if (next == null) { - return null; - } - - this.lastTime = next.getPublishedDate().getTime(); - this.lastPersistentEntry.put(this.feedMetadataIdKey, this.lastTime + ""); - - if (metadataStore != null){ - metadataStore.write(this.lastPersistentEntry); - } - - return next; - } -} \ No newline at end of file + SyndEntry next = this.entries.poll(); + if (next == null) { + return null; + } + this.lastTime = next.getPublishedDate().getTime(); + this.metadataStore.put(this.metadataKey, this.lastTime + ""); + return next; + } + + + private static class SyndEntryComparator implements Comparator { + + public int compare(SyndEntry entry1, SyndEntry entry2) { + if (entry1 == null || entry2 == null) { + + } + return entry1.getPublishedDate().compareTo(entry2.getPublishedDate()); + } + } + +} 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 index 1fffb8781a..5b5364a569 100644 --- 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 @@ -13,41 +13,36 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.feed.config; +import org.w3c.dom.Element; + 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; -import org.w3c.dom.Element; /** * Handles parsing the configuration for the feed inbound channel adapter. * * @author Josh Long * @author Oleg Zhurakousky + * @since 2.0 */ public class FeedMessageSourceBeanDefinitionParser extends AbstractPollingInboundChannelAdapterParser { @Override protected String parseSource(final Element element, final ParserContext parserContext) { - - BeanDefinitionBuilder feedEntryBuilder = - BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.feed.FeedEntryReaderMessageSource"); - IntegrationNamespaceUtils.setValueIfAttributeDefined(feedEntryBuilder, element, "id", "persistentIdentifier"); - BeanDefinitionBuilder feedBuilder = - BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.feed.FeedReaderMessageSource"); + BeanDefinitionBuilder feedEntryBuilder = BeanDefinitionBuilder.genericBeanDefinition( + "org.springframework.integration.feed.FeedEntryReaderMessageSource"); + BeanDefinitionBuilder feedBuilder = BeanDefinitionBuilder.genericBeanDefinition( + "org.springframework.integration.feed.FeedReaderMessageSource"); feedBuilder.addConstructorArgValue(element.getAttribute("feed-url")); - - String metadataStoreStrategy = element.getAttribute("metadata-store"); - if (StringUtils.hasText(metadataStoreStrategy)){ - feedEntryBuilder.addPropertyReference("metadataStore", metadataStoreStrategy); - } - + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(feedEntryBuilder, element, "metadata-store"); feedEntryBuilder.addConstructorArgValue(feedBuilder.getBeanDefinition()); - return BeanDefinitionReaderUtils.registerWithGeneratedName(feedEntryBuilder.getBeanDefinition(), parserContext.getRegistry()); } -} \ No newline at end of file + +} diff --git a/spring-integration-feed/src/test/java/org/springframework/integration/feed/FeedEntryReaderMessageSourceTests.java b/spring-integration-feed/src/test/java/org/springframework/integration/feed/FeedEntryReaderMessageSourceTests.java index a1226963c8..64077262a7 100644 --- a/spring-integration-feed/src/test/java/org/springframework/integration/feed/FeedEntryReaderMessageSourceTests.java +++ b/spring-integration-feed/src/test/java/org/springframework/integration/feed/FeedEntryReaderMessageSourceTests.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.feed; import static junit.framework.Assert.assertEquals; @@ -29,42 +30,47 @@ import java.util.List; import org.junit.Before; import org.junit.Test; + import org.springframework.integration.Message; +import org.springframework.integration.context.metadata.PropertiesPersistingMetadataStore; import com.sun.syndication.feed.synd.SyndEntry; import com.sun.syndication.feed.synd.SyndFeed; /** * @author Oleg Zhurakousky - * + * @author Mark Fisher + * @since 2.0 */ public class FeedEntryReaderMessageSourceTests { + @Before - public void prepare(){ - File persisterFile = new File(System.getProperty("java.io.tmpdir") + "/spring-integration/", "feedReader.last.entry"); - if (persisterFile.exists()){ - persisterFile.delete(); + public void prepare() { + File metadataStoreFile = new File(System.getProperty("java.io.tmpdir") + "/spring-integration/", "metadata-store.properties"); + if (metadataStoreFile.exists()) { + metadataStoreFile.delete(); } } @Test(expected=IllegalArgumentException.class) - public void testFailureWhenNotInitialized(){ + public void testFailureWhenNotInitialized() { FeedEntryReaderMessageSource feedEntrySource = new FeedEntryReaderMessageSource(mock(FeedReaderMessageSource.class)); feedEntrySource.receive(); } - + @Test - public void testReceieveFeedWithNoEntries(){ + public void testReceiveFeedWithNoEntries() { FeedReaderMessageSource feedReaderSource = mock(FeedReaderMessageSource.class); SyndFeed feed = mock(SyndFeed.class); when(feedReaderSource.receiveSyndFeed()).thenReturn(feed); FeedEntryReaderMessageSource feedEntrySource = new FeedEntryReaderMessageSource(feedReaderSource); - feedEntrySource.setPersistentIdentifier("feedReader"); + feedEntrySource.setBeanName("feedReader"); feedEntrySource.afterPropertiesSet(); assertNull(feedEntrySource.receive()); } + @Test - public void testReceieveFeedWithEntriesSorted(){ + public void testReceiveFeedWithEntriesSorted() { FeedReaderMessageSource feedReaderSource = mock(FeedReaderMessageSource.class); SyndFeed feed = mock(SyndFeed.class); SyndEntry entry1 = mock(SyndEntry.class); @@ -79,7 +85,7 @@ public class FeedEntryReaderMessageSourceTests { when(feedReaderSource.receiveSyndFeed()).thenReturn(feed); FeedEntryReaderMessageSource feedEntrySource = new FeedEntryReaderMessageSource(feedReaderSource); - feedEntrySource.setPersistentIdentifier("feedReader"); + feedEntrySource.setComponentName("feedReader"); feedEntrySource.afterPropertiesSet(); Message entryMessage = feedEntrySource.receive(); assertEquals(entry2, entryMessage.getPayload()); @@ -89,15 +95,18 @@ public class FeedEntryReaderMessageSourceTests { entryMessage = feedEntrySource.receive(); assertNull(entryMessage); } - // will test, that last feed entry is remembered between the sessions + + // will test that last feed entry is remembered between the sessions // and no duplicate entries are retrieved @Test - public void testReceieveFeedWithRealEntriesAndRepeatWithPersistentIdentifier() throws Exception{ - FeedReaderMessageSource feedReaderSource = - new FeedReaderMessageSource(new URL("file:src/test/java/org/springframework/integration/feed/sample.rss")); - + public void testReceiveFeedWithRealEntriesAndRepeatWithPersistentMetadataStore() throws Exception { + URL url = new URL("file:src/test/java/org/springframework/integration/feed/sample.rss"); + FeedReaderMessageSource feedReaderSource = new FeedReaderMessageSource(url); FeedEntryReaderMessageSource feedEntrySource = new FeedEntryReaderMessageSource(feedReaderSource); - feedEntrySource.setPersistentIdentifier("feedReader"); + feedEntrySource.setBeanName("feedReader"); + PropertiesPersistingMetadataStore metadataStore = new PropertiesPersistingMetadataStore(); + metadataStore.afterPropertiesSet(); + feedEntrySource.setMetadataStore(metadataStore); feedEntrySource.afterPropertiesSet(); SyndEntry entry1 = feedEntrySource.receive().getPayload(); SyndEntry entry2 = feedEntrySource.receive().getPayload(); @@ -113,23 +122,29 @@ public class FeedEntryReaderMessageSourceTests { assertEquals("Spring Integration adapters", entry3.getTitle().trim()); assertEquals(1272044098000L, entry3.getPublishedDate().getTime()); + metadataStore.destroy(); + metadataStore.afterPropertiesSet(); + // now test that what's been read is no longer retrieved feedEntrySource = new FeedEntryReaderMessageSource(feedReaderSource); - feedEntrySource.setPersistentIdentifier("feedReader"); + feedEntrySource.setBeanName("feedReader"); + metadataStore = new PropertiesPersistingMetadataStore(); + metadataStore.afterPropertiesSet(); + feedEntrySource.setMetadataStore(metadataStore); feedEntrySource.afterPropertiesSet(); assertNull(feedEntrySource.receive()); assertNull(feedEntrySource.receive()); assertNull(feedEntrySource.receive()); } - // will test, that last feed entry is NOT remembered between the sessions, since - // persister is not used due to the lack of persistentIdentifier (id attribute in xml) - // and the same entries are retrieved again + + // will test that last feed entry is NOT remembered between the sessions, since + // no persistent MetadataStore is provided and the same entries are retrieved again @Test - public void testReceieveFeedWithRealEntriesAndRepeatNoPersistentIdentifier() throws Exception{ - FeedReaderMessageSource feedReaderSource = - new FeedReaderMessageSource(new URL("file:src/test/java/org/springframework/integration/feed/sample.rss")); - + public void testReceiveFeedWithRealEntriesAndRepeatNoPersistentMetadataStore() throws Exception { + URL url = new URL("file:src/test/java/org/springframework/integration/feed/sample.rss"); + FeedReaderMessageSource feedReaderSource = new FeedReaderMessageSource(url); FeedEntryReaderMessageSource feedEntrySource = new FeedEntryReaderMessageSource(feedReaderSource); + feedEntrySource.setBeanName("feedReader"); feedEntrySource.afterPropertiesSet(); SyndEntry entry1 = feedEntrySource.receive().getPayload(); SyndEntry entry2 = feedEntrySource.receive().getPayload(); @@ -148,6 +163,7 @@ public class FeedEntryReaderMessageSourceTests { // UNLIKE the previous test // now test that what's been read is read AGAIN feedEntrySource = new FeedEntryReaderMessageSource(feedReaderSource); + feedEntrySource.setBeanName("feedReader"); feedEntrySource.afterPropertiesSet(); entry1 = feedEntrySource.receive().getPayload(); entry2 = feedEntrySource.receive().getPayload(); @@ -162,5 +178,6 @@ public class FeedEntryReaderMessageSourceTests { assertEquals("Spring Integration adapters", entry3.getTitle().trim()); assertEquals(1272044098000L, entry3.getPublishedDate().getTime()); - } + } + } diff --git a/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests-file-context.xml b/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests-file-context.xml index cf4b30660f..c832a3ac07 100644 --- a/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests-file-context.xml +++ b/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests-file-context.xml @@ -5,19 +5,19 @@ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd http://www.springframework.org/schema/integration/feed http://www.springframework.org/schema/integration/feed/spring-integration-feed-2.0.xsd"> - - + - + - - + + + \ No newline at end of file diff --git a/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests-file-usage-context.xml b/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests-file-usage-context.xml index 701caed05c..041b88a519 100644 --- a/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests-file-usage-context.xml +++ b/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests-file-usage-context.xml @@ -18,4 +18,6 @@ - \ No newline at end of file + + + diff --git a/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests.java b/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests.java index 79a5a37f0e..05b8e9b274 100644 --- a/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests.java +++ b/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedMessageSourceBeanDefinitionParserTests.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.feed.config; import static junit.framework.Assert.assertEquals; @@ -31,6 +32,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; @@ -51,32 +53,35 @@ import com.sun.syndication.fetcher.impl.HttpURLFeedFetcher; /** * @author Oleg Zhurakousky - * + * @author Mark Fisher + * @since 2.0 */ public class FeedMessageSourceBeanDefinitionParserTests { + private static CountDownLatch latch; + @Before - public void prepare(){ + public void prepare() { File persisterFile = new File(System.getProperty("java.io.tmpdir") + "/spring-integration/", "feedAdapter.last.entry"); - if (persisterFile.exists()){ + if (persisterFile.exists()) { persisterFile.delete(); } } @Test - public void validateSuccessfullConfigurationWithCustomMetastore(){ - ClassPathXmlApplicationContext context = - new ClassPathXmlApplicationContext("FeedMessageSourceBeanDefinitionParserTests-file-context.xml", this.getClass()); + public void validateSuccessfulConfigurationWithCustomMetadataStore() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "FeedMessageSourceBeanDefinitionParserTests-file-context.xml", this.getClass()); SourcePollingChannelAdapter adapter = context.getBean("feedAdapter", SourcePollingChannelAdapter.class); FeedEntryReaderMessageSource source = (FeedEntryReaderMessageSource) TestUtils.getPropertyValue(adapter, "source"); - MetadataStore metaStore = (MetadataStore) TestUtils.getPropertyValue(source, "metadataStore"); - assertTrue(metaStore instanceof SampleMetadataStore); + MetadataStore metadataStore = (MetadataStore) TestUtils.getPropertyValue(source, "metadataStore"); + assertTrue(metadataStore instanceof SampleMetadataStore); + assertEquals(metadataStore, context.getBean("customMetadataStore")); FeedReaderMessageSource feedReaderMessageSource = (FeedReaderMessageSource) TestUtils.getPropertyValue(source, "feedReaderMessageSource"); AbstractFeedFetcher fetcher = (AbstractFeedFetcher) TestUtils.getPropertyValue(feedReaderMessageSource, "fetcher"); assertTrue(fetcher instanceof FileUrlFeedFetcher); - - context = - new ClassPathXmlApplicationContext("FeedMessageSourceBeanDefinitionParserTests-http-context.xml", this.getClass()); + context = new ClassPathXmlApplicationContext( + "FeedMessageSourceBeanDefinitionParserTests-http-context.xml", this.getClass()); adapter = context.getBean("feedAdapter", SourcePollingChannelAdapter.class); source = (FeedEntryReaderMessageSource) TestUtils.getPropertyValue(adapter, "source"); feedReaderMessageSource = (FeedReaderMessageSource) TestUtils.getPropertyValue(source, "feedReaderMessageSource"); @@ -84,69 +89,72 @@ public class FeedMessageSourceBeanDefinitionParserTests { assertTrue(fetcher instanceof HttpURLFeedFetcher); context.destroy(); } - + @Test - public void validateSuccessfullNewsRetrievalWithFileUrlAndMessageHistory() throws Exception{ - File persisterFile = new File(System.getProperty("java.io.tmpdir") + "/spring-integration/", "feedAdapterUsage.last.entry"); - if (persisterFile.exists()){ + public void validateSuccessfulNewsRetrievalWithFileUrlAndMessageHistory() throws Exception { + File persisterFile = new File(System.getProperty("java.io.tmpdir") + "/spring-integration/", "message-store.properties"); + if (persisterFile.exists()) { persisterFile.delete(); } //Test file samples.rss has 3 news items latch = spy(new CountDownLatch(3)); - ClassPathXmlApplicationContext context = - new ClassPathXmlApplicationContext("FeedMessageSourceBeanDefinitionParserTests-file-usage-context.xml", this.getClass()); + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "FeedMessageSourceBeanDefinitionParserTests-file-usage-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("FeedMessageSourceBeanDefinitionParserTests-file-usage-context.xml", this.getClass()); + context = new ClassPathXmlApplicationContext( + "FeedMessageSourceBeanDefinitionParserTests-file-usage-context.xml", this.getClass()); latch.await(5, TimeUnit.SECONDS); verify(latch, times(0)).countDown(); context.destroy(); } + @Test - public void validateSuccessfullNewsRetrievalWithFileUrlNoPersistentIdentifier() throws Exception{ + public void validateSuccessfulNewsRetrievalWithFileUrlNoPersistentIdentifier() throws Exception{ //Test file samples.rss has 3 news items latch = spy(new CountDownLatch(3)); - ClassPathXmlApplicationContext context = - new ClassPathXmlApplicationContext("FeedMessageSourceBeanDefinitionParserTests-file-usage-noid-context.xml", this.getClass()); + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "FeedMessageSourceBeanDefinitionParserTests-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("FeedMessageSourceBeanDefinitionParserTests-file-usage-noid-context.xml", this.getClass()); + context = new ClassPathXmlApplicationContext( + "FeedMessageSourceBeanDefinitionParserTests-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 validateSuccessfullNewsRetrievalWithHttpUrl() throws Exception{ + public void validateSuccessfulNewsRetrievalWithHttpUrl() throws Exception{ final CountDownLatch latch = new CountDownLatch(3); MessageHandler handler = spy(new MessageHandler() { public void handleMessage(Message message) throws MessagingException { latch.countDown(); } }); - ApplicationContext context = - new ClassPathXmlApplicationContext("FeedMessageSourceBeanDefinitionParserTests-http-context.xml", this.getClass()); + ApplicationContext context = new ClassPathXmlApplicationContext( + "FeedMessageSourceBeanDefinitionParserTests-http-context.xml", this.getClass()); DirectChannel feedChannel = context.getBean("feedChannel", DirectChannel.class); feedChannel.subscribe(handler); latch.await(5, TimeUnit.SECONDS); verify(handler, atLeast(3)).handleMessage(Mockito.any(Message.class)); } - - public static class SampleService{ - public void receiveFeedEntry(Message message){ + + + public static class SampleService { + + public void receiveFeedEntry(Message message) { MessageHistory history = MessageHistory.read(message); assertTrue(history.size() == 3); Properties historyItem = history.get(0); @@ -163,20 +171,24 @@ public class FeedMessageSourceBeanDefinitionParserTests { latch.countDown(); } } - - public static class SampleServiceNoHistory{ - public void receiveFeedEntry(SyndEntry entry){ + + + public static class SampleServiceNoHistory { + + public void receiveFeedEntry(SyndEntry entry) { latch.countDown(); } } - - public static class SampleMetadataStore implements MetadataStore{ - public void write(Properties metadata) { + + public static class SampleMetadataStore implements MetadataStore { + + public void put(String key, String value) { } - public Properties load() { - return new Properties(); + public String get(String key) { + return null; } } + } diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractInboundTwitterEndpointSupport.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractInboundTwitterEndpointSupport.java index b56254be21..df453334f7 100644 --- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractInboundTwitterEndpointSupport.java +++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractInboundTwitterEndpointSupport.java @@ -18,18 +18,18 @@ package org.springframework.integration.twitter.inbound; import java.util.ArrayList; import java.util.List; -import java.util.Properties; import java.util.concurrent.ScheduledFuture; +import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.Message; -import org.springframework.integration.context.metadata.FileBasedPropertiesStore; +import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.context.metadata.MetadataStore; +import org.springframework.integration.context.metadata.SimpleMetadataStore; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.history.HistoryWritingMessagePostProcessor; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.twitter.oauth.OAuthConfiguration; import org.springframework.util.Assert; -import org.springframework.util.StringUtils; import twitter4j.DirectMessage; import twitter4j.Status; @@ -45,6 +45,7 @@ import twitter4j.Twitter; * * @author Josh Long * @author Oleg Zhurakousky + * @author Mark Fisher * @since 2.0 */ public abstract class AbstractInboundTwitterEndpointSupport extends MessageProducerSupport { @@ -53,8 +54,6 @@ public abstract class AbstractInboundTwitterEndpointSupport extends MessagePr private volatile String metadataKey; - private volatile Properties lastPersistentEntry = new Properties(); - protected volatile OAuthConfiguration configuration; protected volatile long markerId = -1; @@ -65,15 +64,9 @@ public abstract class AbstractInboundTwitterEndpointSupport extends MessagePr private volatile ScheduledFuture twitterUpdatePollingTask; - private volatile String persistentIdentifier; - private final HistoryWritingMessagePostProcessor historyWritingPostProcessor = new HistoryWritingMessagePostProcessor(); - public void setPersistentIdentifier(String persistentIdentifier) { - this.persistentIdentifier = persistentIdentifier; - } - public void setConfiguration(OAuthConfiguration configuration) { this.configuration = configuration; } @@ -96,21 +89,22 @@ public abstract class AbstractInboundTwitterEndpointSupport extends MessagePr Assert.notNull(this.configuration, "'configuration' can't be null"); this.twitter = this.configuration.getTwitter(); Assert.notNull(this.twitter, "'twitter' instance can't be null"); - metadataKey = this.getComponentType() + "@" + this.getComponentName() - + "#" + this.configuration.getConsumerKey(); - try { - if (StringUtils.hasText(this.persistentIdentifier)) { - if (this.metadataStore == null) { - logger.info("Creating FileBasedPropertiesStore"); - metadataStore = new FileBasedPropertiesStore(this.persistentIdentifier); - ((FileBasedPropertiesStore) metadataStore).afterPropertiesSet(); + if (this.metadataStore == null) { + // first try to look for a 'messageStore' in the context + BeanFactory beanFactory = this.getBeanFactory(); + if (beanFactory != null) { + MetadataStore metadataStore = IntegrationContextUtils.getMetadataStore(beanFactory); + if (metadataStore != null) { + this.metadataStore = metadataStore; } - lastPersistentEntry = metadataStore.load(); + } + if (this.metadataStore == null) { + this.metadataStore = new SimpleMetadataStore(); } } - catch (Exception e) { - logger.warn("Failed to initialize and load from MetadataStore. Potential duplicates possible.", e); - } + Assert.hasText(this.getComponentName(), "Inbound Twitter adapter must have a name"); + this.metadataKey = this.getComponentType() + "." + this.getComponentName() + + "." + this.configuration.getConsumerKey(); } protected void forwardAll(List tResponses) { @@ -154,7 +148,7 @@ public abstract class AbstractInboundTwitterEndpointSupport extends MessagePr else { throw new IllegalArgumentException("Unsupported type of Twitter message: " + message.getClass()); } - String lastId = lastPersistentEntry.getProperty(this.metadataKey); + String lastId = this.metadataStore.get(this.metadataKey); long lastTweetId = 0; if (lastId != null) { @@ -163,15 +157,12 @@ public abstract class AbstractInboundTwitterEndpointSupport extends MessagePr if (id > lastTweetId) { sendMessage(twtMsg); markLastStatusId(id); - if (metadataStore != null) { - metadataStore.write(this.lastPersistentEntry); - } } } } protected void markLastStatusId(long statusId) { - lastPersistentEntry.put(metadataKey, String.valueOf(statusId)); + this.metadataStore.put(this.metadataKey, String.valueOf(statusId)); } } diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/InboundDirectMessageStatusEndpointTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/InboundDirectMessageStatusEndpointTests.java index 9c43b5859f..e94bdc5c8c 100644 --- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/InboundDirectMessageStatusEndpointTests.java +++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/InboundDirectMessageStatusEndpointTests.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.twitter.inbound; import static junit.framework.Assert.assertEquals; @@ -28,6 +29,7 @@ import java.util.List; import org.junit.Before; import org.junit.Test; import org.mockito.Mockito; + import org.springframework.integration.Message; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.twitter.oauth.OAuthConfiguration; @@ -41,36 +43,18 @@ import twitter4j.Twitter; /** * @author Oleg Zhurakousky - * */ public class InboundDirectMessageStatusEndpointTests { + private DirectMessage firstMessage; + private DirectMessage secondMessage; + private Twitter twitter; - @Test - public void testTwitterMockedUpdates() throws Exception{ - QueueChannel channel = new QueueChannel(); - - InboundDirectMessageEndpoint endpoint = new InboundDirectMessageEndpoint(); - endpoint.setOutputChannel(channel); - ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); - scheduler.afterPropertiesSet(); - endpoint.setTaskScheduler(scheduler); - - endpoint.setConfiguration(this.getTestConfigurationForDirectMessages()); - endpoint.afterPropertiesSet(); - endpoint.start(); - Message message1 = channel.receive(3000); - assertNotNull(message1); - // should be second message since its timestamp is newer - assertEquals(secondMessage.getId(), ((DirectMessage)message1.getPayload()).getId()); - Message message2 = channel.receive(100); - assertNull(message2); // should be null, since - } - + @Before - public void prepare(){ + public void prepare() { twitter = mock(Twitter.class); firstMessage = mock(DirectMessage.class); when(firstMessage.getCreatedAt()).thenReturn(new Date(5555555555L)); @@ -79,27 +63,47 @@ public class InboundDirectMessageStatusEndpointTests { when(secondMessage.getCreatedAt()).thenReturn(new Date(2222222222L)); when(secondMessage.getId()).thenReturn(2000); } - + + + @Test + public void testTwitterMockedUpdates() throws Exception{ + QueueChannel channel = new QueueChannel(); + InboundDirectMessageEndpoint endpoint = new InboundDirectMessageEndpoint(); + endpoint.setOutputChannel(channel); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.afterPropertiesSet(); + endpoint.setTaskScheduler(scheduler); + endpoint.setConfiguration(this.getTestConfigurationForDirectMessages()); + endpoint.setBeanName("twitterEndpoint"); + endpoint.afterPropertiesSet(); + endpoint.start(); + Message message1 = channel.receive(3000); + assertNotNull(message1); + // should be second message since its timestamp is newer + assertEquals(secondMessage.getId(), ((DirectMessage)message1.getPayload()).getId()); + Message message2 = channel.receive(100); + assertNull(message2); // should be null, since + } + + @SuppressWarnings("unchecked") private OAuthConfiguration getTestConfigurationForDirectMessages() throws Exception{ OAuthConfiguration configuration = mock(OAuthConfiguration.class); RateLimitStatus rateLimitStatus = mock(RateLimitStatus.class); when(twitter.getRateLimitStatus()).thenReturn(rateLimitStatus); when(configuration.getTwitter()).thenReturn(twitter); - when(rateLimitStatus.getSecondsUntilReset()).thenReturn(2464); when(rateLimitStatus.getRemainingHits()).thenReturn(250); - + ResponseList responses = mock(ResponseList.class); List testMessages = new ArrayList(); - testMessages.add(firstMessage); testMessages.add(secondMessage); - + when(responses.iterator()).thenReturn(testMessages.iterator()); when(twitter.getDirectMessages()).thenReturn(responses); - when(twitter.getDirectMessages(Mockito.any(Paging.class))).thenReturn(responses); return configuration; } + }